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) == ()