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 diff --git a/src/managed.rs b/src/managed.rs index e6601de..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"Window maintenance uses this bound; managed job claims retain the public 100-item 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, diff --git a/tests/integrations/pg-mdm/time/audit.py b/tests/integrations/pg-mdm/time/audit.py index b612746..262363a 100644 --- a/tests/integrations/pg-mdm/time/audit.py +++ b/tests/integrations/pg-mdm/time/audit.py @@ -109,6 +109,57 @@ 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} + # 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 "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, ( + "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(), + "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(), + "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 +206,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 +220,36 @@ 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 + # 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: + 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) + observed_counter_intervals.append({ + "from_heartbeat": latest_heartbeat.isoformat(), + "to_heartbeat": hb, + "elapsed_seconds": elapsed, + "processed_delta": delta, + "individual_cycle_attribution": False, + }) + if latest_heartbeat is None or parsed > latest_heartbeat: + latest_heartbeat = parsed + latest_processed = processed + 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 @@ -287,6 +369,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 @@ -343,6 +427,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_counter_intervals": observed_counter_intervals, "two_receipts": received, "before": before_state, "after": actual,