From 2beb3b24b4947d57598e1a4b37a4db72053fd6b5 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 17:17:28 +0000 Subject: [PATCH] feat(reporting-storage): deliver reporting subscriptions through the PostgreSQL production store PgReportingProductionStore now serves managed subscription delivery, as the memory store does since the previous change. Both stores keep one admitted owed set per generation, and each owed obligation's frozen constituents. - PostgreSQL keeps the enrollment (producer and frozen destination binding) and the constituents in two private tables, adcp_reporting_production_owed_sets and adcp_reporting_production_owed_constituents. The constituents are written in the obligation's transaction and never rewritten. - Planning, dispatch and the materializer context read the owed set; an owed set has no generation-wide source binding. Activation re-enrolls owed sets from their bindings. - Reconciliation records, materializer discovery and status checkpoints record the generation's offering. A destination binding's record_id derives from its full generation key, so offerings no longer share an identity: the SQL validator previously hashed only (account, consumer, id, version) and would have rejected every subscription binding. Binding lookups in the validators and the materializer scope carry the offering. - create_schema drops the account-only CHECKs on those three tables and widens the materializer discovery key to include offering_id, in place and idempotently, for schemas created by the previous release. Earlier workers refuse the upgraded schema: stop them first. Production generations keep their account-only CHECK, and the obligation trigger joins them on the offering too. - Status checkpoints are still keyed per (account, consumer, id, version), so both stores refuse a second offering of one subscription version for the same account and owner with reporting_production_offering_conflict. Under that rule the PostgreSQL production store serves owed sets with status notifications enabled. - The required-schema manifests and pinned catalog counts are regenerated. The managed-delivery tests run against both stores, with and without status notifications, and cover the second-offering refusal and the schema upgrade. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 19 +- docs/reporting-ledger-migration.md | 13 +- .../ledger/delivery_pg.py | 5 +- .../src/adcp_reporting_storage/ledger/pg.py | 14 +- .../reporting_ledger_account_generations.sql | 18 +- .../reporting_ledger_reconciliation.sql | 51 +++- .../ledger/reporting_materializer.sql | 34 ++- .../ledger/reporting_production.sql | 27 +- .../ledger/reporting_status_notifications.sql | 12 +- .../adcp_reporting_storage/materializer/pg.py | 36 ++- .../materializer/required_schema.json | 12 +- .../outbox/required_schema.json | 10 +- .../outbox/required_status_schema.json | 4 - .../outbox/status_pg.py | 15 +- .../adcp_reporting_storage/production/pg.py | 266 +++++++++++++++++- .../production/required_schema.json | 2 +- src/adcp/reporting/ledger/_delivery_state.py | 2 + src/adcp/reporting/production/memory.py | 9 +- .../test_reporting_feed_installed_pg.py | 2 +- .../test_reporting_feed_migration.py | 4 +- .../test_reporting_provisional_catalog.py | 2 +- ...est_reporting_provisional_catalog_scope.py | 4 +- .../test_reporting_receipt_migration.py | 2 +- .../test_reporting_subscription_ledger_pg.py | 43 +++ .../test_reporting_subscription_production.py | 151 ++++++++-- 25 files changed, 643 insertions(+), 114 deletions(-) diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index a3240e078..12fbb4b31 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -522,14 +522,25 @@ Supply `bind_managed_generation(configuration, version)`. stays pending under the same identity and is retried. - **Requests.** The mounted production handler checks each task against the grant and adds `subscription_coverage`. +- **Stores.** The memory and PostgreSQL production stores both serve managed + subscription delivery, with or without status notifications. PostgreSQL + keeps each owed set's enrollment, and each owed obligation's frozen + constituents, in the private `adcp_reporting_production_owed_sets` and + `adcp_reporting_production_owed_constituents` tables. Its reconciliation + records, materializer discovery and status checkpoints record the + generation's offering. Running `create_schema` with this release lifts + those tables' account-only checks and widens the discovery key in place. **Limits.** These are not served yet: - subscriptions that resolve a receipt-reconciled (`consumer_receipt`) offering; -- PostgreSQL production stores for subscriptions (they are refused at startup); -- status notifications for subscription generations: a PostgreSQL ledger store - with them enabled refuses an owed set. `PgReportingLedgerStore` otherwise holds - owed sets, after the generation-key widening described in +- more than one offering of one subscription version for the same account and + owner under production. Status checkpoints hold one configuration scope per + version, so admitting a second offering fails that subscription's planning + with `reporting_production_offering_conflict`; +- status notifications for subscription generations outside production: a + `PgReportingLedgerStore` with them enabled refuses an owed set. It otherwise + holds owed sets, after the generation-key widening described in [ledger migrations](reporting-ledger-migration.md). Keep the engine's `status_retention_days` equal to the service's. diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md index f84692566..40b422863 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -68,10 +68,19 @@ In PostgreSQL this is a **stopped-worker maintenance upgrade**: - `reporting_consumer_statuses` gains `offering_id`; - the SDK's own references (reconciliation records, materializer discovery, status scope checkpoints, production generations) are re-pointed at the - widened key. They hold account-level generations only, enforced by a - `CHECK`, until each feature serves subscriptions. + widened key. Reconciliation records, materializer discovery and status + scope checkpoints also hold subscription generations, for managed + delivery. Production generations hold account-level generations only, + enforced by a `CHECK`. 3. Restart the workers on the new release. +A schema created by a release that widened the key but did not yet serve +managed subscription delivery still carries an account-only `CHECK` on +reconciliation records, materializer discovery and status scope checkpoints, +and a four-column materializer discovery key. `create_schema()` drops those +checks and widens the discovery key in place, losslessly and idempotently. +Earlier workers refuse the upgraded schema, so stop them first here too. + A foreign key outside the SDK that references `reporting_configurations` refuses the upgrade with a maintenance error, rather than being dropped. Re-point it at the widened key, then run the upgrade. The upgrade is diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/delivery_pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/delivery_pg.py index 881a234f3..ea407ddc9 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/delivery_pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/delivery_pg.py @@ -75,6 +75,7 @@ "receipt_chain_key", "receipt_status", "supersedes_receipt_id", + "offering_id", ) _SELECT_IDENTITY = ", ".join("r." + column for column in _IDENTITY_COLUMNS) _FEED_JOIN = ( @@ -377,9 +378,9 @@ async def _insert(self, connection: Any, record: ReportingDeliveryRecord) -> Non " (account_id, consumer_id, namespace, record_id, record_kind, delivery_config_id," " delivery_config_version, reporting_obligation_id, reporting_revision_id," " reporting_materialization_id, reporting_adjustment_id, attempt_number," - " receipt_chain_key, receipt_status, supersedes_receipt_id, payload," + " receipt_chain_key, receipt_status, supersedes_receipt_id, offering_id, payload," " content_sha256, change_id)" - " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s," + " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s," " %s::jsonb, %s, %s)", ( *storage_identity(record), diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 4d0746ce4..65408316c 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -929,11 +929,19 @@ async def _lock_account(connection: Any, account_id: str) -> None: # -- configurations --------------------------------------------------- + def _status_covers_owed_sets(self) -> bool: + """Whether status notifications may cover reporting-subscription generations. + + A status checkpoint records its generation's offering but is keyed per + (account, consumer, delivery_config_id, version): two offerings of one + subscription version would share a configuration checkpoint. Only a + store that admits at most one offering per version may serve them. + """ + return not self._notifications_enabled + async def put_configuration(self, configuration: ReportingConfiguration) -> None: reject_reserved_authoritative_party(configuration) - if configuration.owed_set and self._notifications_enabled: - # Status checkpoints are keyed per account-level generation; they - # do not yet hold one per resolved offering. + if configuration.owed_set and not self._status_covers_owed_sets(): raise LedgerConflictError( "UNSUPPORTED_FEATURE", "status notifications do not yet cover reporting-subscription generations", diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql index 4c8882a94..250b1343c 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql @@ -91,13 +91,13 @@ BEGIN ALTER TABLE reporting_consumer_statuses ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''; - -- These tables serve account-level generations only; a subscription - -- generation is refused by their CHECK until each feature serves it. + -- Managed delivery records subscription generations under their offering + -- (reconciliation records, materializer discovery and status checkpoints). + -- Production generations serve account-level generations only; a + -- subscription generation is refused by their CHECK. IF to_regclass('reporting_reconciliation_records') IS NOT NULL THEN ALTER TABLE reporting_reconciliation_records - ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' - CONSTRAINT reporting_reconciliation_records_account_generation - CHECK (offering_id = ''), + ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', ADD CONSTRAINT reporting_reconciliation_records_generation_fkey FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) @@ -106,9 +106,7 @@ BEGIN END IF; IF to_regclass('reporting_materializer_discovery') IS NOT NULL THEN ALTER TABLE reporting_materializer_discovery - ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' - CONSTRAINT reporting_materializer_discovery_account_generation - CHECK (offering_id = ''), + ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', ADD CONSTRAINT reporting_materializer_discovery_generation_fkey FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) @@ -117,9 +115,7 @@ BEGIN END IF; IF to_regclass('reporting_status_scope_checkpoints') IS NOT NULL THEN ALTER TABLE reporting_status_scope_checkpoints - ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' - CONSTRAINT reporting_status_scope_checkpoints_account_generation - CHECK (offering_id = ''), + ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', ADD CONSTRAINT reporting_status_scope_checkpoints_generation_fkey FOREIGN KEY (account_id, consumer_namespace, delivery_config_id, version, offering_id) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql index b79a6797d..440975c5d 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql @@ -108,10 +108,9 @@ BEGIN CHECK ((receipt_status IS NOT NULL) = (receipt_chain_key IS NOT NULL)), CHECK ((record_kind = 'materialization_attempt') = (attempt_number IS NOT NULL)), CHECK (payload->>'kind' = record_kind), - -- Account-level generations only, until managed delivery serves - -- reporting subscriptions (see reporting_ledger_account_generations.sql). - offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' - CONSTRAINT reporting_reconciliation_records_account_generation CHECK (offering_id = ''), + -- '' for an account-level generation; a reporting subscription's + -- generation names its resolved offering (managed delivery). + offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', CONSTRAINT reporting_reconciliation_records_generation_fkey FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id, @@ -132,6 +131,14 @@ BEGIN CREATE UNIQUE INDEX IF NOT EXISTS reporting_receipt_one_successor ON reporting_reconciliation_records(account_id, consumer_id, supersedes_receipt_id) WHERE supersedes_receipt_id IS NOT NULL; + -- A schema created by an earlier release refused subscription generations + -- here; managed delivery now records them under their offering. + IF EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'reporting_reconciliation_records'::regclass + AND conname = 'reporting_reconciliation_records_account_generation') THEN + ALTER TABLE reporting_reconciliation_records + DROP CONSTRAINT IF EXISTS reporting_reconciliation_records_account_generation; + END IF; CREATE TABLE IF NOT EXISTS reporting_receipt_heads ( account_id TEXT COLLATE "C" NOT NULL, @@ -432,8 +439,10 @@ BEGIN RETURN; END IF; SELECT b.payload INTO binding FROM reporting_reconciliation_records b - WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version) - = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version) + WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version, + b.offering_id) + = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version, + r.offering_id) AND b.record_kind = 'destination_binding'; SELECT d.payload INTO delivery FROM reporting_reconciliation_records d WHERE (d.account_id, d.consumer_id, d.delivery_config_id, d.delivery_config_version, @@ -527,12 +536,17 @@ BEGIN 'verification_path', 'verification_profile', 'verified_at', 'verified_format'])) THEN RAISE EXCEPTION 'reporting payload is not closed retained evidence' USING ERRCODE = '23514'; END IF; + -- An account-level generation key omits offering_id (ledger_asdict). generation := jsonb_build_object('account_id', r.account_id, 'consumer_id', r.consumer_id, 'delivery_config_id', r.delivery_config_id, - 'delivery_config_version', r.delivery_config_version); + 'delivery_config_version', r.delivery_config_version) + || CASE WHEN r.offering_id = '' THEN '{}'::jsonb + ELSE jsonb_build_object('offering_id', r.offering_id) END; IF NOT EXISTS (SELECT 1 FROM reporting_configurations c - WHERE (c.account_id, c.consumer_id, c.delivery_config_id, c.delivery_config_version) - = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version)) + WHERE (c.account_id, c.consumer_id, c.delivery_config_id, c.delivery_config_version, + c.offering_id) + = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version, + r.offering_id)) OR (r.reporting_obligation_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM reporting_obligations o WHERE (o.account_id, o.consumer_id, o.delivery_config_id, o.delivery_config_version, o.reporting_obligation_id) @@ -552,7 +566,9 @@ BEGIN WHEN 'destination_binding' THEN reporting_identity_sha256( '{"account_id":' || to_json(r.account_id)::text || ',"consumer_id":' || to_json(r.consumer_id)::text || ',"delivery_config_id":' || to_json(r.delivery_config_id)::text || ',"delivery_config_version":' || - r.delivery_config_version::text || '}') + r.delivery_config_version::text || + CASE WHEN r.offering_id = '' THEN '' + ELSE ',"offering_id":' || to_json(r.offering_id)::text END || '}') WHEN 'obligation_delivery' THEN r.reporting_obligation_id WHEN 'materialization_check' THEN r.payload->>'check_id' WHEN 'revision_receipt' THEN r.payload->>'reporting_receipt_id' @@ -594,8 +610,10 @@ BEGIN 'consumer_id', r.consumer_id, 'reporting_obligation_id', r.reporting_obligation_id) OR NOT EXISTS ( SELECT 1 FROM reporting_reconciliation_records b - WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version) - = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version) + WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version, + b.offering_id) + = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version, + r.offering_id) AND b.record_kind = 'destination_binding') THEN RAISE EXCEPTION 'reporting frozen binding is unavailable' USING ERRCODE = '23514'; END IF; @@ -622,8 +640,10 @@ BEGIN END IF; IF r.record_kind = 'materialization' AND r.payload->>'status' IN ('available', 'delivered') THEN IF EXISTS (SELECT 1 FROM reporting_reconciliation_records b - WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version) - = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version) + WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version, + b.offering_id) + = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version, + r.offering_id) AND b.record_kind = 'destination_binding' AND (r.payload->>'status' IS DISTINCT FROM b.payload->>'success_status' OR r.payload#>>'{verification,verification_profile}' IS DISTINCT FROM b.payload->>'verification_profile')) THEN @@ -740,7 +760,8 @@ BEGIN proposed := proposed - mutable; SELECT EXISTS (SELECT 1 FROM reporting_reconciliation_records r WHERE r.account_id=OLD.account_id AND r.consumer_id=OLD.consumer_id AND r.delivery_config_id = OLD.delivery_config_id - AND r.delivery_config_version = OLD.delivery_config_version) INTO referenced; + AND r.delivery_config_version = OLD.delivery_config_version + AND r.offering_id = OLD.offering_id) INTO referenced; ELSIF TG_TABLE_NAME = 'reporting_obligations' THEN SELECT EXISTS (SELECT 1 FROM reporting_reconciliation_records r WHERE r.account_id = OLD.account_id AND r.reporting_obligation_id = OLD.reporting_obligation_id) INTO referenced; diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_materializer.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_materializer.sql index b57830c58..5eb42725a 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_materializer.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_materializer.sql @@ -23,14 +23,32 @@ BEGIN delivery_config_version BIGINT NOT NULL CHECK (delivery_config_version > 0), after_obligation_id TEXT COLLATE "C" NOT NULL DEFAULT '', complete BOOLEAN NOT NULL DEFAULT FALSE, - offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' - CONSTRAINT reporting_materializer_discovery_account_generation CHECK (offering_id = ''), - PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version), + -- '' for an account-level generation; a reporting subscription's + -- generation names its resolved offering (managed delivery). + offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', + PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, + offering_id), CONSTRAINT reporting_materializer_discovery_generation_fkey FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) ); + -- A schema created by an earlier release keyed discovery without the + -- offering and refused subscription generations: widen it losslessly. + IF EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'reporting_materializer_discovery'::regclass + AND conname = 'reporting_materializer_discovery_account_generation') THEN + ALTER TABLE reporting_materializer_discovery + DROP CONSTRAINT IF EXISTS reporting_materializer_discovery_account_generation; + END IF; + IF EXISTS (SELECT 1 FROM pg_constraint pk + WHERE pk.conrelid = 'reporting_materializer_discovery'::regclass AND pk.contype = 'p' + AND array_length(pk.conkey, 1) = 4) THEN + ALTER TABLE reporting_materializer_discovery + DROP CONSTRAINT reporting_materializer_discovery_pkey, + ADD CONSTRAINT reporting_materializer_discovery_pkey PRIMARY KEY + (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id); + END IF; CREATE INDEX IF NOT EXISTS reporting_materializer_discovery_pending ON reporting_materializer_discovery (account_id, consumer_id, delivery_config_id, delivery_config_version) WHERE NOT complete; @@ -295,6 +313,7 @@ BEGIN WHERE account_id = NEW.account_id AND consumer_id = NEW.consumer_id AND delivery_config_id = NEW.delivery_config_id AND delivery_config_version = NEW.delivery_config_version + AND offering_id = NEW.offering_id ORDER BY consumer_id LOOP PERFORM reporting_materializer_dirty(NEW.account_id, r.consumer_id, @@ -312,8 +331,9 @@ BEGIN IF NEW.record_kind = 'destination_binding' THEN PERFORM pg_advisory_xact_lock(hashtext('adcp.reporting:' || NEW.account_id)); INSERT INTO reporting_materializer_discovery - (account_id, consumer_id, delivery_config_id, delivery_config_version) - VALUES (NEW.account_id, NEW.consumer_id, NEW.delivery_config_id, NEW.delivery_config_version) + (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) + VALUES (NEW.account_id, NEW.consumer_id, NEW.delivery_config_id, + NEW.delivery_config_version, NEW.offering_id) ON CONFLICT DO NOTHING; PERFORM reporting_materializer_wake(NEW.account_id); ELSIF NEW.record_kind = 'materialization_check' THEN @@ -412,8 +432,8 @@ BEGIN -- One-time indexed backfill seeds bindings, not account enumeration or -- recurring whole-ledger scans. Reinstall never rewinds a discovery cursor. INSERT INTO reporting_materializer_discovery - (account_id, consumer_id, delivery_config_id, delivery_config_version) - SELECT account_id, consumer_id, delivery_config_id, delivery_config_version + (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) + SELECT account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id FROM reporting_reconciliation_records WHERE record_kind='destination_binding' ON CONFLICT DO NOTHING; INSERT INTO reporting_materializer_accounts (account_id, due_at) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_production.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_production.sql index e43007149..14f2b5fb3 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_production.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_production.sql @@ -640,7 +640,8 @@ BEGIN SELECT o.account_id, o.consumer_id, o.delivery_config_id, o.delivery_config_version, o.reporting_obligation_id,o.period_end FROM reporting_obligations o JOIN reporting_production_generations g - USING(account_id, consumer_id, delivery_config_id, delivery_config_version) + USING(account_id, consumer_id, delivery_config_id, delivery_config_version, + offering_id) WHERE o.reporting_obligation_id=NEW.reporting_obligation_id AND o.account_id=NEW.account_id ON CONFLICT(account_id,reporting_obligation_id) DO UPDATE SET state='pending'; @@ -675,3 +676,27 @@ BEGIN END IF; END $production_source_progress$; + +-- Reporting-subscription owed sets under managed delivery. Private SDK state: +-- the adcp_ prefix keeps it outside the reporting_* catalog contract, and like +-- the owed-set progress it has no foreign key. An owed set is enrolled once +-- with its producer and frozen managed destination binding. Each owed +-- obligation freezes the constituents it owes from the host's authorized pool, +-- written in the obligation's transaction. Neither is ever rewritten. +CREATE TABLE IF NOT EXISTS adcp_reporting_production_owed_sets ( + account_id TEXT COLLATE "C" NOT NULL, + consumer_id TEXT COLLATE "C" NOT NULL, + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + offering_id TEXT COLLATE "C" NOT NULL CHECK (offering_id <> ''), + producer_key TEXT COLLATE "C" NOT NULL CHECK (producer_key ~ '^[0-9a-f]{64}$'), + method JSONB NOT NULL CHECK (jsonb_typeof(method) = 'object'), + PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id) +); +CREATE TABLE IF NOT EXISTS adcp_reporting_production_owed_constituents ( + account_id TEXT COLLATE "C" NOT NULL, + reporting_obligation_id TEXT COLLATE "C" NOT NULL, + constituents JSONB NOT NULL CHECK ( + jsonb_typeof(constituents) = 'array' AND jsonb_array_length(constituents) > 0), + PRIMARY KEY (account_id, reporting_obligation_id) +); diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_notifications.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_notifications.sql index 6c98f0b70..4d8f240f3 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_notifications.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_notifications.sql @@ -34,8 +34,9 @@ BEGIN lease_expires_at TIMESTAMPTZ, PRIMARY KEY (account_id, consumer_namespace, delivery_config_id, version, scope_kind, obligation_namespace), - offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' - CONSTRAINT reporting_status_scope_checkpoints_account_generation CHECK (offering_id = ''), + -- '' for an account-level generation; a reporting subscription's + -- generation names its resolved offering (managed delivery). + offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', CONSTRAINT reporting_status_scope_checkpoints_generation_fkey FOREIGN KEY (account_id, consumer_namespace, delivery_config_id, version, offering_id) REFERENCES reporting_configurations @@ -51,6 +52,13 @@ BEGIN CHECK (coalesce(scope->>'reporting_obligation_id', '') = obligation_namespace), CHECK (snapshot->>'health' IN ('waiting','healthy','delayed','action_required','complete')) ); + -- A schema created by an earlier release refused subscription generations here. + IF EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'reporting_status_scope_checkpoints'::regclass + AND conname = 'reporting_status_scope_checkpoints_account_generation') THEN + ALTER TABLE reporting_status_scope_checkpoints + DROP CONSTRAINT IF EXISTS reporting_status_scope_checkpoints_account_generation; + END IF; CREATE INDEX IF NOT EXISTS reporting_status_scope_due ON reporting_status_scope_checkpoints (account_id, next_due_at) WHERE next_due_at IS NOT NULL; diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/pg.py index 5d5603eda..15031518d 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/pg.py @@ -105,6 +105,14 @@ def _scope_args(scope: ReportingDeliveryScope) -> tuple[str | int, ...]: ) +#: Appended to a selected candidate or work row: its obligation's offering, +#: '' for an account-level generation. Obligation ids are unique per account. +_WITH_OFFERING = ( + " || jsonb_build_object('offering_id',(SELECT o.offering_id FROM reporting_obligations o" + " WHERE o.account_id={0}.account_id AND o.reporting_obligation_id={0}.reporting_obligation_id))" +) + + def _scope(row: dict[str, Any]) -> ReportingDeliveryScope: return ReportingDeliveryScope( ReportingConfigurationGenerationKey( @@ -112,6 +120,7 @@ def _scope(row: dict[str, Any]) -> ReportingDeliveryScope: row["consumer_id"], row["delivery_config_id"], row["delivery_config_version"], + row.get("offering_id") or None, ), row["consumer_id"], row["reporting_obligation_id"], @@ -255,22 +264,23 @@ async def _discover_on(self, connection: Any, account_id: str) -> bool: row = await ( await connection.execute( "SELECT consumer_id, delivery_config_id, delivery_config_version," - " after_obligation_id" + " after_obligation_id, offering_id" " FROM reporting_materializer_discovery WHERE account_id=%s AND NOT complete" - " ORDER BY consumer_id, delivery_config_id, delivery_config_version" + " ORDER BY consumer_id, delivery_config_id, delivery_config_version, offering_id" " LIMIT 1 FOR UPDATE", (account_id,), ) ).fetchone() if row is None: return False - consumer, config, version, after = row + consumer, config, version, after, offering = row obligations = await ( await connection.execute( "SELECT reporting_obligation_id FROM reporting_obligations WHERE account_id=%s" " AND consumer_id=%s AND delivery_config_id=%s AND delivery_config_version=%s" + " AND offering_id=%s" " AND reporting_obligation_id>%s ORDER BY reporting_obligation_id LIMIT 32", - (account_id, consumer, config, version, after), + (account_id, consumer, config, version, offering, after), ) ).fetchall() for (obligation_id,) in obligations: @@ -286,7 +296,7 @@ async def _discover_on(self, connection: Any, account_id: str) -> bool: await connection.execute( "UPDATE reporting_materializer_discovery SET after_obligation_id=%s,complete=%s" " WHERE account_id=%s AND consumer_id=%s AND delivery_config_id=%s" - " AND delivery_config_version=%s", + " AND delivery_config_version=%s AND offering_id=%s", ( obligations[-1][0] if obligations else after, len(obligations) < 32, @@ -294,6 +304,7 @@ async def _discover_on(self, connection: Any, account_id: str) -> bool: consumer, config, version, + offering, ), ) return True @@ -391,7 +402,9 @@ async def _claim_account_on( now = await _now(connection) pending = await ( await connection.execute( - "SELECT to_jsonb(w) FROM reporting_materializer_work w WHERE account_id=%s" + "SELECT to_jsonb(w)" + + _WITH_OFFERING.format("w") + + " FROM reporting_materializer_work w WHERE account_id=%s" " AND state='pending' AND due_at<=%s" " AND (lease_until IS NULL OR lease_until<=%s)" " ORDER BY due_at,reporting_materialization_id LIMIT 1 FOR UPDATE", @@ -403,7 +416,9 @@ async def _claim_account_on( discovered = await self._discover_on(connection, account_id) row = await ( await connection.execute( - "SELECT to_jsonb(c) FROM reporting_materializer_candidates c WHERE account_id=%s" + "SELECT to_jsonb(c)" + + _WITH_OFFERING.format("c") + + " FROM reporting_materializer_candidates c WHERE account_id=%s" " AND due_at<=%s AND NOT EXISTS (" " SELECT 1 FROM reporting_materializer_work w WHERE w.account_id=c.account_id" " AND w.consumer_id=c.consumer_id AND w.delivery_config_id=c.delivery_config_id" @@ -494,7 +509,8 @@ async def _claim_account_on( ).fetchone() assert inserted is not None await self._park_on(connection, scope, "ready") - return await self._lease_on(connection, inserted[0], keys, lease_seconds) + work = {**inserted[0], "offering_id": scope.generation_key.offering_id or ""} + return await self._lease_on(connection, work, keys, lease_seconds) async def _lease_on( self, @@ -584,7 +600,9 @@ async def _held_on( ) -> dict[str, Any] | None: row = await ( await connection.execute( - "SELECT to_jsonb(w) FROM reporting_materializer_work w WHERE account_id=%s" + "SELECT to_jsonb(w)" + + _WITH_OFFERING.format("w") + + " FROM reporting_materializer_work w WHERE account_id=%s" " AND consumer_id=%s AND reporting_materialization_id=%s AND state='pending'" " AND lease_token=%s::uuid AND lease_until>clock_timestamp() FOR UPDATE", ( diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/required_schema.json b/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/required_schema.json index 505b8b9e1..d35f60cb1 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/required_schema.json +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/materializer/required_schema.json @@ -339,10 +339,6 @@ "enabled": true, "fingerprint": "d2dc21e781a468c735dd8b305b5405e779c85c27657683c19570cb1ba4b94aa7" }, - "constraint:reporting_materializer_discovery.reporting_materializer_discovery_account_generation": { - "enabled": true, - "fingerprint": "8acc83717805dad2b34cd3943ff8ceea21b1a876685f4e3bca5a44e99f38a768" - }, "constraint:reporting_materializer_discovery.reporting_materializer_discovery_delivery_config_version_check": { "enabled": true, "fingerprint": "aa64b81d15585376b0e6a0904344e39bc5010257097cee5c8118967efdac4d34" @@ -353,7 +349,7 @@ }, "constraint:reporting_materializer_discovery.reporting_materializer_discovery_pkey": { "enabled": true, - "fingerprint": "cb0658b07aa5249a6934cca6248293967db4ab1a848d706c3d5310fa98f08785" + "fingerprint": "32ac40fcb53b3780f37f76b062d5a06465e786ee5007f31748b6c964b7b0d20f" }, "constraint:reporting_materializer_notification_events.reporting_materializer_notifi_account_id_consumer_namespac_fkey": { "enabled": true, @@ -577,7 +573,7 @@ }, "function:reporting_materializer_binding_dirty()": { "enabled": true, - "fingerprint": "eea2454f1ded0e1bc966fac4cb567e43520182595e9961cd6052c6b7f5cf94e3" + "fingerprint": "7eb581b8d9e8dbaf55bbcfb3e2ba5fa2e2831b2a2bf5b38cfab9a236f0446eef" }, "function:reporting_materializer_dirty(a text, c text, d text, v bigint, o text)": { "enabled": true, @@ -593,7 +589,7 @@ }, "function:reporting_materializer_source_dirty()": { "enabled": true, - "fingerprint": "b7f635fe5457d6f32114fef36ce994df40dcb4904eff4fa083fcbe34f603d850" + "fingerprint": "611195faa8ecbb85d1bc70e66c7430461a30d34095a4696df72894bfdd613bdf" }, "function:reporting_materializer_wake(a text)": { "enabled": true, @@ -633,7 +629,7 @@ }, "index:reporting_materializer_discovery.reporting_materializer_discovery_pkey": { "enabled": true, - "fingerprint": "b6b8787db0e681f9bee61e99514097d8d0c150fcb78102e439a92c0483521f62" + "fingerprint": "d4f791f892c75bfc2fa84f19e6a16caa1dfe7ea49e4a5c053ff9f48889a37d2b" }, "index:reporting_materializer_notification_events.reporting_materializer_notifi_account_id_consumer_namespace_key": { "enabled": true, diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json index de538ddf0..6dd29edf4 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json @@ -1187,10 +1187,6 @@ "enabled": true, "fingerprint": "6336638ff31659c0ea0a0fa6c67a7ed65e1b1bf792d17b5fbaec138fa6869c17" }, - "constraint:reporting_reconciliation_records.reporting_reconciliation_records_account_generation": { - "enabled": true, - "fingerprint": "8acc83717805dad2b34cd3943ff8ceea21b1a876685f4e3bca5a44e99f38a768" - }, "constraint:reporting_reconciliation_records.reporting_reconciliation_records_attempt_number_check": { "enabled": true, "fingerprint": "f9dafc82e1fb7cbb9f2557001d24a8f78ecd56b79335ed1d32b655ff1b2ded90" @@ -1437,7 +1433,7 @@ }, "function:reporting_reconciliation_evidence(r reporting_reconciliation_records)": { "enabled": true, - "fingerprint": "cfa4e3f5b4c6505d2fe72d9828d95d3760555b1957b6d0754e3bac86c925bfdf" + "fingerprint": "a4da1979850b457d288924c30e1b8c6494d88a0fbc872c421d3aee7916159971" }, "function:reporting_reconciliation_guard()": { "enabled": true, @@ -1453,11 +1449,11 @@ }, "function:reporting_reconciliation_reference_immutable()": { "enabled": true, - "fingerprint": "56fa1c28e2461a3ca5d7a68f8356f6ffb7652c0418dfaa6fad87963c849a92c1" + "fingerprint": "79878862c9c677cf09565ccd0165628b218c8ef6ef0d0f1b6db4f83190b65175" }, "function:reporting_reconciliation_validate(r reporting_reconciliation_records)": { "enabled": true, - "fingerprint": "7f041091b87305a1f802f091e6caaa4cb2eb235f03e97324bbfe04a792485161" + "fingerprint": "b54f5cd5ca5ee8e47cd94fb3747fe88dfbe66103e139909fa255e3ccaca5627b" }, "function:reporting_sorted_strings(document jsonb)": { "enabled": true, diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_status_schema.json b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_status_schema.json index dc66de7db..fb6401f47 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_status_schema.json +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_status_schema.json @@ -679,10 +679,6 @@ "enabled": true, "fingerprint": "12b7a569e49f4bf374d5dade9ec0a0d812e4dea251366467c95bcd58bee90b9d" }, - "constraint:reporting_status_scope_checkpoints.reporting_status_scope_checkpoints_account_generation": { - "enabled": true, - "fingerprint": "8acc83717805dad2b34cd3943ff8ceea21b1a876685f4e3bca5a44e99f38a768" - }, "constraint:reporting_status_scope_checkpoints.reporting_status_scope_checkpoints_account_id_fkey": { "enabled": true, "fingerprint": "b5d2dccd1371a3885ddd1eea403915a5571b631c238a6e2bd65bba562bded98f" diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py index cf78d9a91..a9eb32e7d 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py @@ -293,11 +293,18 @@ async def _lock_scopes_on(self, connection: Any, snapshot: ReportingStatusSnapsh # typed scopes have exactly the same row-lock/notification boundary. for scope in projection_scopes(snapshot): await connection.execute( - f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, scope, fingerprint," # nosec B608 - " snapshot, source_sequence, baseline, publishable, selector_writer_floor)" - ' VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,\'{"health":"waiting"}\',0,FALSE,FALSE,2)' + f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, offering_id, scope," # nosec B608 + " fingerprint, snapshot, source_sequence, baseline, publishable," + " selector_writer_floor) VALUES (%s,%s,%s,%s,%s,%s,%s,%s::jsonb,%s," + ' \'{"health":"waiting"}\',0,FALSE,FALSE,2)' f" ON CONFLICT ({_KEY}) DO NOTHING", # nosec B608 - (*scope.checkpoint_key, json.dumps(ledger_asdict(scope)), "0" * 64), + ( + *scope.checkpoint_key, + # The scope's generation: '' for an account-level one. + (scope.generation_key.offering_id if scope.generation_key else None) or "", + json.dumps(ledger_asdict(scope)), + "0" * 64, + ), ) await ( await connection.execute( diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/pg.py index 5669742d4..275dff3d7 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/pg.py @@ -12,8 +12,10 @@ from importlib.resources import files from typing import TYPE_CHECKING, Any +from pydantic import TypeAdapter + from adcp.reporting.ledger._delivery_state import decode_record -from adcp.reporting.ledger._record_rows import _configuration_from_row +from adcp.reporting.ledger._record_rows import _CONFIGURATION_COLUMNS, _configuration_from_row from adcp.reporting.ledger.delivery_models import ( ReportingDeliveryPrincipal, ReportingDeliveryScope, @@ -21,6 +23,7 @@ ) from adcp.reporting.ledger.models import ( ReportingConfiguration, + ReportingConfigurationGenerationKey, ReportingDeliveryEscalation, ReportingObligationRecord, ) @@ -87,6 +90,36 @@ _ProducerSample = tuple[int, datetime | None, str, str, str, int] +#: A reporting subscription's admitted owed set of an activated account, with +#: its current generation. Every selected column is a fixed SDK identifier. +_OWED_SET_CONFIGURATION = ( + "SELECT s.producer_key, " # nosec B608 + + _CONFIGURATION_COLUMNS + + " FROM adcp_reporting_production_owed_sets s JOIN reporting_configurations c" + " USING(account_id,consumer_id,delivery_config_id,delivery_config_version,offering_id)" + " WHERE EXISTS(SELECT 1 FROM reporting_production_accounts a" + " WHERE a.account_id=s.account_id)" +) +_OWED_SET_KEY = ( + " account_id=%s AND consumer_id=%s AND delivery_config_id=%s" + " AND delivery_config_version=%s AND offering_id=%s" +) +_CONSTITUENTS: TypeAdapter[list[ReportingConstituent]] = TypeAdapter(list[ReportingConstituent]) + + +def _owed_identity(key: ReportingConfigurationGenerationKey) -> tuple[str, str, str, int, str]: + return ( + key.account_id, + key.consumer_id, + key.delivery_config_id, + key.delivery_config_version, + key.offering_id or "", + ) + + +def _constituents_document(constituents: Sequence[ReportingConstituent]) -> str: + return json.dumps([item.model_dump(mode="json") for item in constituents]) + class _ProductionConnection: """Only fixed SDK identifiers are selected; account/writer ordering stays intact.""" @@ -429,6 +462,180 @@ async def _enroll_on( if original is None or original[0] != document: raise ReportingNotificationError("reporting_production_destination_conflict") + # -- reporting-subscription owed sets (managed delivery) ------------------- + + def _status_covers_owed_sets(self) -> bool: + # Admission enrolls at most one offering per subscription version. + return True + + async def admit_subscription_generation( + self, + configuration: ReportingConfiguration, + binding: ReportingDestinationBinding, + *, + offering_id: str, + ) -> None: + """Install a reporting subscription's generation with its managed binding. + + An owed set has no generation-wide source denominator: each obligation + freezes its own constituents (``commit_subscription_obligation``). + Idempotent; a changed generation or binding is a conflict. + """ + offering = self._owner()._subscription_offering( + configuration, binding, offering_id=offering_id + ) + async with self._connection() as connection, connection.transaction(): + await self._lock_account(connection, configuration.account_id) + token = _ADMISSION_CONNECTION.set((id(self), asyncio.current_task(), connection)) + try: + await self.put_configuration(configuration) + await self.put_destination_binding(binding) + await self._enroll_owed_set_on( + connection, configuration, offering._producer_key, binding + ) + finally: + _ADMISSION_CONNECTION.reset(token) + + async def _enroll_owed_set_on( + self, + connection: Any, + configuration: ReportingConfiguration, + producer_key: str, + destination: ReportingDestinationBinding, + ) -> None: + owner = self._owner() + offering = owner._subscription_offering(configuration, destination) + document = owner._destination_binding(destination, offering).wire() + identity = _owed_identity(configuration.generation_key) + other = await ( + await connection.execute( + "SELECT 1 FROM adcp_reporting_production_owed_sets WHERE account_id=%s" + " AND consumer_id=%s AND delivery_config_id=%s AND delivery_config_version=%s" + " AND offering_id<>%s LIMIT 1", + identity, + ) + ).fetchone() + if other is not None: + # Status checkpoints hold one configuration scope per version. + raise ReportingNotificationError("reporting_production_offering_conflict") + await connection.execute( + "INSERT INTO adcp_reporting_production_owed_sets" + " (account_id,consumer_id,delivery_config_id,delivery_config_version,offering_id," + " producer_key,method) VALUES(%s,%s,%s,%s,%s,%s,%s::jsonb) ON CONFLICT DO NOTHING", + (*identity, producer_key, json.dumps(document)), + ) + row = await ( + await connection.execute( + "SELECT producer_key,method FROM adcp_reporting_production_owed_sets WHERE" + + _OWED_SET_KEY, + identity, + ) + ).fetchone() + if row is None or row[0] != producer_key: + raise ReportingNotificationError("reporting_production_source_conflict") + if row[1] != document: + raise ReportingNotificationError("reporting_production_destination_conflict") + + async def subscription_generation_admitted(self, configuration: ReportingConfiguration) -> bool: + """Whether this exact generation is already admitted (it is bound once).""" + async with self._connection() as connection: + row = await ( + await connection.execute( + "SELECT " + _CONFIGURATION_COLUMNS + " FROM adcp_reporting_production_owed_sets" + " JOIN reporting_configurations" + " USING(account_id,consumer_id,delivery_config_id,delivery_config_version," + "offering_id) WHERE" + _OWED_SET_KEY, + _owed_identity(configuration.generation_key), + ) + ).fetchone() + return row is not None and _configuration_from_row(row) == configuration + + async def _check_owed_set_on( + self, connection: Any, configuration: ReportingConfiguration + ) -> None: + """An admitted, current owed set of an activated account and a ready producer. + + Planning commits outside a producer turn: any ready offering's producer + qualifies; dispatch re-checks the host's pool within its own turn. + """ + owner = self._owner() + ready = {o._producer_key for o in owner.offerings if owner._offering_ready(o)} + row = await ( + await connection.execute( + _OWED_SET_CONFIGURATION + " AND" + _OWED_SET_KEY, + _owed_identity(configuration.generation_key), + ) + ).fetchone() + if ( + not configuration.owed_set + or row is None + or row[0] not in ready + or _configuration_from_row(row[1:]) != configuration + ): + raise LedgerConflictError("HISTORY_UNAVAILABLE", "producer generation is unavailable") + + async def commit_subscription_obligation( + self, + configuration: ReportingConfiguration, + obligation: ReportingObligationRecord, + constituents: tuple[ReportingConstituent, ...], + ) -> ReportingObligationRecord: + """Commit an owed obligation with the constituents it freezes, atomically.""" + if obligation.generation_key != configuration.generation_key or not constituents: + raise LedgerConflictError("HISTORY_UNAVAILABLE", "owed obligation is unbound") + async with self._connection() as connection, connection.transaction(): + await self._lock_account(connection, configuration.account_id) + await self._check_owed_set_on(connection, configuration) + token = _ADMISSION_CONNECTION.set((id(self), asyncio.current_task(), connection)) + try: + stored = await self.commit_obligation(obligation) + finally: + _ADMISSION_CONNECTION.reset(token) + await connection.execute( + "INSERT INTO adcp_reporting_production_owed_constituents" + " (account_id,reporting_obligation_id,constituents) VALUES(%s,%s,%s::jsonb)" + " ON CONFLICT DO NOTHING", + ( + stored.account_id, + stored.reporting_obligation_id, + _constituents_document(constituents), + ), + ) + if await self._owed_constituents_on(connection, stored) != constituents: + raise LedgerConflictError( + "HISTORY_UNAVAILABLE", "owed obligation constituents are immutable" + ) + return stored + + async def _owed_constituents_on( + self, connection: Any, obligation: ReportingObligationRecord + ) -> tuple[ReportingConstituent, ...] | None: + row = await ( + await connection.execute( + "SELECT constituents FROM adcp_reporting_production_owed_constituents" + " WHERE account_id=%s AND reporting_obligation_id=%s", + (obligation.account_id, obligation.reporting_obligation_id), + ) + ).fetchone() + return None if row is None else tuple(_CONSTITUENTS.validate_python(row[0])) + + async def production_owed_sets( + self, producer_keys: Sequence[str] + ) -> tuple[ReportingConfiguration, ...]: + """Admitted owed-set generations of these producers, in key order.""" + if not producer_keys: + return () + async with self._connection() as connection: + rows = await ( + await connection.execute( + _OWED_SET_CONFIGURATION + " AND s.producer_key=ANY(%s) AND NOT c.quarantined" + " ORDER BY account_id,consumer_id,delivery_config_id,delivery_config_version," + "offering_id", + (list(producer_keys),), + ) + ).fetchall() + return tuple(_configuration_from_row(row[1:]) for row in rows) + async def lease_period_close( self, *, worker_id: str, now: datetime, lease_seconds: float ) -> LeasedConfiguration | None: @@ -659,6 +866,21 @@ async def _source_connection( async def producer_constituents( self, configuration: ReportingConfiguration, obligation: ReportingObligationRecord ) -> tuple[ReportingConstituent, ...]: + if configuration.owed_set: + # Each owed obligation froze its own constituents at planning. + async with self._connection() as connection, connection.transaction(): + await self._lock_account(connection, configuration.account_id) + await self._check_owed_set_on(connection, configuration) + frozen = ( + await self._owed_constituents_on(connection, obligation) + if obligation.generation_key == configuration.generation_key + else None + ) + if frozen is None: + raise LedgerConflictError( + "HISTORY_UNAVAILABLE", "owed obligation constituents are unavailable" + ) + return frozen async with self._source_connection(configuration) as (connection, identity): if obligation.generation_key != configuration.generation_key or set( obligation.media_buy_ids @@ -868,6 +1090,16 @@ async def activate_production(self, *, account_id: str) -> bool: if binding.generation_key not in configurations: continue configuration = configurations[binding.generation_key] + if configuration.owed_set: + try: + offering = owner._subscription_offering(configuration, binding) + # Unsupported or unavailable bindings remain unadmitted. + except Exception: # nosec B112 + continue + await self._enroll_owed_set_on( + connection, configuration, offering._producer_key, binding + ) + continue try: offering = owner._configuration_offering(configuration, binding) # Unsupported or unavailable bindings remain unadmitted. @@ -880,7 +1112,37 @@ async def _materializer_context_on( self, connection: Any, scope: ReportingDeliveryScope ) -> MaterializerContext: context = await super()._materializer_context_on(connection, scope) - if isinstance(connection, _ProductionConnection): + if isinstance(connection, _ProductionConnection) and scope.generation_key.offering_id: + # A reporting subscription's owed set: no generation-wide source binding. + row = await ( + await connection.execute( + "SELECT a.policy,s.producer_key,NULL::jsonb,s.method" + " FROM reporting_production_accounts a" + " LEFT JOIN adcp_reporting_production_owed_sets s" + " ON s.account_id=a.account_id AND s.consumer_id=%s AND s.delivery_config_id=%s" + " AND s.delivery_config_version=%s AND s.offering_id=%s" + " WHERE a.account_id=%s", + ( + scope.consumer_id, + scope.generation_key.delivery_config_id, + scope.generation_key.delivery_config_version, + scope.generation_key.offering_id, + scope.principal.account_id, + ), + ) + ).fetchone() + if row is None: + raise ReportingNotificationError("reporting_production_activation_required") + owner = self._owner() + owner._check_context( + context, + key_for(context.binding, context.obligation, owner.keys), + row[0], + row[1], + row[2], + row[3], + ) + elif isinstance(connection, _ProductionConnection): owner = self._owner() row = await ( await connection.execute( diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/required_schema.json b/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/required_schema.json index a731ca5be..b15752621 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/required_schema.json +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/production/required_schema.json @@ -1061,7 +1061,7 @@ }, "function:reporting_production_source_dirty()": { "enabled": true, - "fingerprint": "e63cafa384ff68fb1de7f2daaf538cd936018f994ecdc7a27b36c395c5677f5a" + "fingerprint": "bbdb62b36233bc7d388404ea4c34c58bf4820b0c535a7b8ae52077e3ad0b1db2" }, "function:reporting_production_source_progress_guard()": { "enabled": true, diff --git a/src/adcp/reporting/ledger/_delivery_state.py b/src/adcp/reporting/ledger/_delivery_state.py index b9d213881..6876e18c4 100644 --- a/src/adcp/reporting/ledger/_delivery_state.py +++ b/src/adcp/reporting/ledger/_delivery_state.py @@ -192,6 +192,8 @@ def storage_identity(record: ReportingDeliveryRecord) -> tuple[str | int | None, receipt_chain(receipt) if receipt is not None else None, receipt.status if receipt is not None else None, receipt.supersedes_reporting_receipt_id if receipt is not None else None, + # '' for an account-level generation, as stored. + generation.offering_id or "", ) diff --git a/src/adcp/reporting/production/memory.py b/src/adcp/reporting/production/memory.py index 3fba83025..187c60309 100644 --- a/src/adcp/reporting/production/memory.py +++ b/src/adcp/reporting/production/memory.py @@ -5,7 +5,7 @@ import json import weakref from collections.abc import Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any @@ -281,6 +281,13 @@ def _enroll_owed_set( if previous != producer_key: raise ReportingNotificationError("reporting_production_source_conflict") offering = owner._subscription_offering(configuration, destination) + key = configuration.generation_key + if any( + other != key and replace(other, offering_id=key.offering_id) == key + for other in self._production_owed_sets + ): + # Status checkpoints hold one configuration scope per version. + raise ReportingNotificationError("reporting_production_offering_conflict") document = canonical_json_utf8_v1(owner._destination_binding(destination, offering).wire()) original = self._production_destination_bindings.setdefault( (configuration.generation_key, destination.consumer_id), document diff --git a/tests/conformance/reporting/test_reporting_feed_installed_pg.py b/tests/conformance/reporting/test_reporting_feed_installed_pg.py index 0a7397d46..13c208ed9 100644 --- a/tests/conformance/reporting/test_reporting_feed_installed_pg.py +++ b/tests/conformance/reporting/test_reporting_feed_installed_pg.py @@ -110,7 +110,7 @@ async def test_installed_python310_cold_continuation_freezes_mutable_history_and ) assert ready["result"] == { "installed": True, - "materializer_objects": 189, + "materializer_objects": 188, "receipt_objects": 102, "feed_objects": 33, } diff --git a/tests/conformance/reporting/test_reporting_feed_migration.py b/tests/conformance/reporting/test_reporting_feed_migration.py index 465a30c4f..a79f9939b 100644 --- a/tests/conformance/reporting/test_reporting_feed_migration.py +++ b/tests/conformance/reporting/test_reporting_feed_migration.py @@ -80,7 +80,7 @@ async def test_feed_migration_preserves_parent_catalog_receipts_pending_and_fair .read_text() ) ) - assert len(parent_manifest) == 774 + assert len(parent_manifest) == 772 assert {key: original[key] for key in parent_manifest} == parent_manifest assert original == {**parent_manifest, **PROVISIONAL_OBJECTS} new = PgReportingFeedStore(pool=pool, notifications=notifications) @@ -96,7 +96,7 @@ async def test_feed_migration_preserves_parent_catalog_receipts_pending_and_fair assert actual == {**original, **MANIFEST} assert {k: v for k, v in actual.items() if "reporting_feed_" in k} == MANIFEST assert len(MANIFEST) == 33 - for package, count in (("materializer", 189), ("receipts", 102)): + for package, count in (("materializer", 188), ("receipts", 102)): required = json.loads( files("adcp_reporting_storage." + package) .joinpath("required_schema.json") diff --git a/tests/conformance/reporting/test_reporting_provisional_catalog.py b/tests/conformance/reporting/test_reporting_provisional_catalog.py index df834f357..735c7a5f1 100644 --- a/tests/conformance/reporting/test_reporting_provisional_catalog.py +++ b/tests/conformance/reporting/test_reporting_provisional_catalog.py @@ -56,7 +56,7 @@ async def test_atomic_observation_bootstrap_preserves_exact_767_parent_objects( .read_text() ) ) - assert len(expected_parent) == 774 + assert len(expected_parent) == 772 async with isolated_reporting_pool(autocommit=autocommit) as pool: async with pool.connection() as connection, connection.transaction(): for name in chain: diff --git a/tests/conformance/reporting/test_reporting_provisional_catalog_scope.py b/tests/conformance/reporting/test_reporting_provisional_catalog_scope.py index 4c0d264cd..70df22cb1 100644 --- a/tests/conformance/reporting/test_reporting_provisional_catalog_scope.py +++ b/tests/conformance/reporting/test_reporting_provisional_catalog_scope.py @@ -68,14 +68,14 @@ async def test_exact_catalog_scope_preserves_full_contract_and_excludes_overload .read_text() ) ) - assert len(expected) == 801 and len(PROVISIONAL_REQUIRED_OBJECTS) == 27 + assert len(expected) == 799 and len(PROVISIONAL_REQUIRED_OBJECTS) == 27 async with isolated_reporting_pool(autocommit=autocommit) as pool: await PgReportingReceiptStore(pool=pool).create_schema() async with pool.connection() as connection: full = CatalogProbe(connection) assert await schema_objects(full) == expected assert len(full.calls) == 6 - assert sum(len(rows) for _, _, rows in full.calls) == 801 + assert sum(len(rows) for _, _, rows in full.calls) == 799 await connection.execute( "CREATE TABLE reporting_unrelated_catalog_scope (value integer);" " CREATE FUNCTION reporting_provisional_immutable(integer) RETURNS integer" diff --git a/tests/conformance/reporting/test_reporting_receipt_migration.py b/tests/conformance/reporting/test_reporting_receipt_migration.py index f9f88e09c..add02cfdf 100644 --- a/tests/conformance/reporting/test_reporting_receipt_migration.py +++ b/tests/conformance/reporting/test_reporting_receipt_migration.py @@ -51,7 +51,7 @@ async def test_populated_repeated_concurrent_migration_preserves_parent_objects_ async with pool.connection() as c: actual = await schema_objects(c) assert {key: actual[key] for key in original} == original - assert len(PARENT_MANIFEST) == 189 + assert len(PARENT_MANIFEST) == 188 assert {key: actual[key] for key in PARENT_MANIFEST} == PARENT_MANIFEST assert { key: value for key, value in actual.items() if "reporting_receipt_ingestion_" in key diff --git a/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py b/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py index 7a3c1fa6f..5b790b9cf 100644 --- a/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py +++ b/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py @@ -197,3 +197,46 @@ async def test_status_notifications_refuse_subscription_generations() -> None: with pytest.raises(LedgerConflictError) as refused: await _plan(world, AGENCY, store, "2026-10-04T00:00:00Z") assert refused.value.code == "UNSUPPORTED_FEATURE" + + +async def test_an_account_only_managed_delivery_schema_is_upgraded() -> None: + """A schema whose managed-delivery tables still refuse subscription generations.""" + from adcp_reporting_storage.production.pg import PgReportingProductionStore + + async with isolated_reporting_pool(autocommit=True) as pool: + store = PgReportingProductionStore(pool=pool, notifications=True) + await store.create_schema() + async with pool.connection() as connection: + for table in ( + "reporting_reconciliation_records", + "reporting_materializer_discovery", + "reporting_status_scope_checkpoints", + ): + await connection.execute( + f"ALTER TABLE {table} ADD CONSTRAINT {table}_account_generation" # nosec B608 + " CHECK (offering_id = '')" + ) + await connection.execute( + "ALTER TABLE reporting_materializer_discovery" + " DROP CONSTRAINT reporting_materializer_discovery_pkey," + " ADD CONSTRAINT reporting_materializer_discovery_pkey PRIMARY KEY" + " (account_id, consumer_id, delivery_config_id, delivery_config_version)" + ) + for _ in range(2): # idempotent + await store.create_schema() + async with pool.connection() as connection: + checks = await ( + await connection.execute( + "SELECT conname FROM pg_constraint WHERE conname LIKE '%account_generation'" + " ORDER BY conname" + ) + ).fetchall() + primary = await ( + await connection.execute( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint" + " WHERE conname = 'reporting_materializer_discovery_pkey'" + ) + ).fetchone() + # Production generations still serve account-level generations only. + assert checks == [("reporting_production_generations_account_generation",)] + assert primary[0].endswith("delivery_config_version, offering_id)") diff --git a/tests/conformance/reporting/test_reporting_subscription_production.py b/tests/conformance/reporting/test_reporting_subscription_production.py index 31eb38fed..39a523e1f 100644 --- a/tests/conformance/reporting/test_reporting_subscription_production.py +++ b/tests/conformance/reporting/test_reporting_subscription_production.py @@ -1,10 +1,11 @@ -"""Reporting subscriptions delivered through production managed delivery (memory store). +"""Reporting subscriptions delivered through production managed delivery. Mirrors adcontextprotocol/adcp-client#3523: the service binds each managed subscription generation to its destination before its first obligation, freezes each obligation's constituents from the host's authorized pool, produces and materializes owed obligations, stamps ``account_id`` on rows, and holds a -delivery when the owner's authority is lost. +delivery when the owner's authority is lost. Each test runs against the memory +store and, with ``ADCP_PG_TEST_URL``, the PostgreSQL production store. """ from __future__ import annotations @@ -12,7 +13,7 @@ import hashlib import json import sqlite3 -from contextlib import asynccontextmanager +from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import replace from datetime import datetime, timedelta from importlib.resources import files @@ -25,6 +26,7 @@ from adcp.reporting.ledger import ( LedgerConflictError, ProducerOfferings, + ReportingConfigurationGenerationKey, ReportingDestinationBinding, ReportingStatusCaller, ReportingStatusHandler, @@ -56,11 +58,13 @@ from adcp.types import ReportingDeliveryOffering from tests.test_reporting_subscription_engine import Clock, Host -from ._generation_support import END, START +from ._generation_support import END, START, isolated_reporting_pool from ._materializer_support import reference_rows from ._production_support import Source, SQLiteDestination from .test_reliable_reporting_production_admission import Application +pytestmark = pytest.mark.parametrize("backend", ["memory", "postgres"]) + ACCOUNT = "acct-subscriptions" AGENCY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") OFFERING_ID = "managed-adapter" @@ -130,9 +134,34 @@ async def get_adcp_capabilities(self, params, context=None): return capabilities_response(["media_buy"], sandbox=True, idempotency={"supported": False}) +async def _store(stack: AsyncExitStack, backend: str, *, notifications: bool) -> Any: + if backend == "memory": + return None + from adcp_reporting_storage.production.pg import PgReportingProductionStore + + pool = await stack.enter_async_context(isolated_reporting_pool(autocommit=True)) + store = PgReportingProductionStore(pool=pool, notifications=notifications) + await store.create_schema() + return store + + +@asynccontextmanager +async def _seller( + tmp_path, + backend: str, + *, + pool_buys: tuple[str, ...] = ("mb-1",), + notifications: bool = False, +): + async with AsyncExitStack() as stack: + clock = DeterministicReportingClock(END) + store = await _store(stack, backend, notifications=notifications) + async with _composed(tmp_path, clock, store, pool_buys, notifications) as h: + yield h + + @asynccontextmanager -async def _seller(tmp_path, *, pool_buys: tuple[str, ...] = ("mb-1",)): - clock = DeterministicReportingClock(END) +async def _composed(tmp_path, clock, store, pool_buys, notifications): verifier = _subscription_verifier() key = verifier.key writer = SQLiteDestination(tmp_path / "destination.sqlite", key) @@ -273,22 +302,34 @@ async def authorize(account, context, consumer): AccountCapabilities(supported_billing=["operator"], require_operator_auth=True), ), authorize, + notifications=notifications, poll_seconds=60, read_jitter_window=timedelta(0), ) failures: list[Any] = [] h.failures = failures - h.service = ReliableReportingService.memory( - account_context=context, - clock=clock, - production=options, - reporting_subscriptions=ReportingSubscriptionServiceOptions( - engine=engine, - list_principals=lambda: [AGENCY], - resolve_principal=lambda _request, _context: AGENCY, - bind_managed_generation=bind, - planning_error_handler=failures.append, - ), + subscriptions = ReportingSubscriptionServiceOptions( + engine=engine, + list_principals=lambda: [AGENCY], + resolve_principal=lambda _request, _context: AGENCY, + bind_managed_generation=bind, + planning_error_handler=failures.append, + ) + h.service = ( + ReliableReportingService.memory( + account_context=context, + clock=clock, + production=options, + reporting_subscriptions=subscriptions, + ) + if store is None + else ReliableReportingService( + store=store, + account_context=context, + clock=clock, + production=options, + reporting_subscriptions=subscriptions, + ) ) h.service.sources.register( "adapter", @@ -299,6 +340,7 @@ async def authorize(account, context, consumer): h.mount = create_mcp_server(handler) # production requires a mounted handler h.production = h.service._production h.store = h.service.store + h.pool = getattr(store, "_pool", None) try: yield h finally: @@ -351,13 +393,29 @@ def _buy(media_buy_id: str) -> SubscriptionBuy: ) +async def _advance(h, delta: timedelta) -> None: + """Advance the service clock; PostgreSQL schedules held work on its own clock.""" + h.clock.advance(delta) + if h.pool is not None: + async with h.pool.connection() as connection: + await connection.execute( + "UPDATE reporting_production_work SET due_at=due_at-%s WHERE state='pending'", + (delta,), + ) + await connection.execute( + "UPDATE reporting_materializer_accounts SET due_at=due_at-%s" + " WHERE due_at IS NOT NULL", + (delta,), + ) + + def _delivered(h) -> list[dict[str, Any]]: with sqlite3.connect(h.writer.path) as c: return [json.loads(row[0]) for row in c.execute("SELECT content FROM artifacts")] -async def test_owed_obligations_are_bound_produced_and_delivered(tmp_path) -> None: - async with _seller(tmp_path) as h: +async def test_owed_obligations_are_bound_produced_and_delivered(tmp_path, backend) -> None: + async with _seller(tmp_path, backend) as h: await _subscribe(h, [_buy("mb-1")]) await h.service.start() # one full turn: plan, produce, materialize once assert h.failures == [] @@ -408,8 +466,8 @@ async def test_owed_obligations_are_bound_produced_and_delivered(tmp_path) -> No assert len(h.bound) == 1 -async def test_a_buy_outside_the_authorized_pool_writes_nothing(tmp_path) -> None: - async with _seller(tmp_path, pool_buys=("mb-other",)) as h: +async def test_a_buy_outside_the_authorized_pool_writes_nothing(tmp_path, backend) -> None: + async with _seller(tmp_path, backend, pool_buys=("mb-other",)) as h: await _subscribe(h, [_buy("mb-1")]) await h.service.start() assert [failure.subscription_id for failure in h.failures] == ["analytics-hourly"] @@ -417,8 +475,8 @@ async def test_a_buy_outside_the_authorized_pool_writes_nothing(tmp_path) -> Non assert await h.store.list_generation_obligations(key) == () -async def test_a_lost_grant_holds_delivery(tmp_path) -> None: - async with _seller(tmp_path) as h: +async def test_a_lost_grant_holds_delivery(tmp_path, backend) -> None: + async with _seller(tmp_path, backend) as h: await _subscribe(h, [_buy("mb-1")]) await h.service.start() # the first turn delivers one of the two obligations assert len(_delivered(h)) == 1 @@ -440,6 +498,51 @@ async def test_a_lost_grant_holds_delivery(tmp_path) -> None: ACCOUNT, timezone="UTC", authorized_at=at("2026-08-01T00:00:00Z") ) await h.engine.reconcile(AGENCY) - h.clock.advance(timedelta(minutes=2)) + await _advance(h, timedelta(minutes=2)) assert (await h.production.materializer.run_once()).state == "verified" assert len(_delivered(h)) == 2 + + +async def test_owed_obligations_are_delivered_with_status_notifications(tmp_path, backend) -> None: + async with _seller(tmp_path, backend, notifications=True) as h: + await _subscribe(h, [_buy("mb-1")]) + await h.service.start() + assert h.failures == [] + while (await h.production.materializer.run_once()).state == "verified": + pass + assert len(_delivered(h)) == 2 + payload = await ReportingStatusHandler(h.store).handle( + {"view": "periods"}, caller=ReportingStatusCaller(ACCOUNT, AGENCY.subject) + ) + assert {p["offering_id"] for p in payload["periods"]} == {OFFERING_ID} + if h.pool is not None: + # Status checkpoints record the subscription generation's offering. + async with h.pool.connection() as connection: + offerings = await ( + await connection.execute( + "SELECT DISTINCT offering_id FROM reporting_status_scope_checkpoints" + ) + ).fetchall() + assert offerings == [(OFFERING_ID,)] + + +async def test_a_second_offering_of_one_version_is_refused(tmp_path, backend) -> None: + async with _seller(tmp_path, backend) as h: + await _subscribe(h, [_buy("mb-1")]) + # Another offering of the same subscription version is already enrolled. + other = (ACCOUNT, AGENCY.subject, "analytics-hourly", 1, "other-offering") + if h.pool is None: + h.store._production_owed_sets.add(ReportingConfigurationGenerationKey(*other)) + else: + async with h.pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_production_owed_sets VALUES" + " (%s,%s,%s,%s,%s,%s,'{}'::jsonb)", + (*other, "0" * 64), + ) + await h.service.start() + # Status checkpoints hold one configuration scope per subscription version. + assert [failure.subscription_id for failure in h.failures] == ["analytics-hourly"] + assert h.failures[0].error.code == "reporting_production_offering_conflict" + (key,) = h.bound + assert await h.store.list_generation_obligations(key) == ()