From 7e00450db09b0ea3bd44f88654f4e378185ad2e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Thu, 22 Oct 2026 07:52:47 +0200 Subject: [PATCH 1/6] test(v0.49): attest installed per-cycle claim bound and observed worker steps --- tests/integrations/pg-mdm/time/audit.py | 72 +++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/tests/integrations/pg-mdm/time/audit.py b/tests/integrations/pg-mdm/time/audit.py index 26882c7..7d402c6 100644 --- a/tests/integrations/pg-mdm/time/audit.py +++ b/tests/integrations/pg-mdm/time/audit.py @@ -109,6 +109,42 @@ def qualify_two_case_outage(before, source, temporal_source): " ORDER BY c.case_key) FROM mdm_steward.policy_cases_v1 c " "WHERE c.case_key IN (" + ",".join(str(k) for k in sorted(selected)) + ")")) controls = {c["case_key"]: c["control"] for c in expected_controls} + # Prove the actual installed managed worker consumes the *same* + # numeric claim bound tested by the public one-item claim fixture. + # batch_size alone is not evidence; the installed body must feed it + # to the claim API, which in turn must enforce max_items. + managed_definition = sql( + "SELECT pg_get_functiondef(" + "'pgreact_internal.managed_cycle(integer)'::regprocedure)") + claim_definition = sql( + "SELECT pg_get_functiondef(" + "'pgreact.claim(text,integer,interval,text[])'::regprocedure)") + normalized_managed = " ".join(managed_definition.split()) + normalized_claim = " ".join(claim_definition.split()) + assert "batch_limit integer := current_setting('pg_react.batch_size')::integer" in normalized_managed, ( + "deployed managed cycle does not read the configured batch claim limit") + assert "pgreact_api.claim(worker, batch_limit, interval '60 seconds')" in normalized_managed, ( + "deployed worker no longer passes the batch limit to the public claim API") + assert "processed := processed + 1" in normalized_managed, ( + "deployed managed cycle no longer counts each claimed execution") + assert "EXIT WHEN count_claimed >= max_items" in normalized_claim, ( + "deployed public claim no longer stops at its requested numeric bound") + assert "IF max_items NOT BETWEEN 1 AND claim_limit" in normalized_claim, ( + "deployed public claim no longer checks the operational bound") + claim_limit = int(sql( + "SELECT max_claims FROM pgreact_internal.operational_settings")) + assert claim_limit >= 1, claim_limit + observations["installed_cycle_claim_bound"] = { + "managed_function": "pgreact_internal.managed_cycle(integer)", + "managed_definition_sha256": hashlib.sha256( + managed_definition.encode()).hexdigest(), + "claim_function": "pgreact.claim(text,integer,interval,text[])", + "claim_definition_sha256": hashlib.sha256( + claim_definition.encode()).hexdigest(), + "per_cycle_configured_claim_limit": 1, + "public_operational_max_claims": claim_limit, + "managed_to_claim_bound_verified": True, + } original_postmaster = timestamp(sql("SELECT pg_postmaster_start_time()")) sql("ALTER SYSTEM SET pg_react.databases = 's1_idle'") sql("ALTER SYSTEM SET pg_react.worker_role = 'mdm_s1_runtime'") @@ -155,6 +191,7 @@ def qualify_two_case_outage(before, source, temporal_source): "worker_state": (status.get("process") or {}).get("state"), "pending_jobs": (status.get("process") or {}).get("pending_jobs"), "processed_jobs": (status.get("process") or {}).get("processed_jobs"), + "heartbeat_at": (status.get("process") or {}).get("heartbeat_at"), }) last_keys = keys if len(received) == 2: @@ -168,6 +205,39 @@ def qualify_two_case_outage(before, source, temporal_source): assert receipt_progress and receipt_progress[-1]["receipt_count"] == 2 assert all(sample["receipt_count"] in (0, 1, 2) for sample in receipt_progress), receipt_progress + # Polling is observational, so skipped heartbeats cannot be construed + # as a per-cycle sample. Compare only observed adjacent heartbeat + # epochs spaced by at most 1.5 poll intervals and retain the rest. + observed_cycle_steps = [] + latest_heartbeat = None + latest_processed = None + for sample in receipt_progress: + hb = sample["heartbeat_at"] + processed = sample["processed_jobs"] + if hb is None or processed is None: + continue + parsed = timestamp(hb) + if latest_heartbeat is not None and parsed > latest_heartbeat: + elapsed = (parsed - latest_heartbeat).total_seconds() + delta = processed - latest_processed + assert delta >= 0, ("managed cumulative processed counter regressed", sample) + close_enough = elapsed <= 1.5 + if close_enough: + assert delta <= 1, { + "error": "managed cycle processed more than batch_size=1", + "last_heartbeat": latest_heartbeat.isoformat(), + "heartbeat": hb, "previous_total": latest_processed, + "new_total": processed} + observed_cycle_steps.append({ + "from_heartbeat": latest_heartbeat.isoformat(), + "to_heartbeat": hb, "elapsed_seconds": elapsed, + "processed_delta": delta, + "single_cycle_observation": close_enough, + }) + if latest_heartbeat is None or parsed > latest_heartbeat: + latest_heartbeat = parsed + latest_processed = processed + observations["observed_cycle_steps"] = observed_cycle_steps new_postmaster = timestamp(sql("SELECT pg_postmaster_start_time()")) assert new_postmaster > original_postmaster assert new_postmaster > max(selected.values()) and new_postmaster >= restarted_at @@ -293,6 +363,8 @@ def qualify_two_case_outage(before, source, temporal_source): "configured_limits": settings, "two_case_drain_seconds": drain_elapsed_seconds, "observed_receipt_progress": receipt_progress, + "per_cycle_claim_bound": observations["installed_cycle_claim_bound"], + "observed_cycle_steps": observed_cycle_steps, "two_receipts": received, "before": before_state, "after": actual, From d9e930ab399fd660aece9b9d657797b6480d38de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Thu, 22 Oct 2026 07:53:14 +0200 Subject: [PATCH 2/6] docs(runtime): accurately describe managed-cycle claim batch limit --- src/managed.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/managed.rs b/src/managed.rs index e6601de..03acc43 100644 --- a/src/managed.rs +++ b/src/managed.rs @@ -91,7 +91,7 @@ pub extern "C-unwind" fn _PG_init() { GucRegistry::define_int_guc( c"pg_react.batch_size", c"Maximum managed maintenance batch size.", - c"Window maintenance uses this bound; managed job claims retain the public 100-item maximum.", + c"Managed maintenance and per-cycle job claims use this bound; claims also honor the operational maximum.", &BATCH_SIZE, 1, 1_000, From 24fbee3db4ce3eb692f501359777e3bdfd13cca9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Thu, 22 Oct 2026 07:53:30 +0200 Subject: [PATCH 3/6] ci(v0.49): require deployed per-cycle claim limit evidence --- .github/workflows/producer-compatibility.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/producer-compatibility.yml b/.github/workflows/producer-compatibility.yml index 776dc30..73c27e4 100644 --- a/.github/workflows/producer-compatibility.yml +++ b/.github/workflows/producer-compatibility.yml @@ -57,7 +57,7 @@ jobs: set -euo pipefail python3 tests/integrations/pg-mdm/time/audit.py \ pgreact:producer-compatibility producer-evidence/two-case-outage --multi-outage - python3 -c 'import json; o=json.load(open("producer-evidence/two-case-outage/observations.json")); checks={c["name"] for c in o["checks"] if c["result"]=="passed"}; assert o["entry_decision"]=="FOCUSED_TWO_CASE_OUTAGE_QUALIFIED", o.get("failure"); assert "two-case-real-outage-drain-and-no-replay" in checks, checks; e=o["multi_outage"]; assert e["exact"] and e["no_replay_across_cycles_and_restart"] and len(e["two_receipts"])==2 and e["configured_limits"]["pg_react.batch_size"]=="1"; assert 0 < e["two_case_drain_seconds"] <= 150 and e["observed_receipt_progress"][-1]["receipt_count"]==2' + python3 -c 'import json; o=json.load(open("producer-evidence/two-case-outage/observations.json")); checks={c["name"] for c in o["checks"] if c["result"]=="passed"}; assert o["entry_decision"]=="FOCUSED_TWO_CASE_OUTAGE_QUALIFIED", o.get("failure"); assert "two-case-real-outage-drain-and-no-replay" in checks, checks; e=o["multi_outage"]; assert e["exact"] and e["no_replay_across_cycles_and_restart"] and len(e["two_receipts"])==2 and e["configured_limits"]["pg_react.batch_size"]=="1"; assert 0 < e["two_case_drain_seconds"] <= 150 and e["observed_receipt_progress"][-1]["receipt_count"]==2; assert e["per_cycle_claim_bound"]["managed_to_claim_bound_verified"] and e["per_cycle_claim_bound"]["per_cycle_configured_claim_limit"]==1' - name: Record and compare inherited packaged shared libraries run: | set -euo pipefail From ef636ac43a3988a5be83cec2bd940e6bacd108cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Thu, 22 Oct 2026 07:58:28 +0200 Subject: [PATCH 4/6] docs(runtime): clarify that managed batch must satisfy claim limit --- src/managed.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/managed.rs b/src/managed.rs index 03acc43..1cf0d09 100644 --- a/src/managed.rs +++ b/src/managed.rs @@ -90,8 +90,8 @@ pub extern "C-unwind" fn _PG_init() { ); GucRegistry::define_int_guc( c"pg_react.batch_size", - c"Maximum managed maintenance batch size.", - c"Managed maintenance and per-cycle job claims use this bound; claims also honor the operational maximum.", + c"Maximum managed maintenance and per-cycle claim batch size.", + c"Claim batch size must not exceed the operational max_claims setting.", &BATCH_SIZE, 1, 1_000, From 2bbe7463e5e6a78393ce1446ce86acc43bead460 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Thu, 22 Oct 2026 08:59:34 +0200 Subject: [PATCH 5/6] test(v0.49): trace installed managed-cycle delegate to bounded claim --- tests/integrations/pg-mdm/time/audit.py | 37 ++++++++++++++++++------- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/tests/integrations/pg-mdm/time/audit.py b/tests/integrations/pg-mdm/time/audit.py index 7d402c6..15d6758 100644 --- a/tests/integrations/pg-mdm/time/audit.py +++ b/tests/integrations/pg-mdm/time/audit.py @@ -109,24 +109,35 @@ def qualify_two_case_outage(before, source, temporal_source): " ORDER BY c.case_key) FROM mdm_steward.policy_cases_v1 c " "WHERE c.case_key IN (" + ",".join(str(k) for k in sorted(selected)) + ")")) controls = {c["case_key"]: c["control"] for c in expected_controls} - # Prove the actual installed managed worker consumes the *same* - # numeric claim bound tested by the public one-item claim fixture. - # batch_size alone is not evidence; the installed body must feed it - # to the claim API, which in turn must enforce max_items. + # The installed compatibility gate delegates to the bound-carrying + # managed worker; inspect both deployed definitions and the public + # claim implementation rather than assuming a direct GUC read. managed_definition = sql( "SELECT pg_get_functiondef(" "'pgreact_internal.managed_cycle(integer)'::regprocedure)") + worker_definition = sql( + "SELECT pg_get_functiondef(" + "'pgreact_internal.managed_cycle_m44(integer)'::regprocedure)") claim_definition = sql( "SELECT pg_get_functiondef(" "'pgreact.claim(text,integer,interval,text[])'::regprocedure)") normalized_managed = " ".join(managed_definition.split()) + normalized_worker = " ".join(worker_definition.split()) normalized_claim = " ".join(claim_definition.split()) - assert "batch_limit integer := current_setting('pg_react.batch_size')::integer" in normalized_managed, ( - "deployed managed cycle does not read the configured batch claim limit") - assert "pgreact_api.claim(worker, batch_limit, interval '60 seconds')" in normalized_managed, ( - "deployed worker no longer passes the batch limit to the public claim API") - assert "processed := processed + 1" in normalized_managed, ( - "deployed managed cycle no longer counts each claimed execution") + assert "RETURN pgreact_internal.managed_cycle_m44(process_pid)" in normalized_managed, ( + "deployed compatibility gate no longer reaches the bounded managed worker") + assert "batch_limit integer := current_setting('pg_react.batch_size')::integer" in normalized_worker, ( + "deployed managed worker does not read the configured batch claim limit") + assert "claim_limit integer := least(batch_limit, 100)" in normalized_worker, ( + "deployed managed worker no longer derives the bounded claim size") + assert "pgreact_api.claim(worker, claim_limit, interval '60 seconds')" in normalized_worker, ( + "deployed worker no longer forwards the derived claim limit") + assert "processed := processed + 1" in normalized_worker, ( + "deployed managed worker no longer counts each claimed execution") + assert "processed_jobs = processed_jobs + processed" in normalized_worker, ( + "deployed worker no longer persists cumulative processed counts") + assert "count_claimed := count_claimed + 1" in normalized_claim, ( + "deployed public claim no longer increments its claim count") assert "EXIT WHEN count_claimed >= max_items" in normalized_claim, ( "deployed public claim no longer stops at its requested numeric bound") assert "IF max_items NOT BETWEEN 1 AND claim_limit" in normalized_claim, ( @@ -138,6 +149,10 @@ def qualify_two_case_outage(before, source, temporal_source): "managed_function": "pgreact_internal.managed_cycle(integer)", "managed_definition_sha256": hashlib.sha256( managed_definition.encode()).hexdigest(), + "bound_carrying_function": "pgreact_internal.managed_cycle_m44(integer)", + "bound_carrying_definition_sha256": hashlib.sha256( + worker_definition.encode()).hexdigest(), + "forwarded_claim_limit_expression": "least(batch_limit, 100)", "claim_function": "pgreact.claim(text,integer,interval,text[])", "claim_definition_sha256": hashlib.sha256( claim_definition.encode()).hexdigest(), @@ -307,6 +322,8 @@ def qualify_two_case_outage(before, source, temporal_source): "SELECT jsonb_object_agg(name,setting) FROM pg_settings WHERE name IN " "('pg_react.poll_interval_ms','pg_react.batch_size','pg_react.max_pending_jobs')")) assert settings["pg_react.batch_size"] == "1", settings + assert int(settings["pg_react.batch_size"]) <= claim_limit, ( + "managed batch exceeds operational max_claims", settings, claim_limit) first_status = json.loads(sql("SELECT pgreact_api.managed_status()")) heartbeat = timestamp(first_status["process"]["heartbeat_at"]) limit = time.monotonic() + 35 From 46b98a958898169297d7ece308764250f1501eaf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Thu, 22 Oct 2026 09:26:57 +0200 Subject: [PATCH 6/6] test(v0.49): avoid inferring a worker cycle from adjacent status polls --- tests/integrations/pg-mdm/time/audit.py | 29 +++++++++++-------------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/tests/integrations/pg-mdm/time/audit.py b/tests/integrations/pg-mdm/time/audit.py index 15d6758..6bc4c0e 100644 --- a/tests/integrations/pg-mdm/time/audit.py +++ b/tests/integrations/pg-mdm/time/audit.py @@ -220,10 +220,13 @@ def qualify_two_case_outage(before, source, temporal_source): assert receipt_progress and receipt_progress[-1]["receipt_count"] == 2 assert all(sample["receipt_count"] in (0, 1, 2) for sample in receipt_progress), receipt_progress - # Polling is observational, so skipped heartbeats cannot be construed - # as a per-cycle sample. Compare only observed adjacent heartbeat - # epochs spaced by at most 1.5 poll intervals and retain the rest. - observed_cycle_steps = [] + # Two status polls cannot establish one worker cycle. A background + # worker can complete several cycles, or update heartbeat twice within + # one cycle, between observations regardless of elapsed time. Preserve + # the measured cumulative-counter deltas without guessing a cycle + # identity. The numeric per-call bound is proved separately by the + # installed SQL call chain and the focused max_items=1 claim test. + observed_counter_intervals = [] latest_heartbeat = None latest_processed = None for sample in receipt_progress: @@ -236,23 +239,17 @@ def qualify_two_case_outage(before, source, temporal_source): elapsed = (parsed - latest_heartbeat).total_seconds() delta = processed - latest_processed assert delta >= 0, ("managed cumulative processed counter regressed", sample) - close_enough = elapsed <= 1.5 - if close_enough: - assert delta <= 1, { - "error": "managed cycle processed more than batch_size=1", - "last_heartbeat": latest_heartbeat.isoformat(), - "heartbeat": hb, "previous_total": latest_processed, - "new_total": processed} - observed_cycle_steps.append({ + observed_counter_intervals.append({ "from_heartbeat": latest_heartbeat.isoformat(), - "to_heartbeat": hb, "elapsed_seconds": elapsed, + "to_heartbeat": hb, + "elapsed_seconds": elapsed, "processed_delta": delta, - "single_cycle_observation": close_enough, + "individual_cycle_attribution": False, }) if latest_heartbeat is None or parsed > latest_heartbeat: latest_heartbeat = parsed latest_processed = processed - observations["observed_cycle_steps"] = observed_cycle_steps + observations["observed_counter_intervals"] = observed_counter_intervals new_postmaster = timestamp(sql("SELECT pg_postmaster_start_time()")) assert new_postmaster > original_postmaster assert new_postmaster > max(selected.values()) and new_postmaster >= restarted_at @@ -381,7 +378,7 @@ def qualify_two_case_outage(before, source, temporal_source): "two_case_drain_seconds": drain_elapsed_seconds, "observed_receipt_progress": receipt_progress, "per_cycle_claim_bound": observations["installed_cycle_claim_bound"], - "observed_cycle_steps": observed_cycle_steps, + "observed_counter_intervals": observed_counter_intervals, "two_receipts": received, "before": before_state, "after": actual,