Coverage for scripts / live_release_validation / actions / convergence.py: 100.00%

1"""convergence: require stable SQS/DLQ and DynamoDB convergence.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import time 

7from typing import Any 

8 

9from ..checks.central_queue import ( 

10 _read_central_job_item, 

11) 

12from ..checks.topology import ( 

13 _queue_counts, 

14) 

15from ..models import RunContext 

16 

17 

18def action_convergence(ctx: RunContext) -> dict[str, Any]: 

19 """Require stable empty SQS/DLQ counters and terminal DynamoDB records.""" 

20 baseline = ctx.checkpoint.state.get("queue_baseline") 

21 if not baseline: 

22 raise RuntimeError("Topology action did not record queue baselines") 

23 

24 deadline = time.monotonic() + ctx.settings.queue_timeout_seconds 

25 stable_observations = 0 

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

27 while time.monotonic() < deadline: 

28 sample = { 

29 region: ctx.job_manager.get_queue_status(region) for region in ctx.deployment_regions 

30 } 

31 counts = {region: _queue_counts(status) for region, status in sample.items()} 

32 samples.append({"at": time.time(), "counts": counts}) 

33 expected_dlq = {region: _queue_counts(status)["dlq"] for region, status in baseline.items()} 

34 converged = all( 

35 values["available"] == 0 

36 and values["in_flight"] == 0 

37 and values["delayed"] == 0 

38 and values["dlq"] == expected_dlq.get(region, 0) 

39 for region, values in counts.items() 

40 ) 

41 stable_observations = stable_observations + 1 if converged else 0 

42 if stable_observations >= 3: 

43 break 

44 time.sleep(ctx.settings.poll_interval_seconds) 

45 if stable_observations < 3: 

46 raise TimeoutError( 

47 "Regional SQS/DLQ counters did not converge for three observations: " 

48 + json.dumps(samples[-5:], sort_keys=True) 

49 ) 

50 

51 dynamodb_records: dict[str, Any] = {} 

52 for central_job in ctx.checkpoint.state.get("central_jobs", []): 

53 job_id = str(central_job["job_id"]) 

54 item = _read_central_job_item(ctx, job_id) 

55 if item.get("status") != "succeeded": 

56 raise RuntimeError(f"DynamoDB record {job_id} regressed to {item.get('status')}") 

57 dynamodb_records[job_id] = item 

58 return { 

59 "stable_observations": stable_observations, 

60 "queue_samples": samples[-10:], 

61 "dynamodb_records": dynamodb_records, 

62 }