Coverage for scripts / live_release_validation / checks / topology.py: 100.00%

1"""Add-on convergence, health stability, and topology evidence validators.""" 

2 

3from __future__ import annotations 

4 

5import base64 

6import hashlib 

7import json 

8import re 

9import time 

10import zlib 

11from datetime import datetime 

12from typing import Any 

13 

14from ..constants import ( 

15 _HEALTHY_STACK_STATUSES, 

16) 

17from ..models import RunContext, to_jsonable, utc_now 

18 

19 

20def _queue_counts(status: dict[str, Any]) -> dict[str, int]: 

21 return { 

22 "available": int(status.get("messages_available", 0)), 

23 "in_flight": int(status.get("messages_in_flight", 0)), 

24 "delayed": int(status.get("messages_delayed", 0)), 

25 "dlq": int(status.get("dlq_messages", 0)), 

26 } 

27 

28 

29_ADDON_EXECUTION_FIELDS = frozenset( 

30 { 

31 "execution_arn", 

32 "state_machine_arn", 

33 "deployment_token", 

34 "cluster_name", 

35 "region", 

36 "input_sha256", 

37 "started_at", 

38 } 

39) 

40 

41 

42_ADDON_REQUIRED_INPUT_FIELDS = frozenset( 

43 { 

44 "ClusterName", 

45 "Region", 

46 "RegistryRegion", 

47 "ProjectName", 

48 "EnabledCharts", 

49 "Charts", 

50 "KedaOperatorRoleArn", 

51 "ImageReplacements", 

52 "DeploymentToken", 

53 } 

54) 

55 

56 

57_ADDON_OPTIONAL_INPUT_FIELDS = frozenset({"EndpointGroupArn"}) 

58 

59 

60_ADDON_TERMINAL_STATUSES = frozenset({"SUCCEEDED", "FAILED", "TIMED_OUT", "ABORTED"}) 

61 

62 

63_ADDON_FAILURE_STATUSES = _ADDON_TERMINAL_STATUSES - {"SUCCEEDED"} 

64 

65 

66_ADDON_CONVERGENCE_TIMEOUT_SECONDS = 2 * 60 * 60 

67 

68 

69_HEALTH_STABILITY_ROUNDS = 3 

70_HEALTH_WARMUP_ATTEMPTS = 3 

71_RETRYABLE_HEALTH_WARMUP_STATUS_CODES = frozenset({429, 502, 503, 504}) 

72_RETRYABLE_HEALTH_WARMUP_ERROR = re.compile( 

73 r"(?:API request failed: (?:429|502|503|504)\b|gateway timeout|service unavailable)", 

74 re.IGNORECASE, 

75) 

76_RETRYABLE_HEALTH_WARMUP_EXCEPTIONS = frozenset( 

77 {"ConnectTimeout", "ConnectionError", "ReadTimeout", "Timeout"} 

78) 

79 

80 

81_MAX_TOPOLOGY_EVIDENCE_CHARS = 2048 

82 

83 

84def _bounded_topology_evidence( 

85 value: Any, 

86 limit: int = _MAX_TOPOLOGY_EVIDENCE_CHARS, 

87) -> str: 

88 """Serialize diagnostic evidence without allowing an unbounded checkpoint.""" 

89 if value is None: 

90 text = "<absent>" 

91 elif isinstance(value, str): 

92 text = value 

93 else: 

94 try: 

95 text = json.dumps(to_jsonable(value), sort_keys=True) 

96 except TypeError, ValueError: 

97 text = str(value) 

98 suffix = "... [truncated]" 

99 if len(text) <= limit: 

100 return text 

101 return text[: limit - len(suffix)] + suffix 

102 

103 

104def _topology_json_object( 

105 value: Any, 

106 description: str, 

107 *, 

108 canonical: bool, 

109) -> dict[str, Any]: 

110 """Decode a JSON object, optionally requiring the provider's exact encoding.""" 

111 if not isinstance(value, str) or not value: 

112 raise RuntimeError(f"{description} is not a non-empty JSON string") 

113 

114 def reject_constant(constant: str) -> None: 

115 raise ValueError(f"non-standard JSON constant {constant}") 

116 

117 try: 

118 parsed = json.loads(value, parse_constant=reject_constant) 

119 except (json.JSONDecodeError, ValueError) as exc: 

120 raise RuntimeError(f"{description} is invalid JSON: {exc}") from exc 

121 if not isinstance(parsed, dict): 

122 raise RuntimeError(f"{description} must be a JSON object") 

123 if canonical and json.dumps(parsed, sort_keys=True, separators=(",", ":")) != value: 

124 raise RuntimeError(f"{description} is not exact canonical JSON") 

125 return parsed 

126 

127 

128def _decode_replay_input_parameter(stored_value: str, description: str) -> str: 

129 """Reverse the helm orchestrator's zlib+base64 replay-input encoding. 

130 

131 The orchestrator stores the convergence execution input encoded because 

132 SSM rejects raw ``{{PLACEHOLDER}}`` tokens. ``input_sha256`` in the 

133 companion ``_execution`` parameter is always computed over the decoded 

134 canonical JSON returned here. 

135 """ 

136 try: 

137 compressed = base64.b64decode(stored_value.encode("ascii"), validate=True) 

138 return zlib.decompress(compressed).decode("utf-8") 

139 except (ValueError, zlib.error, UnicodeDecodeError) as exc: 

140 raise RuntimeError(f"{description} is not zlib+base64 replay input: {exc}") from exc 

141 

142 

143def _ssm_string_parameter(client: Any, name: str) -> str: 

144 """Read one exact String parameter and reject a malformed SDK response.""" 

145 response = client.get_parameter(Name=name) 

146 parameter = response.get("Parameter") if isinstance(response, dict) else None 

147 if not isinstance(parameter, dict): 

148 raise RuntimeError(f"SSM parameter response is malformed for {name}") 

149 if parameter.get("Name") != name or parameter.get("Type") != "String": 

150 raise RuntimeError(f"SSM parameter identity/type is invalid for {name}") 

151 value = parameter.get("Value") 

152 if not isinstance(value, str) or not value: 

153 raise RuntimeError(f"SSM parameter has no String value: {name}") 

154 return value 

155 

156 

157def _epoch_seconds(value: Any, description: str) -> int: 

158 """Normalize an SDK timestamp while rejecting booleans and invalid values.""" 

159 if isinstance(value, datetime): 

160 raw_value: Any = value.timestamp() 

161 else: 

162 raw_value = value 

163 if isinstance(raw_value, bool) or not isinstance(raw_value, int | float): 

164 raise RuntimeError(f"{description} is not a timestamp") 

165 try: 

166 result = int(raw_value) 

167 except (OverflowError, ValueError) as exc: 

168 raise RuntimeError(f"{description} is not a finite timestamp") from exc 

169 if result <= 0: 

170 raise RuntimeError(f"{description} must be positive") 

171 return result 

172 

173 

174def _validate_addon_arns( 

175 ctx: RunContext, 

176 *, 

177 region: str, 

178 stack_name: str, 

179 stack_id: str, 

180 state_machine_arn: str, 

181 execution_arn: str, 

182) -> None: 

183 """Require exact account, partition, Region, and parent state-machine ARNs.""" 

184 partition = ctx.session.get_partition_for_region(region) 

185 if not partition: 

186 raise RuntimeError(f"Could not resolve AWS partition for {region}") 

187 escaped_partition = re.escape(str(partition)) 

188 escaped_region = re.escape(region) 

189 escaped_account = re.escape(ctx.settings.expected_account) 

190 escaped_stack_name = re.escape(stack_name) 

191 

192 stack_pattern = ( 

193 rf"arn:{escaped_partition}:cloudformation:{escaped_region}:{escaped_account}:" 

194 rf"stack/{escaped_stack_name}/[^/:\s]+" 

195 ) 

196 if re.fullmatch(stack_pattern, stack_id) is None: 

197 raise RuntimeError(f"Regional stack ID has the wrong ARN identity: {stack_id}") 

198 

199 state_machine_pattern = ( 

200 rf"arn:{escaped_partition}:states:{escaped_region}:{escaped_account}:" 

201 r"stateMachine:([^:\s]+)" 

202 ) 

203 state_machine_match = re.fullmatch(state_machine_pattern, state_machine_arn) 

204 if state_machine_match is None: 

205 raise RuntimeError( 

206 f"Add-on state-machine ARN has the wrong account/partition/Region: {state_machine_arn}" 

207 ) 

208 execution_pattern = ( 

209 rf"arn:{escaped_partition}:states:{escaped_region}:{escaped_account}:" 

210 rf"execution:{re.escape(state_machine_match.group(1))}:[^:\s]+" 

211 ) 

212 if re.fullmatch(execution_pattern, execution_arn) is None: 

213 raise RuntimeError( 

214 f"Add-on execution ARN is not an execution of {state_machine_arn}: {execution_arn}" 

215 ) 

216 

217 

218def _state_machine_stack_resource( 

219 ctx: RunContext, 

220 *, 

221 region: str, 

222 stack_id: str, 

223 state_machine_arn: str, 

224) -> dict[str, str]: 

225 """Prove the physical state machine belongs to the exact regional stack ARN.""" 

226 cloudformation = ctx.session.client("cloudformation", region_name=region) 

227 pages = cloudformation.get_paginator("list_stack_resources").paginate(StackName=stack_id) 

228 matches = [ 

229 resource 

230 for page in pages 

231 for resource in page.get("StackResourceSummaries", []) 

232 if resource.get("ResourceType") == "AWS::StepFunctions::StateMachine" 

233 and resource.get("PhysicalResourceId") == state_machine_arn 

234 ] 

235 if len(matches) != 1: 

236 raise RuntimeError( 

237 f"State machine {state_machine_arn} is not exactly one " 

238 f"AWS::StepFunctions::StateMachine resource in stack {stack_id}" 

239 ) 

240 resource = matches[0] 

241 logical_id = resource.get("LogicalResourceId") 

242 resource_status = resource.get("ResourceStatus") 

243 if not isinstance(logical_id, str) or not logical_id: 

244 raise RuntimeError(f"State-machine stack resource lacks a logical ID in {stack_id}") 

245 if resource_status not in _HEALTHY_STACK_STATUSES: 

246 raise RuntimeError( 

247 f"State-machine stack resource {logical_id} is not complete: {resource_status}" 

248 ) 

249 return { 

250 "logical_id": logical_id, 

251 "physical_id": state_machine_arn, 

252 "resource_type": "AWS::StepFunctions::StateMachine", 

253 "status": str(resource_status), 

254 } 

255 

256 

257def _validate_terminal_validator( 

258 output: dict[str, Any], 

259 *, 

260 key: str, 

261 deployment_token: str, 

262 count_pairs: tuple[tuple[str, str], ...], 

263) -> dict[str, Any]: 

264 """Validate one terminal convergence payload and each expected/actual count pair.""" 

265 validator = output.get(key) 

266 if not isinstance(validator, dict): 

267 raise RuntimeError(f"Step Functions output lacks object {key}") 

268 if validator.get("status") != "validated": 

269 raise RuntimeError(f"Step Functions output {key}.status is not exactly 'validated'") 

270 if validator.get("DeploymentToken") != deployment_token: 

271 raise RuntimeError(f"Step Functions output {key} has a stale deployment token") 

272 for expected_key, validated_key in count_pairs: 

273 expected = validator.get(expected_key) 

274 validated = validator.get(validated_key) 

275 if ( 

276 isinstance(expected, bool) 

277 or isinstance(validated, bool) 

278 or not isinstance(expected, int) 

279 or not isinstance(validated, int) 

280 or expected < 0 

281 or validated < 0 

282 ): 

283 raise RuntimeError( 

284 f"Step Functions output {key} has invalid counts {expected_key}/{validated_key}" 

285 ) 

286 if expected != validated: 

287 raise RuntimeError( 

288 f"Step Functions output {key} did not validate every item: " 

289 f"{expected_key}={expected}, {validated_key}={validated}" 

290 ) 

291 return validator 

292 

293 

294def _validate_addon_execution_input( 

295 input_value: dict[str, Any], 

296 *, 

297 cluster_name: str, 

298 region: str, 

299 registry_region: str, 

300 project_name: str, 

301 deployment_token: str, 

302) -> None: 

303 """Require the exact current orchestrator input schema and regional identity.""" 

304 fields = set(input_value) 

305 if not _ADDON_REQUIRED_INPUT_FIELDS.issubset(fields) or not fields.issubset( 

306 _ADDON_REQUIRED_INPUT_FIELDS | _ADDON_OPTIONAL_INPUT_FIELDS 

307 ): 

308 raise RuntimeError("Add-on execution input does not use the exact current schema") 

309 if input_value.get("ClusterName") != cluster_name: 

310 raise RuntimeError("Add-on execution input has a stale cluster name") 

311 if input_value.get("Region") != region: 

312 raise RuntimeError("Add-on execution input has a stale Region") 

313 if input_value.get("RegistryRegion") != registry_region: 

314 raise RuntimeError("Add-on execution input has a stale registry Region") 

315 if input_value.get("ProjectName") != project_name: 

316 raise RuntimeError("Add-on execution input has a stale project name") 

317 if input_value.get("DeploymentToken") != deployment_token: 

318 raise RuntimeError("Add-on execution input has a stale deployment token") 

319 enabled_charts = input_value.get("EnabledCharts") 

320 if not isinstance(enabled_charts, list) or not all( 

321 isinstance(item, str) and item for item in enabled_charts 

322 ): 

323 raise RuntimeError("Add-on execution input EnabledCharts must be a string list") 

324 if not isinstance(input_value.get("Charts"), dict): 

325 raise RuntimeError("Add-on execution input Charts must be an object") 

326 if not isinstance(input_value.get("ImageReplacements"), dict): 

327 raise RuntimeError("Add-on execution input ImageReplacements must be an object") 

328 keda_role_arn = input_value.get("KedaOperatorRoleArn") 

329 if not isinstance(keda_role_arn, str | type(None)): 

330 raise RuntimeError("Add-on execution input KedaOperatorRoleArn must be a string or null") 

331 if "EndpointGroupArn" in input_value: 

332 endpoint_group_arn = input_value["EndpointGroupArn"] 

333 if not isinstance(endpoint_group_arn, str) or not endpoint_group_arn: 

334 raise RuntimeError("Add-on execution input EndpointGroupArn must be non-empty") 

335 

336 

337def _poll_addon_execution( 

338 ctx: RunContext, 

339 *, 

340 region: str, 

341 execution: dict[str, Any], 

342 input_json: str, 

343 evidence: dict[str, Any], 

344) -> dict[str, Any]: 

345 """Poll one exact execution to a bounded terminal result and validate its output.""" 

346 execution_arn = str(execution["execution_arn"]) 

347 state_machine_arn = str(execution["state_machine_arn"]) 

348 deployment_token = str(execution["deployment_token"]) 

349 poll_interval = max(0.0, float(ctx.settings.poll_interval_seconds)) 

350 deadline = time.monotonic() + _ADDON_CONVERGENCE_TIMEOUT_SECONDS + poll_interval 

351 stepfunctions = ctx.session.client("stepfunctions", region_name=region) 

352 

353 while True: 

354 response = stepfunctions.describe_execution(executionArn=execution_arn) 

355 if not isinstance(response, dict): 

356 raise RuntimeError(f"DescribeExecution returned a malformed response in {region}") 

357 if response.get("executionArn") != execution_arn: 

358 raise RuntimeError(f"DescribeExecution returned a different execution in {region}") 

359 if response.get("stateMachineArn") != state_machine_arn: 

360 raise RuntimeError(f"DescribeExecution returned a different state machine in {region}") 

361 if response.get("input") != input_json: 

362 raise RuntimeError(f"DescribeExecution returned stale execution input in {region}") 

363 if ( 

364 _epoch_seconds(response.get("startDate"), "DescribeExecution startDate") 

365 != execution["started_at"] 

366 ): 

367 raise RuntimeError(f"DescribeExecution start time changed in {region}") 

368 

369 status = response.get("status") 

370 if not isinstance(status, str): 

371 raise RuntimeError(f"DescribeExecution returned no status in {region}") 

372 observation = { 

373 "observed_at": utc_now(), 

374 "status": status, 

375 "execution_arn": execution_arn, 

376 } 

377 for field in ("error", "cause", "output"): 

378 if field in response: 

379 observation[field] = _bounded_topology_evidence(response.get(field)) 

380 evidence.setdefault("observations", []).append(observation) 

381 ctx.persist() 

382 

383 if status == "RUNNING": 

384 remaining = deadline - time.monotonic() 

385 if remaining <= 0: 

386 raise RuntimeError( 

387 f"Add-on execution {execution_arn} did not finish within " 

388 f"{_ADDON_CONVERGENCE_TIMEOUT_SECONDS + poll_interval:.1f} seconds" 

389 ) 

390 time.sleep(min(poll_interval if poll_interval > 0 else 0.1, remaining)) 

391 continue 

392 if status not in _ADDON_TERMINAL_STATUSES: 

393 raise RuntimeError(f"Add-on execution {execution_arn} has unknown status {status}") 

394 

395 evidence["execution_status"] = status 

396 if status in _ADDON_FAILURE_STATUSES: 

397 terminal = { 

398 field: _bounded_topology_evidence(response.get(field)) 

399 for field in ("error", "cause", "output") 

400 } 

401 evidence["terminal"] = {"status": status, **terminal} 

402 ctx.persist() 

403 raise RuntimeError( 

404 f"Add-on execution {execution_arn} ended {status}; " 

405 f"error={terminal['error']}; cause={terminal['cause']}; " 

406 f"output={terminal['output']}" 

407 ) 

408 

409 output = _topology_json_object( 

410 response.get("output"), 

411 f"Step Functions output for {region}", 

412 canonical=False, 

413 ) 

414 manifest_validation = _validate_terminal_validator( 

415 output, 

416 key="manifestValidation", 

417 deployment_token=deployment_token, 

418 count_pairs=(("ExpectedCount", "ValidatedCount"),), 

419 ) 

420 helm_validation = _validate_terminal_validator( 

421 output, 

422 key="helmValidation", 

423 deployment_token=deployment_token, 

424 count_pairs=( 

425 ("expected_release_count", "validated_release_count"), 

426 ("expected_resource_count", "validated_resource_count"), 

427 ), 

428 ) 

429 terminal_evidence: dict[str, Any] = { 

430 "status": status, 

431 "manifestValidation": to_jsonable(manifest_validation), 

432 "helmValidation": to_jsonable(helm_validation), 

433 } 

434 evidence["terminal"] = terminal_evidence 

435 ctx.persist() 

436 return terminal_evidence 

437 

438 

439def _converge_region_addons( 

440 ctx: RunContext, 

441 *, 

442 region: str, 

443 stack_name: str, 

444 stack: dict[str, Any], 

445 evidence: dict[str, Any], 

446) -> None: 

447 """Validate persisted identity and wait for exact current add-on convergence.""" 

448 stack_id = str(stack.get("stack_id") or "") 

449 outputs = stack.get("outputs") 

450 if not isinstance(outputs, dict): 

451 raise RuntimeError(f"Regional stack {stack_name} has malformed outputs") 

452 cluster_name = f"{ctx.config.project_name}-{region}" 

453 if outputs.get("ClusterName") != cluster_name: 

454 raise RuntimeError(f"Regional stack {stack_name} has a stale ClusterName output") 

455 deployment_token = outputs.get("AddonDeploymentToken") 

456 if not isinstance(deployment_token, str) or not deployment_token: 

457 raise RuntimeError(f"Regional stack {stack_name} has no AddonDeploymentToken output") 

458 

459 parameter_root = f"/{ctx.config.project_name}/addons/{region}" 

460 execution_parameter = f"{parameter_root}/_execution" 

461 input_parameter = f"{parameter_root}/_input" 

462 ssm = ctx.session.client("ssm", region_name=region) 

463 execution_json = _ssm_string_parameter(ssm, execution_parameter) 

464 input_json = _decode_replay_input_parameter( 

465 _ssm_string_parameter(ssm, input_parameter), 

466 f"SSM parameter {input_parameter}", 

467 ) 

468 execution = _topology_json_object( 

469 execution_json, 

470 f"SSM parameter {execution_parameter}", 

471 canonical=True, 

472 ) 

473 input_value = _topology_json_object( 

474 input_json, 

475 f"SSM parameter {input_parameter}", 

476 canonical=True, 

477 ) 

478 

479 if set(execution) != _ADDON_EXECUTION_FIELDS: 

480 raise RuntimeError(f"SSM parameter {execution_parameter} has an unexpected schema") 

481 for field in ( 

482 "execution_arn", 

483 "state_machine_arn", 

484 "deployment_token", 

485 "cluster_name", 

486 "region", 

487 "input_sha256", 

488 ): 

489 if not isinstance(execution.get(field), str) or not execution[field]: 

490 raise RuntimeError(f"SSM parameter {execution_parameter} has invalid {field}") 

491 started_at = execution.get("started_at") 

492 if isinstance(started_at, bool) or not isinstance(started_at, int) or started_at <= 0: 

493 raise RuntimeError(f"SSM parameter {execution_parameter} has invalid started_at") 

494 if execution["deployment_token"] != deployment_token: 

495 raise RuntimeError(f"SSM parameter {execution_parameter} has a stale deployment token") 

496 if execution["cluster_name"] != cluster_name or execution["region"] != region: 

497 raise RuntimeError(f"SSM parameter {execution_parameter} has stale regional identity") 

498 input_sha256 = hashlib.sha256(input_json.encode("utf-8")).hexdigest() 

499 if execution["input_sha256"] != input_sha256: 

500 raise RuntimeError(f"SSM parameter {execution_parameter} has a stale input SHA-256") 

501 _validate_addon_execution_input( 

502 input_value, 

503 cluster_name=cluster_name, 

504 region=region, 

505 registry_region=ctx.config.global_region, 

506 project_name=ctx.config.project_name, 

507 deployment_token=deployment_token, 

508 ) 

509 _validate_addon_arns( 

510 ctx, 

511 region=region, 

512 stack_name=stack_name, 

513 stack_id=stack_id, 

514 state_machine_arn=execution["state_machine_arn"], 

515 execution_arn=execution["execution_arn"], 

516 ) 

517 stack_resource = _state_machine_stack_resource( 

518 ctx, 

519 region=region, 

520 stack_id=stack_id, 

521 state_machine_arn=execution["state_machine_arn"], 

522 ) 

523 

524 evidence.update( 

525 { 

526 "stack_id": stack_id, 

527 "cluster_name": cluster_name, 

528 "deployment_token": deployment_token, 

529 "execution": to_jsonable(execution), 

530 "input": to_jsonable(input_value), 

531 "input_sha256": input_sha256, 

532 "state_machine_resource": stack_resource, 

533 } 

534 ) 

535 ctx.persist() 

536 _poll_addon_execution( 

537 ctx, 

538 region=region, 

539 execution=execution, 

540 input_json=input_json, 

541 evidence=evidence, 

542 ) 

543 

544 

545def _validate_health_payload( 

546 ctx: RunContext, 

547 payload: Any, 

548 *, 

549 endpoint_region: str | None, 

550) -> dict[str, Any]: 

551 """Require a healthy, well-formed response bound to one deployed cluster.""" 

552 if not isinstance(payload, dict): 

553 raise RuntimeError("health response is not a JSON object") 

554 if payload.get("status") != "healthy": 

555 raise RuntimeError("health response status is not exactly 'healthy'") 

556 timestamp = payload.get("timestamp") 

557 if not isinstance(timestamp, str) or "T" not in timestamp: 

558 raise RuntimeError("health response timestamp is not an ISO date-time") 

559 try: 

560 datetime.fromisoformat(timestamp[:-1] + "+00:00" if timestamp.endswith("Z") else timestamp) 

561 except ValueError as exc: 

562 raise RuntimeError("health response timestamp is not an ISO date-time") from exc 

563 payload_region = payload.get("region") 

564 if payload_region not in ctx.deployment_regions: 

565 raise RuntimeError(f"health response Region is not deployed: {payload_region!r}") 

566 if endpoint_region is not None and payload_region != endpoint_region: 

567 raise RuntimeError( 

568 f"regional health response came from {payload_region!r}, expected {endpoint_region!r}" 

569 ) 

570 expected_cluster_id = f"{ctx.config.project_name}-{payload_region}" 

571 if payload.get("cluster_id") != expected_cluster_id: 

572 raise RuntimeError( 

573 f"health response cluster_id is not {expected_cluster_id!r}: " 

574 f"{payload.get('cluster_id')!r}" 

575 ) 

576 return payload 

577 

578 

579def _health_warmup_samples( 

580 ctx: RunContext, 

581 *, 

582 global_url: str, 

583 regional_urls: dict[str, str], 

584) -> list[dict[str, Any]]: 

585 """Warm each endpoint with a checkpointed, resume-bounded attempt budget.""" 

586 probes: list[dict[str, Any]] = [ 

587 {"scope": "global", "region": None, "endpoint": global_url}, 

588 *( 

589 {"scope": "regional", "region": region, "endpoint": regional_urls[region]} 

590 for region in ctx.deployment_regions 

591 if region in regional_urls 

592 ), 

593 ] 

594 raw_samples = ctx.checkpoint.state.get("topology_health_warmup_samples") 

595 if raw_samples is None: 

596 samples: list[dict[str, Any]] = [] 

597 ctx.checkpoint.state["topology_health_warmup_samples"] = samples 

598 ctx.persist() 

599 elif isinstance(raw_samples, list) and all(isinstance(sample, dict) for sample in raw_samples): 

600 samples = raw_samples 

601 else: 

602 raise RuntimeError("Topology health warm-up checkpoint is malformed") 

603 

604 probes_by_key = {(probe["scope"], probe["region"]): probe for probe in probes} 

605 histories: dict[tuple[Any, Any], list[dict[str, Any]]] = {key: [] for key in probes_by_key} 

606 for sample in samples: 

607 required_fields = { 

608 "scope", 

609 "region", 

610 "endpoint", 

611 "attempt", 

612 "timestamp", 

613 "latency_seconds", 

614 "payload", 

615 "error", 

616 "status_code", 

617 "retryable", 

618 } 

619 if not required_fields.issubset(sample): 

620 raise RuntimeError("Topology health warm-up checkpoint outcome is incomplete") 

621 latency = sample["latency_seconds"] 

622 status_code = sample["status_code"] 

623 error = sample["error"] 

624 retryable = sample["retryable"] 

625 if ( 

626 not isinstance(sample["timestamp"], str) 

627 or not sample["timestamp"] 

628 or isinstance(latency, bool) 

629 or not isinstance(latency, (int, float)) 

630 or latency < 0 

631 or not isinstance(retryable, bool) 

632 or ( 

633 status_code is not None 

634 and (isinstance(status_code, bool) or not isinstance(status_code, int)) 

635 ) 

636 ): 

637 raise RuntimeError("Topology health warm-up checkpoint outcome is malformed") 

638 if error is None: 

639 if ( 

640 not isinstance(sample["payload"], dict) 

641 or status_code != 200 

642 or retryable is not False 

643 ): 

644 raise RuntimeError("Topology health warm-up checkpoint success is malformed") 

645 elif not isinstance(error, str) or not error or sample["payload"] is not None: 

646 raise RuntimeError("Topology health warm-up checkpoint failure is malformed") 

647 

648 key = (sample.get("scope"), sample.get("region")) 

649 probe = probes_by_key.get(key) 

650 if probe is None or sample.get("endpoint") != probe["endpoint"]: 

651 raise RuntimeError("Topology health warm-up checkpoint identity changed") 

652 histories[key].append(sample) 

653 for history in histories.values(): 

654 attempts = [sample.get("attempt") for sample in history] 

655 if attempts != list(range(1, len(history) + 1)) or len(history) > _HEALTH_WARMUP_ATTEMPTS: 

656 raise RuntimeError("Topology health warm-up checkpoint ordering is invalid") 

657 successes = [sample for sample in history if sample.get("error") is None] 

658 if len(successes) > 1 or (successes and history[-1] is not successes[0]): 

659 raise RuntimeError("Topology health warm-up checkpoint success is inconsistent") 

660 

661 interval = min(max(0.0, float(ctx.settings.poll_interval_seconds)), 5.0) 

662 for probe in probes: 

663 key = (probe["scope"], probe["region"]) 

664 history = histories[key] 

665 if history and history[-1].get("error") is None: 

666 continue 

667 if history and history[-1].get("retryable") is not True: 

668 raise RuntimeError( 

669 f"Health warm-up previously failed for {probe['endpoint']}: " 

670 f"{history[-1].get('error')}" 

671 ) 

672 if len(history) >= _HEALTH_WARMUP_ATTEMPTS: 

673 raise RuntimeError( 

674 f"Health warm-up attempt budget is exhausted for {probe['endpoint']}: " 

675 f"{history[-1].get('error')}" 

676 ) 

677 

678 # Every iteration ends in `break` (success), `continue` (retryable, with 

679 # attempts left) or a raise, so the loop never runs off the end of the 

680 # range: the budget check above guarantees at least one attempt, and the 

681 # last attempt raises instead of continuing. 

682 for attempt in range(len(history) + 1, _HEALTH_WARMUP_ATTEMPTS + 1): # pragma: no branch 

683 started = time.monotonic() 

684 try: 

685 payload = ctx.aws_client.call_api( 

686 method="GET", 

687 path="/api/v1/health", 

688 region=probe["region"], 

689 max_attempts=1, 

690 ) 

691 except Exception as exc: 

692 call_error = _bounded_topology_evidence(f"{type(exc).__name__}: {exc}") 

693 status_code = getattr(exc, "status_code", None) 

694 has_structured_status = isinstance(status_code, int) and not isinstance( 

695 status_code, bool 

696 ) 

697 retryable = ( 

698 status_code in _RETRYABLE_HEALTH_WARMUP_STATUS_CODES 

699 if has_structured_status 

700 else ( 

701 type(exc).__name__ in _RETRYABLE_HEALTH_WARMUP_EXCEPTIONS 

702 or _RETRYABLE_HEALTH_WARMUP_ERROR.search(call_error) is not None 

703 ) 

704 ) 

705 sample = { 

706 **probe, 

707 "attempt": attempt, 

708 "timestamp": utc_now(), 

709 "latency_seconds": round(max(0.0, time.monotonic() - started), 6), 

710 "payload": None, 

711 "error": call_error, 

712 "status_code": status_code, 

713 "retryable": retryable, 

714 } 

715 samples.append(sample) 

716 history.append(sample) 

717 ctx.persist() 

718 if retryable and attempt < _HEALTH_WARMUP_ATTEMPTS: 

719 if interval > 0: 

720 time.sleep(interval) 

721 continue 

722 raise RuntimeError( 

723 f"Health warm-up call failed for {probe['endpoint']} on attempt " 

724 f"{attempt}: {call_error}" 

725 ) from exc 

726 

727 validation_error: str | None = None 

728 try: 

729 _validate_health_payload(ctx, payload, endpoint_region=probe["region"]) 

730 except RuntimeError as exc: 

731 validation_error = _bounded_topology_evidence(str(exc)) 

732 sample = { 

733 **probe, 

734 "attempt": attempt, 

735 "timestamp": utc_now(), 

736 "latency_seconds": round(max(0.0, time.monotonic() - started), 6), 

737 "payload": to_jsonable(payload), 

738 "error": validation_error,