Coverage for scripts / live_release_validation / actions / destroy.py: 100.00%
160 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""destroy: remove all run-owned infrastructure in dependency order."""
3from __future__ import annotations
5import copy
6import hashlib
7import json
8import time
9from typing import Any
11from ..cleanup.retained import _retained_resource_cleanup
12from ..cleanup.workloads import cleanup_workloads
13from ..models import RunContext, to_jsonable, utc_now
14from ..ownership.cleanup_role import (
15 _delete_log_cleanup_helper,
16 _ensure_log_cleanup_helper,
17)
18from ..ownership.ecr import (
19 _checkpoint_new_ecr_images,
20 _checkpoint_new_ecr_repositories,
21 _record_ecr_repository_creation,
22)
23from ..ownership.kms import (
24 _checkpoint_retained_kms_keys,
25)
26from ..ownership.stacks import (
27 _authorize_owned_stack,
28 _owned_stack_record,
29 _prepared_change_set_authority,
30 _reconcile_stack_ownership,
31 _record_prepared_stack_identity,
32 _verify_target_stack_absence,
33)
36def _workload_cleanup_snapshot_sha256(ctx: RunContext) -> str:
37 payload = to_jsonable(
38 {
39 "jobs": copy.deepcopy(ctx.checkpoint.state.get("jobs", [])),
40 "central_jobs": copy.deepcopy(ctx.checkpoint.state.get("central_jobs", [])),
41 }
42 )
43 encoded = json.dumps(
44 payload,
45 sort_keys=True,
46 separators=(",", ":"),
47 ensure_ascii=True,
48 allow_nan=False,
49 ).encode("utf-8")
50 return hashlib.sha256(encoded).hexdigest()
53def _record_workload_cleanup_barrier(
54 ctx: RunContext,
55 cleanup_result: dict[str, Any],
56) -> dict[str, Any]:
57 if (
58 cleanup_result.get("complete") is not True
59 or cleanup_result.get("errors")
60 or cleanup_result.get("unresolved")
61 ):
62 raise RuntimeError("Cannot checkpoint an incomplete workload cleanup barrier")
63 barrier = {
64 "complete": True,
65 "completed_at": str(cleanup_result.get("ended_at") or utc_now()),
66 "snapshot_sha256": _workload_cleanup_snapshot_sha256(ctx),
67 "job_count": len(ctx.checkpoint.state.get("jobs", [])),
68 "central_job_count": len(ctx.checkpoint.state.get("central_jobs", [])),
69 }
70 ctx.checkpoint.state["workload_cleanup_barrier"] = barrier
71 ctx.persist()
72 return barrier
75def _validated_workload_cleanup_barrier(ctx: RunContext) -> dict[str, Any]:
76 barrier = ctx.checkpoint.state.get("workload_cleanup_barrier")
77 if not isinstance(barrier, dict) or barrier.get("complete") is not True:
78 raise RuntimeError("Checkpoint lacks a complete workload cleanup barrier")
79 expected = str(barrier.get("snapshot_sha256") or "")
80 current = _workload_cleanup_snapshot_sha256(ctx)
81 if not expected or expected != current:
82 raise RuntimeError("Checkpoint workload identity changed after cleanup completed")
83 return barrier
86def _resume_workload_cleanup_after_stack_absence(
87 ctx: RunContext,
88) -> tuple[dict[str, Any], dict[str, Any]]:
89 """Use an existing barrier, or create one only for a proven empty legacy run."""
90 jobs = ctx.checkpoint.state.get("jobs", [])
91 central_jobs = ctx.checkpoint.state.get("central_jobs", [])
92 if not isinstance(jobs, list) or not isinstance(central_jobs, list):
93 raise RuntimeError("Checkpoint workload collections must be lists")
95 if ctx.checkpoint.state.get("workload_cleanup_barrier") is None:
96 if jobs or central_jobs:
97 raise RuntimeError(
98 "Target stacks are absent but no completed workload cleanup barrier "
99 "was checkpointed"
100 )
101 workload_cleanup = cleanup_workloads(ctx)
102 barrier = _record_workload_cleanup_barrier(ctx, workload_cleanup)
103 else:
104 barrier = _validated_workload_cleanup_barrier(ctx)
105 workload_cleanup = {
106 "complete": True,
107 "reconciled_from_checkpoint_barrier": True,
108 "barrier": copy.deepcopy(barrier),
109 }
110 _validated_workload_cleanup_barrier(ctx)
111 return workload_cleanup, barrier
114def _record_target_stack_absence(
115 ctx: RunContext,
116 stack_absence: dict[str, Any],
117 *,
118 source: str,
119) -> dict[str, Any]:
120 if stack_absence.get("all_absent") is not True:
121 raise RuntimeError("Cannot checkpoint target stack absence while a stack remains")
122 workload_barrier = _validated_workload_cleanup_barrier(ctx)
123 proof = {
124 "verified_at": utc_now(),
125 "source": source,
126 "workload_cleanup_snapshot_sha256": workload_barrier["snapshot_sha256"],
127 "stack_absence": copy.deepcopy(stack_absence),
128 }
129 ctx.checkpoint.state["target_stacks_absent"] = proof
130 ctx.checkpoint.state.setdefault("target_stack_absence_proofs", []).append(proof)
131 ctx.persist()
132 return proof
135def destroy_deployment(ctx: RunContext) -> dict[str, Any]:
136 """Retry exact-owned teardown, preserving every structured attempt."""
137 if not ctx.checkpoint.deployment_attempted:
138 return {"needed": False, "attempts": []}
140 initial_absence = _verify_target_stack_absence(ctx)
141 if ctx.checkpoint.destroyed and initial_absence["all_absent"]:
142 workload_cleanup, workload_barrier = _resume_workload_cleanup_after_stack_absence(ctx)
143 absence_proof = _record_target_stack_absence(
144 ctx,
145 initial_absence,
146 source="destroy-already-destroyed-initial-absence",
147 )
148 _checkpoint_retained_kms_keys(ctx)
149 retained_cleanup = _retained_resource_cleanup(ctx)
150 final_absence = _verify_target_stack_absence(ctx)
151 if not final_absence["all_absent"]:
152 raise RuntimeError(
153 "A target stack reappeared during repeated retained cleanup: "
154 + json.dumps(final_absence["residual"], sort_keys=True)
155 )
156 completion_proof = _record_target_stack_absence(
157 ctx,
158 final_absence,
159 source="destroy-already-destroyed-completion",
160 )
161 return {
162 "needed": True,
163 "already_destroyed": True,
164 "workload_cleanup": workload_cleanup,
165 "workload_cleanup_barrier": workload_barrier,
166 "stack_absence_proof": absence_proof,
167 "stack_absence_completion_proof": completion_proof,
168 "stack_absence": final_absence,
169 "retained_cleanup": retained_cleanup,
170 "attempts": ctx.checkpoint.state.get("destroy_attempts", []),
171 "workload_cleanup_attempts": ctx.checkpoint.state.get("workload_cleanup_attempts", []),
172 "retained_cleanup_attempts": ctx.checkpoint.state.get("retained_cleanup_attempts", []),
173 }
174 if ctx.checkpoint.destroyed:
175 ctx.checkpoint.destroyed = False
176 for action_name in ("destroy", "final-inventory"):
177 if action_name in ctx.checkpoint.completed_actions:
178 ctx.checkpoint.completed_actions.remove(action_name)
179 ctx.checkpoint.state.setdefault("stale_destroyed_reconciliations", []).append(
180 {"at": utc_now(), "stack_absence": initial_absence}
181 )
182 ctx.persist()
184 if initial_absence["all_absent"]:
185 workload_cleanup, workload_barrier = _resume_workload_cleanup_after_stack_absence(ctx)
186 absence_proof = _record_target_stack_absence(
187 ctx,
188 initial_absence,
189 source="destroy-resume-initial-absence",
190 )
191 _checkpoint_new_ecr_repositories(ctx)
192 _checkpoint_new_ecr_images(ctx)
193 _checkpoint_retained_kms_keys(ctx)
194 retained_cleanup = _retained_resource_cleanup(ctx)
195 final_absence = _verify_target_stack_absence(ctx)
196 if not final_absence["all_absent"]:
197 raise RuntimeError(
198 "A target stack reappeared during resumed retained cleanup: "
199 + json.dumps(final_absence["residual"], sort_keys=True)
200 )
201 completion_proof = _record_target_stack_absence(
202 ctx,
203 final_absence,
204 source="destroy-resume-completion",
205 )
206 ctx.checkpoint.destroyed = True
207 ctx.persist()
208 return {
209 "needed": True,
210 "resumed_after_stack_absence": True,
211 "workload_cleanup": workload_cleanup,
212 "workload_cleanup_barrier": workload_barrier,
213 "stack_absence_proof": absence_proof,
214 "stack_absence_completion_proof": completion_proof,
215 "stack_absence": final_absence,
216 "retained_cleanup": retained_cleanup,
217 "attempts": ctx.checkpoint.state.get("destroy_attempts", []),
218 "workload_cleanup_attempts": ctx.checkpoint.state.get("workload_cleanup_attempts", []),
219 "retained_cleanup_attempts": ctx.checkpoint.state.get("retained_cleanup_attempts", []),
220 }
222 workload_cleanup = cleanup_workloads(ctx)
223 if not workload_cleanup.get("complete"):
224 raise RuntimeError(
225 "Workload cleanup is an unresolved teardown barrier: "
226 + json.dumps(
227 {
228 "errors": workload_cleanup.get("errors", []),
229 "unresolved": workload_cleanup.get("unresolved", []),
230 },
231 sort_keys=True,
232 )
233 )
234 workload_cleanup_barrier = _record_workload_cleanup_barrier(ctx, workload_cleanup)
235 _reconcile_stack_ownership(ctx)
236 _checkpoint_new_ecr_repositories(ctx)
237 _checkpoint_new_ecr_images(ctx)
238 _checkpoint_retained_kms_keys(ctx)
240 attempts = ctx.checkpoint.state.setdefault("destroy_attempts", [])
241 for invocation_attempt in range(1, ctx.settings.destroy_attempts + 1):
242 sequence = len(attempts) + 1
243 started_at = utc_now()
244 helper_outcomes: list[dict[str, Any]] = []
246 def on_cleanup_complete(
247 name: str,
248 details: dict[str, Any],
249 destroy_sequence: int = sequence,
250 outcomes: list[dict[str, Any]] = helper_outcomes,
251 ) -> None:
252 outcome = {
253 "destroy_sequence": destroy_sequence,
254 "name": name,
255 "at": utc_now(),
256 "details": copy.deepcopy(details),
257 }
258 outcomes.append(outcome)
259 ctx.checkpoint.state.setdefault("destroy_helper_outcomes", []).append(outcome)
260 ctx.persist()
262 try:
263 helper_authority = _ensure_log_cleanup_helper(ctx)
264 _reconcile_stack_ownership(ctx)
265 expected_stack_ids = {
266 name: (
267 str(record["stack_id"])
268 if (record := _owned_stack_record(ctx, str(region), name)) is not None
269 else None
270 )
271 for name, region in ctx.checkpoint.state["target_stack_regions"].items()
272 }
273 prepared_change_sets = _prepared_change_set_authority(ctx)
275 def on_prepared(
276 stack_name: str,
277 region: str,
278 stack_id: str,
279 change_set_id: str,
280 change_set_type: str,
281 target_ids: dict[str, str | None] = expected_stack_ids,
282 change_sets: dict[str, dict[str, dict[str, str]]] = prepared_change_sets,
283 ) -> None:
284 _record_prepared_stack_identity(
285 ctx,
286 stack_name,
287 region,
288 stack_id,
289 change_set_id,
290 change_set_type,
291 )
292 target_ids[stack_name] = stack_id
293 change_sets.setdefault(stack_name, {})[change_set_id] = {
294 "change_set_id": change_set_id,
295 "stack_id": stack_id,
296 "change_set_type": change_set_type,
297 }
299 overall, successful, failed = ctx.stack_manager.destroy_orchestrated(
300 force=True,
301 parallel=False,
302 max_workers=1,
303 expected_stack_ids=expected_stack_ids,
304 prepared_change_sets=prepared_change_sets,
305 authorize_stack=lambda name, region, stack_id: _authorize_owned_stack(
306 ctx,
307 name,
308 region,
309 stack_id,
310 ),
311 allow_bootstrap=False,
312 bootstrap_stacks=ctx.checkpoint.state["bootstrap_stacks"],
313 on_cleanup_complete=on_cleanup_complete,
314 strict_deployment_token=f"{ctx.settings.run_id}-teardown",
315 on_change_set_prepared=on_prepared,
316 on_ecr_repository_created=lambda region, repository: (
317 _record_ecr_repository_creation(ctx, region, repository)
318 ),
319 )
320 attempt: dict[str, Any] = {
321 "sequence": sequence,
322 "invocation_attempt": invocation_attempt,
323 "started_at": started_at,
324 "overall_success": overall,
325 "successful_stacks": successful,
326 "failed_stacks": failed,
327 "helper_outcomes": helper_outcomes,
328 "log_cleanup_helper": helper_authority,
329 }
330 if overall:
331 absence_before_cleanup = _verify_target_stack_absence(ctx)
332 attempt["stack_absence_before_retained_cleanup"] = absence_before_cleanup
333 if not absence_before_cleanup["all_absent"]:
334 raise RuntimeError(
335 "Target stack absence was not proved after destroy: "
336 + json.dumps(absence_before_cleanup["residual"], sort_keys=True)
337 )
338 attempt["target_stack_absence_proof"] = _record_target_stack_absence(
339 ctx,
340 absence_before_cleanup,
341 source="destroy-before-retained-cleanup",
342 )
343 attempt["retained_cleanup"] = _retained_resource_cleanup(ctx)
344 absence_before_completion = _verify_target_stack_absence(ctx)
345 attempt["stack_absence_before_completion"] = absence_before_completion
346 if not absence_before_completion["all_absent"]:
347 raise RuntimeError(
348 "A target stack reappeared during retained cleanup: "
349 + json.dumps(absence_before_completion["residual"], sort_keys=True)
350 )
351 attempt["target_stack_absence_completion_proof"] = _record_target_stack_absence(
352 ctx,
353 absence_before_completion,
354 source="destroy-completion",
355 )
356 except Exception as exc: # noqa: BLE001 - retry and preserve teardown evidence
357 overall = False
358 if "attempt" not in locals() or attempt.get("sequence") != sequence:
359 attempt = {
360 "sequence": sequence,
361 "invocation_attempt": invocation_attempt,
362 "started_at": started_at,
363 "successful_stacks": [],
364 "failed_stacks": [],
365 "helper_outcomes": helper_outcomes,
366 }
367 attempt["overall_success"] = False
368 attempt["error"] = f"{type(exc).__name__}: {exc}"
369 if not overall:
370 try:
371 attempt["log_cleanup_helper_cleanup"] = _delete_log_cleanup_helper(ctx)
372 except Exception as helper_exc: # noqa: BLE001 - retain both teardown failures
373 helper_error = f"{type(helper_exc).__name__}: {helper_exc}"
374 attempt["log_cleanup_helper_cleanup_error"] = helper_error
375 previous_error = str(attempt.get("error") or "")
376 attempt["error"] = "; ".join(
377 part for part in (previous_error, f"cleanup helper: {helper_error}") if part
378 )
379 attempt["ended_at"] = utc_now()
380 attempts.append(attempt)
381 ctx.persist()
382 if overall:
383 ctx.checkpoint.destroyed = True
384 ctx.persist()
385 return {
386 "needed": True,
387 "workload_cleanup": workload_cleanup,
388 "workload_cleanup_barrier": workload_cleanup_barrier,
389 "workload_cleanup_attempts": ctx.checkpoint.state.get(
390 "workload_cleanup_attempts", []
391 ),
392 "attempts": attempts,
393 "retained_cleanup_attempts": ctx.checkpoint.state.get(
394 "retained_cleanup_attempts", []
395 ),
396 "stack_absence": attempt["stack_absence_before_completion"],
397 }
398 if invocation_attempt < ctx.settings.destroy_attempts:
399 time.sleep(ctx.settings.destroy_retry_delay_seconds)
401 last_attempt = attempts[-1]
402 last_failure = last_attempt.get("error") or ", ".join(last_attempt.get("failed_stacks", []))
403 raise RuntimeError(
404 "Orchestrated teardown did not succeed after "
405 f"{ctx.settings.destroy_attempts} invocation attempts; last failure: "
406 f"{last_failure or 'unknown'}"
407 )
410def action_destroy(ctx: RunContext) -> dict[str, Any]:
411 """Destroy all run-owned stacks and retained resources."""
412 return destroy_deployment(ctx)