From 9494c95b56bf93b21a28fbf2b95b97b0d1b4d728 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:33:15 +0000 Subject: [PATCH 1/2] fix(reporting): key status checkpoints by resolved offering A reporting subscription keeps one configuration generation per resolved offering of a version, but status checkpoints were keyed per subscription version: (account, consumer, delivery_config_id, version, scope kind, obligation). Two offerings of one version for one account and owner shared a configuration checkpoint. The memory store silently merged the second into the first; the PostgreSQL store refused owed sets whenever status notifications were on, and would otherwise have dropped the second scope on ON CONFLICT DO NOTHING. The production stores refused a second offering with reporting_production_offering_conflict for the same reason. ReportingStatusScope.checkpoint_key now ends with the generation's offering_id ('' for an account-level generation). checkpoint_digest feeds the status cause_id and the projection history keys; it digests the six original columns when the offering is '', so every id and digest that deployments already store is unchanged. The memory store re-keys an old six-column state image. PostgreSQL: reporting_status_checkpoint_offerings.sql widens the checkpoint primary key to include offering_id, adds offering_id to the status and projection status event tables (filled from their checkpoint), re-points their checkpoint references and per-generation unique keys at the widened key under their existing names, and defines the checkpoint and event guards, which now compare the offering. Migration 0006 includes it; it is a stopped-worker upgrade, and a no-op without status notifications. The status store, projection and production create_schema chains run it too, and the status and projection manifests are regenerated. The refusals that existed only because of the shared checkpoint are lifted. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 20 +- docs/reporting-ledger-migration.md | 24 ++ docs/reporting-subscriptions-seller.md | 2 +- .../src/adcp_reporting_storage/ledger/pg.py | 17 +- .../reporting_projection_notifications.sql | 40 +-- .../reporting_status_checkpoint_offerings.sql | 269 ++++++++++++++++++ .../ledger/reporting_status_notifications.sql | 74 +---- .../src/adcp_reporting_storage/migration.py | 1 + ..._reporting_status_checkpoint_offerings.sql | 15 + .../outbox/required_status_schema.json | 26 +- .../outbox/status_pg.py | 29 +- .../adcp_reporting_storage/production/pg.py | 12 +- .../adcp_reporting_storage/projection/pg.py | 3 +- .../projection/required_schema.json | 16 +- .../src/adcp_reporting_storage/storage.py | 8 +- .../reporting/ledger/notification_models.py | 33 ++- src/adcp/reporting/outbox/status_memory.py | 9 +- src/adcp/reporting/production/memory.py | 9 +- src/adcp/reporting/projection/history.py | 6 +- .../reporting/_empty_ledger_support.py | 1 + .../reporting/_production_packaging.py | 1 + ...st_reporting_status_projection_contract.py | 64 ++++- .../test_reporting_status_transactions.py | 3 +- .../test_reporting_storage_migrations_pg.py | 6 +- .../test_reporting_subscription_ledger_pg.py | 152 +++++++++- .../test_reporting_subscription_production.py | 13 +- tests/test_reporting_storage_facade.py | 11 +- 27 files changed, 655 insertions(+), 209 deletions(-) create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0006_reporting_status_checkpoint_offerings.sql diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index e0b7f8d80..63a8d460a 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -413,8 +413,11 @@ platform = reporting.install(MySeller()) # then serve the platform as usual never changes the schema: it refuses one that is behind with `REPORTING_SCHEMA_BEHIND`, naming the command and the pending versions. - `0002 reporting_subscription_generations` widens the generation key and - is a stopped-worker upgrade: stop every reporting worker before running - it. `--status` flags it. + `0006 reporting_status_checkpoint_offerings` keys status checkpoints and + status events by the generation's resolved offering. Both are + stopped-worker upgrades: stop every reporting worker before running them. + `--status` flags them. Stores whose schema you install with + `create_schema()` apply `0006`'s change there too. - **Principals.** Reporting APIs name a principal with `PrincipalRef(tenant_id, principal_id)`; `grants_task` and `authorizes_delivery` take it. One service serves one tenant. @@ -632,18 +635,13 @@ Supply `bind_managed_generation(configuration, version)`. 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. + Status checkpoints, and their status events, are keyed by the offering, so + several offerings of one subscription version for the same account and + owner each keep their own status (migration `0006`). **Limits.** These are not served yet: -- subscriptions that resolve a receipt-reconciled (`consumer_receipt`) offering; -- 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). +- subscriptions that resolve a receipt-reconciled (`consumer_receipt`) offering. 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 40b422863..776b51791 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -86,6 +86,30 @@ 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 idempotent; a fresh schema is created in the widened layout directly. +### Status checkpoints keyed by offering (migration `0006`) + +Status checkpoints, and the status events that reference them, are keyed by +the generation's resolved offering too, so several offerings of one +subscription version for the same account and owner each keep their own +status: `reporting_status_scope_checkpoints` is keyed by `(account_id, +consumer_namespace, delivery_config_id, version, scope_kind, +obligation_namespace, offering_id)`. Account-level checkpoints keep +`offering_id` `''`, so their keys, status `cause_id` values and projection +history keys are unchanged. + +This is also a **stopped-worker upgrade**: stop every reporting worker, run +`python -m adcp migrate` (migration `0006 reporting_status_checkpoint_offerings`) +or the status, projection or production store's `create_schema()` once with the +new release, then restart the workers. Under the schema lock it widens the +checkpoint primary key; the status event tables (`reporting_status_notification_events`, +and `reporting_projection_notification_events` where the projection is +installed) gain `offering_id`, filled from their checkpoint, and their +checkpoint reference and per-generation unique key include it. Constraints +keep their names. It changes nothing on a schema without status +notifications. A foreign key outside the SDK that references +`reporting_status_scope_checkpoints` refuses the upgrade instead of being +dropped. + ## Historical account-generation upgrade The following notes describe the pre-ownership SDK release and are retained for diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index b440aa2e9..4d6f4c00d 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -87,7 +87,7 @@ account.authorization = wind_down_grant(account.authorization) Pass one storage object: `memory_storage()` in development, `postgres_storage(dsn=... | pool=...)` in production. Process-local storage is refused outside development (`REPORTING_STORAGE_NOT_DURABLE`). -Apply schema changes with `python -m adcp migrate [--status | --dry-run]` before starting a release. Startup refuses a schema that is behind (`REPORTING_SCHEMA_BEHIND`). `0002 reporting_subscription_generations` is a stopped-worker upgrade: stop every reporting worker first. +Apply schema changes with `python -m adcp migrate [--status | --dry-run]` before starting a release. Startup refuses a schema that is behind (`REPORTING_SCHEMA_BEHIND`). `0002 reporting_subscription_generations` and `0006 reporting_status_checkpoint_offerings` are stopped-worker upgrades: stop every reporting worker first. ## Workers and scaling 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 65408316c..95be25a3d 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 @@ -922,30 +922,15 @@ async def _lock_account(connection: Any, account_id: str) -> None: await connection.execute( "SELECT 1 FROM reporting_status_scope_checkpoints WHERE account_id=%s" " ORDER BY account_id, consumer_namespace, delivery_config_id, version," - " scope_kind, obligation_namespace FOR UPDATE", + " scope_kind, obligation_namespace, offering_id FOR UPDATE", (account_id,), ) ).fetchall() # -- 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 not self._status_covers_owed_sets(): - raise LedgerConflictError( - "UNSUPPORTED_FEATURE", - "status notifications do not yet cover reporting-subscription generations", - ) key = configuration.generation_key payload = _configuration_payload(configuration) digest = _fingerprint(payload) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_projection_notifications.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_projection_notifications.sql index 138bb6a71..d1e84bdea 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_projection_notifications.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_projection_notifications.sql @@ -21,12 +21,18 @@ BEGIN admission_epoch BIGINT NOT NULL DEFAULT 2 CHECK (admission_epoch=2), fired_at TIMESTAMPTZ NOT NULL, snapshot JSONB NOT NULL, + -- Its checkpoint's offering: '' for an account-level generation. The + -- checkpoint key includes it (reporting_status_checkpoint_offerings.sql, + -- which runs first and upgrades a table created on the six-column key). + offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', PRIMARY KEY (account_id, consumer_namespace, notification_id), UNIQUE (account_id, consumer_namespace, delivery_config_id, version, - scope_kind, obligation_namespace, cause_generation), + scope_kind, obligation_namespace, offering_id, cause_generation), FOREIGN KEY (account_id, consumer_namespace, delivery_config_id, version, - scope_kind, obligation_namespace) + scope_kind, obligation_namespace, offering_id) REFERENCES reporting_status_scope_checkpoints, + CONSTRAINT reporting_projection_notification_events_offering CHECK + (coalesce(snapshot #>> '{cause,scope,generation_key,offering_id}', '') = offering_id), CHECK ((scope_kind = 'configuration') = (obligation_namespace = '')), UNIQUE (account_id, consumer_namespace, notification_type, cause_kind, cause_id, cause_generation), @@ -251,34 +257,8 @@ BEGIN ON reporting_projection_webhook_attempts FOR EACH ROW EXECUTE FUNCTION reporting_projection_webhook_attempt_guard(); END IF; - CREATE OR REPLACE FUNCTION reporting_projection_event_guard() - RETURNS TRIGGER LANGUAGE plpgsql AS $event_guard$ - DECLARE checkpoint reporting_status_scope_checkpoints; ids jsonb; - BEGIN - IF NOT EXISTS (SELECT 1 FROM reporting_projection_accounts - WHERE account_id=NEW.account_id AND current_input IS NOT NULL - AND notifications_enabled) THEN - RAISE EXCEPTION 'status_projection_activation_required' USING ERRCODE='23514'; - END IF; - SELECT * INTO checkpoint FROM reporting_status_scope_checkpoints - WHERE account_id=NEW.account_id AND consumer_namespace=NEW.consumer_namespace - AND delivery_config_id=NEW.delivery_config_id AND version=NEW.version - AND scope_kind=NEW.scope_kind AND obligation_namespace=NEW.obligation_namespace; - IF NOT FOUND OR NOT checkpoint.initialized OR NOT checkpoint.publishable - OR checkpoint.generation<>NEW.cause_generation OR checkpoint.fingerprint<>NEW.fingerprint - OR checkpoint.snapshot->>'health' IS DISTINCT FROM NEW.snapshot #>> '{cause,health}' THEN - RAISE EXCEPTION 'invalid status event checkpoint' USING ERRCODE='23514'; - END IF; - SELECT coalesce(jsonb_agg(issue_id ORDER BY issue_id COLLATE "C"), '[]') INTO ids - FROM (SELECT DISTINCT i->>'issue_id' COLLATE "C" AS issue_id - FROM jsonb_array_elements(checkpoint.snapshot->'issues') i - ORDER BY issue_id LIMIT 16) selected; - IF ids IS DISTINCT FROM NEW.snapshot #> '{cause,issue_ids}' THEN - RAISE EXCEPTION 'invalid status event issues' USING ERRCODE='23514'; - END IF; - RETURN NEW; - END - $event_guard$; + -- reporting_status_checkpoint_offerings.sql defines the guard: it compares + -- the full checkpoint key, offering included. IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_projection_notification_events'::regclass AND tgname='reporting_projection_event_guard') THEN diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql new file mode 100644 index 000000000..3bad3c22a --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql @@ -0,0 +1,269 @@ +-- Key status checkpoints, and the status events that reference them, by the +-- generation's resolved offering. +-- +-- A reporting subscription keeps one generation per resolved offering of a +-- version (adcp-client-python#1729), so two offerings of one subscription +-- version, for one account and one owner, must not share a checkpoint. The +-- checkpoint key is (account_id, consumer_namespace, delivery_config_id, +-- version, scope_kind, obligation_namespace, offering_id), with offering_id '' +-- for every account-level generation, so existing rows keep their key and +-- their uniqueness. Each status event table carries its checkpoint's offering, +-- references the widened key and is unique per checkpoint generation under it. +-- +-- Idempotent, and a no-op where status notifications are not installed. It +-- runs after reporting_status_selector_version.sql (and reporting_projection.sql +-- where the projection is installed) and before +-- reporting_projection_notifications.sql, which creates its event table on the +-- widened key. reporting_status_notifications.sql creates the checkpoint and +-- event tables on their published six-column key; this file widens them, an +-- earlier release's projection event table too, and defines the checkpoint and +-- event guards, which compare the offering. Constraints keep their names. +-- +-- A schema still on the six-column key is widened under the schema lock. This +-- is a maintenance upgrade: stop every reporting worker first, because workers +-- of earlier releases write the six-column key. A foreign key onto the +-- checkpoint key that the SDK does not own refuses the upgrade instead of +-- being dropped. + +DO $checkpoint_offerings$ +DECLARE + key_columns TEXT[]; + reference RECORD; + events TEXT; +BEGIN + PERFORM pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema())); + IF to_regclass('reporting_status_scope_checkpoints') IS NULL THEN + RETURN; + END IF; + SELECT ARRAY( + SELECT attribute.attname::TEXT + FROM unnest(pk.conkey) WITH ORDINALITY AS key_column(attnum, position) + JOIN pg_attribute attribute + ON attribute.attrelid = pk.conrelid AND attribute.attnum = key_column.attnum + ORDER BY key_column.position + ) + INTO key_columns + FROM pg_constraint pk + WHERE pk.conrelid = 'reporting_status_scope_checkpoints'::regclass AND pk.contype = 'p'; + + IF key_columns IS DISTINCT FROM ARRAY[ + 'account_id', 'consumer_namespace', 'delivery_config_id', 'version', + 'scope_kind', 'obligation_namespace', 'offering_id'] THEN + IF key_columns IS DISTINCT FROM ARRAY[ + 'account_id', 'consumer_namespace', 'delivery_config_id', 'version', + 'scope_kind', 'obligation_namespace'] THEN + RAISE EXCEPTION 'Keying status checkpoints by offering requires a stopped-worker maintenance migration' + USING HINT = 'reporting_status_scope_checkpoints has an unexpected primary key.'; + END IF; + LOCK TABLE reporting_status_scope_checkpoints IN ACCESS EXCLUSIVE MODE; + -- Only the SDK's own status event tables reference the key; each is + -- re-pointed at the widened key below. + FOR reference IN + SELECT fk.conrelid::regclass::TEXT AS table_name, fk.conname::TEXT AS constraint_name + FROM pg_constraint fk + WHERE fk.contype = 'f' AND fk.confrelid = 'reporting_status_scope_checkpoints'::regclass + LOOP + IF reference.table_name NOT IN ( + 'reporting_status_notification_events', + 'reporting_projection_notification_events' + ) THEN + RAISE EXCEPTION 'Keying status checkpoints by offering requires a stopped-worker maintenance migration' + USING HINT = 'A foreign key outside the SDK references reporting_status_scope_checkpoints; re-point it at the widened key first.'; + END IF; + EXECUTE format('LOCK TABLE %I IN ACCESS EXCLUSIVE MODE', reference.table_name); + EXECUTE format( + 'ALTER TABLE %I DROP CONSTRAINT %I', reference.table_name, reference.constraint_name + ); + END LOOP; + ALTER TABLE reporting_status_scope_checkpoints + DROP CONSTRAINT reporting_status_scope_checkpoints_pkey, + ADD CONSTRAINT reporting_status_scope_checkpoints_pkey PRIMARY KEY + (account_id, consumer_namespace, delivery_config_id, version, + scope_kind, obligation_namespace, offering_id); + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'reporting_status_scope_checkpoints'::regclass + AND conname = 'reporting_status_scope_checkpoints_offering') THEN + ALTER TABLE reporting_status_scope_checkpoints + ADD CONSTRAINT reporting_status_scope_checkpoints_offering + CHECK (coalesce(scope #>> '{generation_key,offering_id}', '') = offering_id); + END IF; + + FOREACH events IN ARRAY ARRAY[ + 'reporting_status_notification_events', 'reporting_projection_notification_events'] LOOP + CONTINUE WHEN to_regclass(events) IS NULL; + CONTINUE WHEN EXISTS (SELECT 1 FROM pg_constraint fk + WHERE fk.conrelid = events::regclass AND fk.contype = 'f' + AND fk.confrelid = 'reporting_status_scope_checkpoints'::regclass); + EXECUTE format('LOCK TABLE %I IN ACCESS EXCLUSIVE MODE', events); + IF NOT EXISTS (SELECT 1 FROM pg_attribute + WHERE attrelid = events::regclass AND attname = 'offering_id' AND NOT attisdropped) THEN + EXECUTE format( + 'ALTER TABLE %I ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT %L', + events, ''); + -- An event of an offering's checkpoint takes that offering. Events + -- are otherwise immutable: only this new column is filled, once. + IF EXISTS (SELECT 1 FROM reporting_status_scope_checkpoints WHERE offering_id <> '') THEN + EXECUTE format('ALTER TABLE %I DISABLE TRIGGER USER', events); + EXECUTE format( + 'UPDATE %I e SET offering_id = c.offering_id' + ' FROM reporting_status_scope_checkpoints c' + ' WHERE c.account_id = e.account_id' + ' AND c.consumer_namespace = e.consumer_namespace' + ' AND c.delivery_config_id = e.delivery_config_id AND c.version = e.version' + ' AND c.scope_kind = e.scope_kind' + ' AND c.obligation_namespace = e.obligation_namespace' + ' AND c.offering_id <> %L', events, ''); + EXECUTE format('ALTER TABLE %I ENABLE TRIGGER USER', events); + END IF; + END IF; + -- One event per checkpoint generation, under the widened key. The + -- replacement takes the name PostgreSQL gave the six-column one. + FOR reference IN + SELECT unique_key.conname::TEXT AS constraint_name + FROM pg_constraint unique_key + WHERE unique_key.conrelid = events::regclass AND unique_key.contype = 'u' + AND ARRAY( + SELECT attribute.attname::TEXT + FROM unnest(unique_key.conkey) WITH ORDINALITY AS key_column(attnum, position) + JOIN pg_attribute attribute ON attribute.attrelid = unique_key.conrelid + AND attribute.attnum = key_column.attnum + ORDER BY key_column.position) = ARRAY[ + 'account_id', 'consumer_namespace', 'delivery_config_id', 'version', + 'scope_kind', 'obligation_namespace', 'cause_generation'] + LOOP + EXECUTE format('ALTER TABLE %I DROP CONSTRAINT %I', events, reference.constraint_name); + EXECUTE format( + 'ALTER TABLE %I ADD CONSTRAINT %I UNIQUE (account_id, consumer_namespace,' + ' delivery_config_id, version, scope_kind, obligation_namespace, offering_id,' + ' cause_generation)', events, reference.constraint_name); + END LOOP; + -- PostgreSQL names the reference as it named the six-column one. + EXECUTE format( + 'ALTER TABLE %I ADD FOREIGN KEY (account_id, consumer_namespace,' + ' delivery_config_id, version, scope_kind, obligation_namespace, offering_id)' + ' REFERENCES reporting_status_scope_checkpoints', events); + IF NOT EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = events::regclass AND conname = events || '_offering') THEN + EXECUTE format( + 'ALTER TABLE %I ADD CONSTRAINT %I CHECK' + ' (coalesce(snapshot #>> %L, %L) = offering_id)', + events, events || '_offering', '{cause,scope,generation_key,offering_id}', ''); + END IF; + END LOOP; + + -- A checkpoint never changes its key, offering included, and an obligation + -- scope names an obligation of its own generation. + CREATE OR REPLACE FUNCTION reporting_status_scope_guard() + RETURNS TRIGGER LANGUAGE plpgsql AS $scope_guard$ + DECLARE owner reporting_obligations; + BEGIN + IF TG_OP='UPDATE' AND ( + ROW(NEW.account_id, NEW.consumer_namespace, NEW.delivery_config_id, NEW.version, + NEW.scope_kind, NEW.obligation_namespace, NEW.offering_id) IS DISTINCT FROM + ROW(OLD.account_id, OLD.consumer_namespace, OLD.delivery_config_id, OLD.version, + OLD.scope_kind, OLD.obligation_namespace, OLD.offering_id) + OR NEW.generation < OLD.generation OR NEW.generation > OLD.generation + 1 + OR NEW.source_sequence < OLD.source_sequence + OR (OLD.initialized AND NOT NEW.initialized) + OR (OLD.baseline AND NOT NEW.baseline) + OR (OLD.initialized AND + ((NEW.generation <> OLD.generation) <> (NEW.fingerprint <> OLD.fingerprint)))) THEN + RAISE EXCEPTION 'invalid status checkpoint transition' USING ERRCODE='23514'; + END IF; + IF NEW.scope_kind='obligation' THEN + SELECT * INTO owner FROM reporting_obligations + WHERE account_id=NEW.account_id AND consumer_id=NEW.consumer_namespace + AND reporting_obligation_id=NEW.obligation_namespace; + IF NOT FOUND OR owner.delivery_config_id <> NEW.delivery_config_id + OR owner.delivery_config_version <> NEW.version + OR owner.offering_id <> NEW.offering_id THEN + RAISE EXCEPTION 'invalid status scope owner' USING ERRCODE='23514'; + END IF; + END IF; + IF NEW.initialized AND ( + NOT NEW.publishable OR NEW.snapshot->>'health' IS NULL + OR jsonb_typeof(NEW.snapshot->'issues') IS DISTINCT FROM 'array' + OR CASE WHEN NEW.snapshot->>'health' IN ('delayed','action_required') THEN + NOT EXISTS (SELECT 1 FROM jsonb_array_elements(NEW.snapshot->'issues') i + WHERE i->>'severity'=NEW.snapshot->>'health') + ELSE jsonb_array_length(NEW.snapshot->'issues')<>0 END) THEN + RAISE EXCEPTION 'invalid status issue invariant' USING ERRCODE='23514'; + END IF; + RETURN NEW; + END + $scope_guard$; + IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_status_scope_checkpoints'::regclass + AND tgname='reporting_status_scope_guard' AND NOT tgisinternal) THEN + CREATE TRIGGER reporting_status_scope_guard BEFORE INSERT OR UPDATE + ON reporting_status_scope_checkpoints FOR EACH ROW EXECUTE FUNCTION reporting_status_scope_guard(); + END IF; + + -- An event publishes exactly its checkpoint's current generation. + CREATE OR REPLACE FUNCTION reporting_status_event_guard() + RETURNS TRIGGER LANGUAGE plpgsql AS $event_guard$ + DECLARE checkpoint reporting_status_scope_checkpoints; ids jsonb; + BEGIN + SELECT * INTO checkpoint FROM reporting_status_scope_checkpoints + WHERE account_id=NEW.account_id AND consumer_namespace=NEW.consumer_namespace + AND delivery_config_id=NEW.delivery_config_id AND version=NEW.version + AND scope_kind=NEW.scope_kind AND obligation_namespace=NEW.obligation_namespace + AND offering_id=NEW.offering_id; + IF NOT FOUND OR NOT checkpoint.initialized OR NOT checkpoint.publishable + OR checkpoint.generation<>NEW.cause_generation OR checkpoint.fingerprint<>NEW.fingerprint + OR checkpoint.snapshot->>'health' IS DISTINCT FROM NEW.snapshot #>> '{cause,health}' THEN + RAISE EXCEPTION 'invalid status event checkpoint' USING ERRCODE='23514'; + END IF; + SELECT coalesce(jsonb_agg(issue_id ORDER BY issue_id COLLATE "C"), '[]') INTO ids + FROM (SELECT DISTINCT i->>'issue_id' COLLATE "C" AS issue_id + FROM jsonb_array_elements(checkpoint.snapshot->'issues') i + ORDER BY issue_id LIMIT 16) selected; + IF ids IS DISTINCT FROM NEW.snapshot #> '{cause,issue_ids}' THEN + RAISE EXCEPTION 'invalid status event issues' USING ERRCODE='23514'; + END IF; + RETURN NEW; + END + $event_guard$; + IF to_regclass('reporting_status_notification_events') IS NOT NULL AND NOT EXISTS ( + SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_status_notification_events'::regclass + AND tgname='reporting_status_event_guard' AND NOT tgisinternal) THEN + CREATE TRIGGER reporting_status_event_guard BEFORE INSERT + ON reporting_status_notification_events FOR EACH ROW EXECUTE FUNCTION reporting_status_event_guard(); + END IF; + + IF to_regclass('reporting_projection_accounts') IS NULL THEN + RETURN; + END IF; + -- The projection's status events: the same checkpoint check, after + -- activation. reporting_projection_notifications.sql installs its trigger. + CREATE OR REPLACE FUNCTION reporting_projection_event_guard() + RETURNS TRIGGER LANGUAGE plpgsql AS $projection_event_guard$ + DECLARE checkpoint reporting_status_scope_checkpoints; ids jsonb; + BEGIN + IF NOT EXISTS (SELECT 1 FROM reporting_projection_accounts + WHERE account_id=NEW.account_id AND current_input IS NOT NULL + AND notifications_enabled) THEN + RAISE EXCEPTION 'status_projection_activation_required' USING ERRCODE='23514'; + END IF; + SELECT * INTO checkpoint FROM reporting_status_scope_checkpoints + WHERE account_id=NEW.account_id AND consumer_namespace=NEW.consumer_namespace + AND delivery_config_id=NEW.delivery_config_id AND version=NEW.version + AND scope_kind=NEW.scope_kind AND obligation_namespace=NEW.obligation_namespace + AND offering_id=NEW.offering_id; + IF NOT FOUND OR NOT checkpoint.initialized OR NOT checkpoint.publishable + OR checkpoint.generation<>NEW.cause_generation OR checkpoint.fingerprint<>NEW.fingerprint + OR checkpoint.snapshot->>'health' IS DISTINCT FROM NEW.snapshot #>> '{cause,health}' THEN + RAISE EXCEPTION 'invalid status event checkpoint' USING ERRCODE='23514'; + END IF; + SELECT coalesce(jsonb_agg(issue_id ORDER BY issue_id COLLATE "C"), '[]') INTO ids + FROM (SELECT DISTINCT i->>'issue_id' COLLATE "C" AS issue_id + FROM jsonb_array_elements(checkpoint.snapshot->'issues') i + ORDER BY issue_id LIMIT 16) selected; + IF ids IS DISTINCT FROM NEW.snapshot #> '{cause,issue_ids}' THEN + RAISE EXCEPTION 'invalid status event issues' USING ERRCODE='23514'; + END IF; + RETURN NEW; + END + $projection_event_guard$; +END +$checkpoint_offerings$; 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 4d8f240f3..43b003187 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 @@ -381,77 +381,9 @@ BEGIN RETURN NEW; END $capture$; - CREATE OR REPLACE FUNCTION reporting_status_scope_guard() - RETURNS TRIGGER LANGUAGE plpgsql AS $scope_guard$ - DECLARE owner reporting_obligations; - BEGIN - IF TG_OP='UPDATE' AND ( - ROW(NEW.account_id, NEW.consumer_namespace, NEW.delivery_config_id, NEW.version, - NEW.scope_kind, NEW.obligation_namespace) IS DISTINCT FROM - ROW(OLD.account_id, OLD.consumer_namespace, OLD.delivery_config_id, OLD.version, - OLD.scope_kind, OLD.obligation_namespace) - OR NEW.generation < OLD.generation OR NEW.generation > OLD.generation + 1 - OR NEW.source_sequence < OLD.source_sequence - OR (OLD.initialized AND NOT NEW.initialized) - OR (OLD.baseline AND NOT NEW.baseline) - OR (OLD.initialized AND - ((NEW.generation <> OLD.generation) <> (NEW.fingerprint <> OLD.fingerprint)))) THEN - RAISE EXCEPTION 'invalid status checkpoint transition' USING ERRCODE='23514'; - END IF; - IF NEW.scope_kind='obligation' THEN - SELECT * INTO owner FROM reporting_obligations - WHERE account_id=NEW.account_id AND consumer_id=NEW.consumer_namespace - AND reporting_obligation_id=NEW.obligation_namespace; - IF NOT FOUND OR owner.delivery_config_id <> NEW.delivery_config_id - OR owner.delivery_config_version <> NEW.version THEN - RAISE EXCEPTION 'invalid status scope owner' USING ERRCODE='23514'; - END IF; - END IF; - IF NEW.initialized AND ( - NOT NEW.publishable OR NEW.snapshot->>'health' IS NULL - OR jsonb_typeof(NEW.snapshot->'issues') IS DISTINCT FROM 'array' - OR CASE WHEN NEW.snapshot->>'health' IN ('delayed','action_required') THEN - NOT EXISTS (SELECT 1 FROM jsonb_array_elements(NEW.snapshot->'issues') i - WHERE i->>'severity'=NEW.snapshot->>'health') - ELSE jsonb_array_length(NEW.snapshot->'issues')<>0 END) THEN - RAISE EXCEPTION 'invalid status issue invariant' USING ERRCODE='23514'; - END IF; - RETURN NEW; - END - $scope_guard$; - CREATE OR REPLACE FUNCTION reporting_status_event_guard() - RETURNS TRIGGER LANGUAGE plpgsql AS $event_guard$ - DECLARE checkpoint reporting_status_scope_checkpoints; ids jsonb; - BEGIN - SELECT * INTO checkpoint FROM reporting_status_scope_checkpoints - WHERE account_id=NEW.account_id AND consumer_namespace=NEW.consumer_namespace - AND delivery_config_id=NEW.delivery_config_id AND version=NEW.version - AND scope_kind=NEW.scope_kind AND obligation_namespace=NEW.obligation_namespace; - IF NOT FOUND OR NOT checkpoint.initialized OR NOT checkpoint.publishable - OR checkpoint.generation<>NEW.cause_generation OR checkpoint.fingerprint<>NEW.fingerprint - OR checkpoint.snapshot->>'health' IS DISTINCT FROM NEW.snapshot #>> '{cause,health}' THEN - RAISE EXCEPTION 'invalid status event checkpoint' USING ERRCODE='23514'; - END IF; - SELECT coalesce(jsonb_agg(issue_id ORDER BY issue_id COLLATE "C"), '[]') INTO ids - FROM (SELECT DISTINCT i->>'issue_id' COLLATE "C" AS issue_id - FROM jsonb_array_elements(checkpoint.snapshot->'issues') i - ORDER BY issue_id LIMIT 16) selected; - IF ids IS DISTINCT FROM NEW.snapshot #> '{cause,issue_ids}' THEN - RAISE EXCEPTION 'invalid status event issues' USING ERRCODE='23514'; - END IF; - RETURN NEW; - END - $event_guard$; - IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_status_scope_checkpoints'::regclass - AND tgname='reporting_status_scope_guard' AND NOT tgisinternal) THEN - CREATE TRIGGER reporting_status_scope_guard BEFORE INSERT OR UPDATE - ON reporting_status_scope_checkpoints FOR EACH ROW EXECUTE FUNCTION reporting_status_scope_guard(); - END IF; - IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_status_notification_events'::regclass - AND tgname='reporting_status_event_guard' AND NOT tgisinternal) THEN - CREATE TRIGGER reporting_status_event_guard BEFORE INSERT - ON reporting_status_notification_events FOR EACH ROW EXECUTE FUNCTION reporting_status_event_guard(); - END IF; + -- reporting_status_scope_guard and reporting_status_event_guard compare the + -- full checkpoint key, offering included: reporting_status_checkpoint_offerings.sql + -- defines them and installs their triggers. IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_status_dirty'::regclass AND tgname='reporting_status_boundary_mark' AND NOT tgisinternal) THEN CREATE TRIGGER reporting_status_boundary_mark AFTER INSERT ON reporting_status_dirty diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py index 24ea1e70f..fb56b1ee3 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py @@ -577,6 +577,7 @@ async def migrate_legacy_reporting( "reporting_status_notifications.sql", "reporting_status_selector_version.sql", "reporting_projection.sql", + "reporting_status_checkpoint_offerings.sql", "reporting_projection_notifications.sql", "reporting_projection_feed.sql", "reporting_production.sql", diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0006_reporting_status_checkpoint_offerings.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0006_reporting_status_checkpoint_offerings.sql new file mode 100644 index 000000000..703da2499 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0006_reporting_status_checkpoint_offerings.sql @@ -0,0 +1,15 @@ +-- adcp-migration: 0006 reporting_status_checkpoint_offerings +-- adcp-upgrade: stopped-worker +-- Keys status checkpoints, and the status events that reference them, by the +-- generation's resolved offering: (account_id, consumer_namespace, +-- delivery_config_id, version, scope_kind, obligation_namespace, offering_id). +-- Two offerings of one subscription version, for one account and owner, keep +-- their own checkpoints. Existing rows keep offering_id '' and their key, and +-- the ids and digests derived from an account-level checkpoint are unchanged. +-- +-- This is a stopped-worker upgrade: stop every reporting worker before running +-- it. Workers of earlier releases insert checkpoints on the six-column key, +-- which no longer exists, and swapping a primary key takes an exclusive lock +-- on the checkpoint and status event tables. Start the new release only after +-- it commits. It changes nothing on a schema without status notifications. +-- include: ledger/reporting_status_checkpoint_offerings.sql 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 fb6401f47..f375315ab 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 @@ -239,6 +239,10 @@ "enabled": true, "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" }, + "column:reporting_status_notification_events.offering_id": { + "enabled": true, + "fingerprint": "128b66e02f14946100273f112c144af7605d626124d20fe16c1ba30c6b19ae3a" + }, "column:reporting_status_notification_events.scope_kind": { "enabled": true, "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" @@ -561,7 +565,7 @@ }, "constraint:reporting_status_notification_events.reporting_status_notification_account_id_consumer_namespac_fkey": { "enabled": true, - "fingerprint": "ee26a73f446c794ba69fc056f5fb1b8e094b25ae48743fe5e23045749e84f1c2" + "fingerprint": "a88d869a73abf6c88d6af29502b32e61e5f9657a03f80959629cb89086522dad" }, "constraint:reporting_status_notification_events.reporting_status_notification_account_id_consumer_namespac_key1": { "enabled": true, @@ -573,7 +577,7 @@ }, "constraint:reporting_status_notification_events.reporting_status_notification_account_id_consumer_namespace_key": { "enabled": true, - "fingerprint": "a21acc6c4e405fcbf7711fe5a82c55206c994c6c969b498f9afe35af859ed206" + "fingerprint": "eea1d95be27f57f078192704cd006232ad6cdb37030fb397e2eccb3a3d430616" }, "constraint:reporting_status_notification_events.reporting_status_notification_events_cause_generation_check": { "enabled": true, @@ -647,6 +651,10 @@ "enabled": true, "fingerprint": "ee42647bcbefde76d58a6b7231c68bc1c4ef047e2523f779b4c28bcd9792eb47" }, + "constraint:reporting_status_notification_events.reporting_status_notification_events_offering": { + "enabled": true, + "fingerprint": "60b6b5efe8e40c3c2c31d72d4026fbe1aa9f26758c74614b160c1cd96f67d8e1" + }, "constraint:reporting_status_notification_events.reporting_status_notification_events_pkey": { "enabled": true, "fingerprint": "e8d63f44b25700915bcb3cb69e2e24b98cf21fd304f36a2e9c2992236309309d" @@ -727,9 +735,13 @@ "enabled": true, "fingerprint": "a306793f3ca65342fe7cd60e06058f661453b3d746d1da301a06e35d1e16d6e9" }, + "constraint:reporting_status_scope_checkpoints.reporting_status_scope_checkpoints_offering": { + "enabled": true, + "fingerprint": "80f5bfd3fca25c4431864e99f221c28c0b36d6459f874da02a78da0150304072" + }, "constraint:reporting_status_scope_checkpoints.reporting_status_scope_checkpoints_pkey": { "enabled": true, - "fingerprint": "9248b24cf88bd9d81f5c44079431cbd72714f7922600c4dd6a99f9a7a190436b" + "fingerprint": "897893cf29e891fd86d26f34b5a10500e585fee2cde3b08bd066e40b73d3a7dd" }, "constraint:reporting_status_scope_checkpoints.reporting_status_scope_checkpoints_scope_kind_check": { "enabled": true, @@ -821,7 +833,7 @@ }, "function:reporting_status_event_guard()": { "enabled": true, - "fingerprint": "e56d1d9017667282bc1a046656fc2833a97252af37dd2c632a8e0e686b5fb175" + "fingerprint": "1df8a065e5be846ea144b110ec4edad86795a585e1eaccce9d27f10be3dc7ae2" }, "function:reporting_status_projection_input(owner text, at_time timestamp with time zone)": { "enabled": true, @@ -829,7 +841,7 @@ }, "function:reporting_status_scope_guard()": { "enabled": true, - "fingerprint": "e52efa691de16f3f72df213ca103d8290807fb127ff790b15d7e68c9133757b7" + "fingerprint": "9291cd021f71ce6e28c720eb221e84f97c6592ef2820b109f6a69de7c2ef8d03" }, "function:reporting_status_webhook_attempt_guard()": { "enabled": true, @@ -885,7 +897,7 @@ }, "index:reporting_status_notification_events.reporting_status_notification_account_id_consumer_namespace_key": { "enabled": true, - "fingerprint": "1c8d9302d57b13d6ffed87ad1ed7ed3199119151ef68a5c28b6a075b01157e3a" + "fingerprint": "5a3f49c499abe6d9d4d161be5c8bc0b65f06b78289c3ad394fd0fb71affbe97d" }, "index:reporting_status_notification_events.reporting_status_notification_events_pkey": { "enabled": true, @@ -901,7 +913,7 @@ }, "index:reporting_status_scope_checkpoints.reporting_status_scope_checkpoints_pkey": { "enabled": true, - "fingerprint": "83b2cc13fef08e3cb5751d0182ff7b6b3b0aab04f7396a233d8a8f85cccd4a99" + "fingerprint": "689f13640980375b350e17731c25ed07813daef5895a59455380e4727421bbea" }, "index:reporting_status_scope_checkpoints.reporting_status_scope_due": { "enabled": true, 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 a9eb32e7d..2b6420898 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 @@ -55,12 +55,15 @@ ) from adcp_reporting_storage.outbox.pg import PgReportingOutbox, database_now +# ReportingStatusScope.checkpoint_key, in column order: offering_id is '' for +# an account-level generation and names a subscription generation's offering. _KEY = ( - "account_id, consumer_namespace, delivery_config_id, version, scope_kind, obligation_namespace" + "account_id, consumer_namespace, delivery_config_id, version, scope_kind," + " obligation_namespace, offering_id" ) _WHERE = ( "account_id=%s AND consumer_namespace=%s AND delivery_config_id=%s AND version=%s" - " AND scope_kind=%s AND obligation_namespace=%s" + " AND scope_kind=%s AND obligation_namespace=%s AND offering_id=%s" ) _CHECKPOINT = ( "scope, fingerprint, generation, snapshot, next_due_at, source_sequence, baseline, publishable," @@ -167,11 +170,10 @@ async def _enqueue_on(connection: Any, event: ReportingDomainEvent) -> None: raise ReportingNotificationError("invalid_status_projection") key = cause.scope.checkpoint_key await connection.execute( - "INSERT INTO reporting_status_notification_events" - " (account_id, consumer_namespace, delivery_config_id, version, scope_kind," - " obligation_namespace, notification_id, notification_type, cause_kind, cause_id," + f"INSERT INTO reporting_status_notification_events ({_KEY}," # nosec B608 + " notification_id, notification_type, cause_kind, cause_id," " cause_generation, fingerprint, fired_at, snapshot)" - " VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s::jsonb)", + " VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s::jsonb)", ( *key, event.notification_id, @@ -263,6 +265,11 @@ async def create_schema(self) -> None: .joinpath("reporting_status_selector_version.sql") .read_text() ) + await connection.execute( + files("adcp_reporting_storage.ledger") + .joinpath("reporting_status_checkpoint_offerings.sql") + .read_text() + ) await validate_status_schema(connection) @asynccontextmanager @@ -293,15 +300,13 @@ 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}, offering_id, scope," # nosec B608 + f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, 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, - # 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, ), @@ -354,8 +359,7 @@ async def _apply_on( rows = await ( await connection.execute( "SELECT scope FROM reporting_status_scope_checkpoints WHERE account_id=%s" - " ORDER BY account_id, consumer_namespace, delivery_config_id, version," - " scope_kind, obligation_namespace", + f" ORDER BY {_KEY}", # nosec B608 (snapshot.account_id,), ) ).fetchall() @@ -484,8 +488,7 @@ async def _rebuild_on(self, connection: Any, account_id: str) -> StatusTurn: await ( await connection.execute( "SELECT 1 FROM reporting_status_scope_checkpoints WHERE account_id=%s" - " ORDER BY account_id, consumer_namespace, delivery_config_id, version," - " scope_kind, obligation_namespace FOR UPDATE", + f" ORDER BY {_KEY} FOR UPDATE", # nosec B608 (account_id,), ) ).fetchall() 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 275dff3d7..beca8ff4c 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 @@ -507,17 +507,6 @@ async def _enroll_owed_set_on( 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," @@ -1023,6 +1012,7 @@ async def create_schema(self) -> None: "reporting_status_notifications.sql", "reporting_status_selector_version.sql", "reporting_projection.sql", + "reporting_status_checkpoint_offerings.sql", "reporting_projection_notifications.sql", "reporting_projection_feed.sql", "reporting_production.sql", diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/pg.py index 577af64d6..98eb9ed09 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/pg.py @@ -80,6 +80,7 @@ async def create_schema(self) -> None: "reporting_status_notifications.sql", "reporting_status_selector_version.sql", "reporting_projection.sql", + "reporting_status_checkpoint_offerings.sql", "reporting_projection_notifications.sql", "reporting_projection_feed.sql", ): @@ -330,7 +331,7 @@ async def _begin_activation(self, *, account_id: str) -> bool: " reporting_status_scope_checkpoints" " WHERE account_id=%s ORDER BY" " consumer_namespace,delivery_config_id,version,scope_kind," - " obligation_namespace FOR UPDATE", + " obligation_namespace,offering_id FOR UPDATE", (account_id,), ) ).fetchall() diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/required_schema.json b/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/required_schema.json index dc59704f2..e7b45c90a 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/required_schema.json +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/projection/required_schema.json @@ -355,6 +355,10 @@ "enabled": true, "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" }, + "column:reporting_projection_notification_events.offering_id": { + "enabled": true, + "fingerprint": "128b66e02f14946100273f112c144af7605d626124d20fe16c1ba30c6b19ae3a" + }, "column:reporting_projection_notification_events.scope_kind": { "enabled": true, "fingerprint": "fdc4b549138bf90a5af5ed00b87a86ff9a0cbea553e7f017b4d95510708f6f88" @@ -785,7 +789,7 @@ }, "constraint:reporting_projection_notification_events.reporting_projection_notifica_account_id_consumer_namespac_fkey": { "enabled": true, - "fingerprint": "ee26a73f446c794ba69fc056f5fb1b8e094b25ae48743fe5e23045749e84f1c2" + "fingerprint": "a88d869a73abf6c88d6af29502b32e61e5f9657a03f80959629cb89086522dad" }, "constraint:reporting_projection_notification_events.reporting_projection_notifica_account_id_consumer_namespac_key1": { "enabled": true, @@ -797,7 +801,7 @@ }, "constraint:reporting_projection_notification_events.reporting_projection_notifica_account_id_consumer_namespace_key": { "enabled": true, - "fingerprint": "a21acc6c4e405fcbf7711fe5a82c55206c994c6c969b498f9afe35af859ed206" + "fingerprint": "eea1d95be27f57f078192704cd006232ad6cdb37030fb397e2eccb3a3d430616" }, "constraint:reporting_projection_notification_events.reporting_projection_notification_event_notification_type_check": { "enabled": true, @@ -875,6 +879,10 @@ "enabled": true, "fingerprint": "e6f7fe9bd1925eb05eeab481da0229b0180558dd2448eda9e51e98faf31d6497" }, + "constraint:reporting_projection_notification_events.reporting_projection_notification_events_offering": { + "enabled": true, + "fingerprint": "60b6b5efe8e40c3c2c31d72d4026fbe1aa9f26758c74614b160c1cd96f67d8e1" + }, "constraint:reporting_projection_notification_events.reporting_projection_notification_events_pkey": { "enabled": true, "fingerprint": "e8d63f44b25700915bcb3cb69e2e24b98cf21fd304f36a2e9c2992236309309d" @@ -1005,7 +1013,7 @@ }, "function:reporting_projection_event_guard()": { "enabled": true, - "fingerprint": "64f503d73db42c174921cbb9cbaf822383e39d35607c2ce0d9ff315c2c5f3074" + "fingerprint": "294683741aba35c138ca5bdefcb0eac25d7942529cf9b4aba2bdb74d244c4d22" }, "function:reporting_projection_feed_immutable()": { "enabled": true, @@ -1097,7 +1105,7 @@ }, "index:reporting_projection_notification_events.reporting_projection_notifica_account_id_consumer_namespace_key": { "enabled": true, - "fingerprint": "ca07092e6e38681fee63f6717cf10f1446bf3b8bc117a73cde19b5dcc987be5c" + "fingerprint": "eacd2bb39d94ed5ed87334e4983e1de1432bdf889dfe95b7c448be04d96a8cfb" }, "index:reporting_projection_notification_events.reporting_projection_notification_events_pkey": { "enabled": true, diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py index 7fd83087c..0433f8a54 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py @@ -78,9 +78,13 @@ async def check_schema() -> None: if pending: versions = ", ".join(item.version for item in pending) stopped = [item.version for item in pending if item.stopped_worker] + upgrades = ( + "is a stopped-worker upgrade" + if len(stopped) == 1 + else "are stopped-worker upgrades" + ) note = ( - f"; {', '.join(stopped)} is a stopped-worker upgrade, so stop every reporting " - "worker first" + f"; {', '.join(stopped)} {upgrades}, so stop every reporting worker first" if stopped else "" ) diff --git a/src/adcp/reporting/ledger/notification_models.py b/src/adcp/reporting/ledger/notification_models.py index 4f94e16cb..2bb9550d5 100644 --- a/src/adcp/reporting/ledger/notification_models.py +++ b/src/adcp/reporting/ledger/notification_models.py @@ -196,12 +196,7 @@ def cause_id(self) -> str: if isinstance(cause, AdjustmentPublished): return cause.reporting_adjustment_id if isinstance(cause, StatusChanged): - from adcp.reporting.canonical_json import canonical_json_utf8_v1 - - return ( - "rpsc_" - + hashlib.sha256(canonical_json_utf8_v1(cause.scope.checkpoint_key)).hexdigest() - ) + return "rpsc_" + cause.scope.checkpoint_digest # Structured encoding: consumers and materializations may reuse IDs. return json.dumps( [cause.consumer_id, cause.reporting_materialization_id], separators=(",", ":") @@ -312,8 +307,14 @@ def __post_init__(self) -> None: object.__setattr__(self, "consumer_id", self.generation_key.consumer_id) @property - def checkpoint_key(self) -> tuple[str, str, str, int, str, str]: - """Six independent, non-null columns. Scope kinds never share an ID space.""" + def checkpoint_key(self) -> tuple[str, str, str, int, str, str, str]: + """Seven independent, non-null columns. Scope kinds never share an ID space. + + The last column is the generation's resolved offering: a reporting + subscription keeps one generation per resolved offering of a version, + so each offering has its own checkpoints. It is ``''`` for an + account-level generation. + """ if self.generation_key is None: raise ReportingNotificationError("invalid_status_scope") return ( @@ -323,8 +324,24 @@ def checkpoint_key(self) -> tuple[str, str, str, int, str, str]: self.generation_key.delivery_config_version, "obligation" if self.reporting_obligation_id is not None else "configuration", self.reporting_obligation_id or "", + self.generation_key.offering_id or "", ) + @property + def checkpoint_digest(self) -> str: + """SHA-256 of the canonical checkpoint key, for persisted ids and digests. + + An account-level generation (offering ``''``) digests the six columns + it always had, so the status ``cause_id`` values and history keys that + deployments already store keep their meaning. Only an offering's + checkpoint digests the seventh column. + """ + from adcp.reporting.canonical_json import canonical_json_utf8_v1 + + key = self.checkpoint_key + identity = key if key[-1] else key[:-1] + return hashlib.sha256(canonical_json_utf8_v1(identity)).hexdigest() + @classmethod def for_obligation( cls, obligation: ReportingObligationRecord, consumer_id: str | None = None diff --git a/src/adcp/reporting/outbox/status_memory.py b/src/adcp/reporting/outbox/status_memory.py index 2732a4eef..393824fab 100644 --- a/src/adcp/reporting/outbox/status_memory.py +++ b/src/adcp/reporting/outbox/status_memory.py @@ -37,7 +37,7 @@ @dataclass class _StatusMemoryState: accounts: dict[str, tuple[int, dict[str, Any]]] = field(default_factory=dict) - checkpoints: dict[tuple[str, str, str, int, str, str], StatusCheckpoint] = field( + checkpoints: dict[tuple[str, str, str, int, str, str, str], StatusCheckpoint] = field( default_factory=dict ) outbox: NotificationState = field(default_factory=NotificationState) @@ -73,6 +73,13 @@ def __init__( if not hasattr(state, "selector_accounts"): state.selector_accounts = {} for key, checkpoint in tuple(state.checkpoints.items()): + # An old shared-state image keys a checkpoint by its six columns, + # before the resolved offering joined the key. Re-key it under the + # scope's own key; an account-level scope's offering is ''. + if key != checkpoint.scope.checkpoint_key: + del state.checkpoints[key] + key = checkpoint.scope.checkpoint_key + state.checkpoints[key] = checkpoint # An old shared-state image has no epoch fields. Decode by keyword # rather than letting new dataclass class defaults label it v2. if "selector_semantics_version" not in vars(checkpoint): diff --git a/src/adcp/reporting/production/memory.py b/src/adcp/reporting/production/memory.py index 187c60309..3fba83025 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, replace +from dataclasses import dataclass from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any @@ -281,13 +281,6 @@ 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/src/adcp/reporting/projection/history.py b/src/adcp/reporting/projection/history.py index fa58e5953..4bed40f39 100644 --- a/src/adcp/reporting/projection/history.py +++ b/src/adcp/reporting/projection/history.py @@ -9,7 +9,6 @@ from __future__ import annotations -import hashlib from collections.abc import Mapping from dataclasses import dataclass from typing import Any, Literal @@ -17,7 +16,6 @@ from pydantic import TypeAdapter from adcp._deferred_adapters import deferred_adapter -from adcp.reporting.canonical_json import canonical_json_utf8_v1 from adcp.reporting.ledger.models import ReportingDeliveryEscalation from adcp.reporting.ledger.notification_models import ( ReportingNotificationError, @@ -53,7 +51,7 @@ def decode_boundary(kind: HistoricalKind, value: dict[str, Any]) -> HistoricalBo def checkpoint_key(checkpoint: StatusCheckpoint) -> str: - return hashlib.sha256(canonical_json_utf8_v1(checkpoint.scope.checkpoint_key)).hexdigest() + return checkpoint.scope.checkpoint_digest def checkpoint_document(checkpoint: StatusCheckpoint) -> dict[str, Any]: @@ -113,7 +111,7 @@ def project_boundary( consumer_status_enabled=consumer_status_enabled, ) ) - key = hashlib.sha256(canonical_json_utf8_v1(scope.checkpoint_key)).hexdigest() + key = scope.checkpoint_digest checkpoint, event = advance_checkpoint( previous.get(key, baselines.get(key)), result, diff --git a/tests/conformance/reporting/_empty_ledger_support.py b/tests/conformance/reporting/_empty_ledger_support.py index 0c1f2d93c..a541463c9 100644 --- a/tests/conformance/reporting/_empty_ledger_support.py +++ b/tests/conformance/reporting/_empty_ledger_support.py @@ -51,6 +51,7 @@ def _storage_split(): "reporting_status_notifications.sql", "reporting_status_selector_version.sql", "reporting_projection.sql", + "reporting_status_checkpoint_offerings.sql", "reporting_projection_notifications.sql", "reporting_projection_feed.sql", "reporting_production.sql", diff --git a/tests/conformance/reporting/_production_packaging.py b/tests/conformance/reporting/_production_packaging.py index 3bc5aaf82..99c4b5ea7 100644 --- a/tests/conformance/reporting/_production_packaging.py +++ b/tests/conformance/reporting/_production_packaging.py @@ -41,6 +41,7 @@ def _storage_split(): "ledger/reporting_status_selector_version.sql", "outbox/required_status_selector_schema.json", "ledger/reporting_projection.sql", + "ledger/reporting_status_checkpoint_offerings.sql", "ledger/reporting_projection_notifications.sql", "ledger/reporting_projection_feed.sql", "projection/required_schema.json", diff --git a/tests/conformance/reporting/test_reporting_status_projection_contract.py b/tests/conformance/reporting/test_reporting_status_projection_contract.py index a31314161..913f1caed 100644 --- a/tests/conformance/reporting/test_reporting_status_projection_contract.py +++ b/tests/conformance/reporting/test_reporting_status_projection_contract.py @@ -9,7 +9,7 @@ import pytest from adcp_reporting_storage import PgStatusNotificationStore -from adcp.reporting.ledger import InMemoryReportingLedgerStore +from adcp.reporting.ledger import InMemoryReportingLedgerStore, ReportingSubscriptionGeneration from adcp.reporting.ledger import ReportingStatusCaller as OwnershipCaller from adcp.reporting.ledger.status import ReportingStatusCaller, ReportingStatusHandler from adcp.reporting.ledger.status_projection import StatusProjectionInput, project_status_scope @@ -297,3 +297,65 @@ async def test_partial_coverage_has_a_public_schema_valid_issue(status_harness): coverage = next(i for i in summary["issues"] if i["code"] == "REPORTING_COVERAGE_INCOMPLETE") assert coverage["issue_id"] in (await h.events())[-1].cause.issue_ids assert await h.status.baseline_ready(account_id="acct_a") + + +async def test_each_resolved_offering_of_a_subscription_version_has_its_own_checkpoint( + status_harness, +): + """One subscription version, one account and owner, two resolved offerings. + + A reporting subscription keeps one generation per resolved offering of a + version (adcp-client-python#1729), so each offering's configuration and + obligations keep their own status checkpoint and event history. + """ + h = status_harness + obligations = {} + for offering in ("pkg-a", "pkg-b"): + config = replace( + configuration(), + subscription=ReportingSubscriptionGeneration("daily", offering), + ) + await h.ledger.put_configuration(config) + obligations[offering] = await h.ledger.commit_obligation( + replace( + obligation_for(config), + reporting_obligation_id=f"rpo_acct_a_{offering}", + offering_id=offering, + ) + ) + await h.status.baseline(account_id="acct_a") + checkpoints = await h.status.checkpoints(account_id="acct_a") + assert sorted( + ( + c.scope.generation_key.offering_id, + c.scope.reporting_obligation_id is not None, + ) + for c in checkpoints + ) == [("pkg-a", False), ("pkg-a", True), ("pkg-b", False), ("pkg-b", True)] + assert len({c.scope.checkpoint_key for c in checkpoints}) == 4 + # Each offering's configuration scope advances and publishes on its own. + h.clock.now = max(o.automated_recovery_deadline_at for o in obligations.values()) + await h.drain() + await ReportingStatusSweeper(h.status).run_once(account_id="acct_a") + events = await h.events(obligation=False) + assert sorted(e.cause.scope.generation_key.offering_id for e in events) == ["pkg-a", "pkg-b"] + assert len({e.cause_id for e in events}) == 2 + assert len({e.notification_id for e in events}) == 2 + + +def test_account_level_checkpoint_identity_is_unchanged(): + """Persisted status ids derived from an account-level checkpoint stay byte-identical.""" + import hashlib + + from adcp.reporting.canonical_json import canonical_json_utf8_v1 + + config = configuration() + scope = ReportingStatusScope("acct_a", config.generation_key, feed_purpose="analytics") + assert scope.checkpoint_key[-1] == "" + legacy = hashlib.sha256(canonical_json_utf8_v1(scope.checkpoint_key[:6])).hexdigest() + assert scope.checkpoint_digest == legacy + subscription = replace(config, subscription=ReportingSubscriptionGeneration("daily", "pkg")) + offering = ReportingStatusScope("acct_a", subscription.generation_key, feed_purpose="analytics") + assert offering.checkpoint_key[:6] == scope.checkpoint_key[:6] + assert offering.checkpoint_key[-1] == "pkg" + assert offering.checkpoint_digest != scope.checkpoint_digest diff --git a/tests/conformance/reporting/test_reporting_status_transactions.py b/tests/conformance/reporting/test_reporting_status_transactions.py index 9ebd44242..4a4f9074d 100644 --- a/tests/conformance/reporting/test_reporting_status_transactions.py +++ b/tests/conformance/reporting/test_reporting_status_transactions.py @@ -275,7 +275,8 @@ async def execute(self, query, params=None, **kwargs): " fenced AS (UPDATE reporting_status_scope_checkpoints SET" " lease_expires_at=moment.at FROM moment WHERE account_id=%s" " AND consumer_namespace=%s AND delivery_config_id=%s AND version=%s" - " AND scope_kind=%s AND obligation_namespace=%s AND lease_token=%s" + " AND scope_kind=%s AND obligation_namespace=%s AND offering_id=%s" + " AND lease_token=%s" " RETURNING lease_expires_at) SELECT lease_expires_at FROM fenced", (*lease.scope.checkpoint_key, lease.token), ) diff --git a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py index 67e9c9a81..199d6016a 100644 --- a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py @@ -47,9 +47,9 @@ async def test_storage_refuses_a_schema_that_is_behind_and_starts_once_migrated( await storage.check_schema() error = refused.value assert error.code == "REPORTING_SCHEMA_BEHIND" and error.recovery == "correctable" - assert "pending: 0001, 0002, 0003, 0004, 0005" in str(error) + assert "pending: 0001, 0002, 0003, 0004, 0005, 0006" in str(error) assert "python -m adcp migrate" in str(error) - assert "0002 is a stopped-worker upgrade" in str(error) + assert "0002, 0006 are stopped-worker upgrades" in str(error) await migrate(pool) await storage.check_schema() @@ -86,7 +86,7 @@ async def test_the_cli_migrates_and_reports_status(capsys) -> None: assert await _run(dsn, status=True, dry_run=False) == 0 status = capsys.readouterr().out assert "0002 reporting_subscription_generations: pending [stopped-worker upgrade" in status - assert "5 pending" in status + assert "6 pending" in status assert await _run(dsn, status=False, dry_run=False) == 0 assert "applied 0002 reporting_subscription_generations (stopped-worker upgrade)" in ( capsys.readouterr().out diff --git a/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py b/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py index 5b790b9cf..fc0409e20 100644 --- a/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py +++ b/tests/conformance/reporting/test_reporting_subscription_ledger_pg.py @@ -185,18 +185,158 @@ async def test_a_foreign_reference_to_the_key_refuses_the_widening() -> None: await PgReportingLedgerStore(pool=pool).create_schema() -async def test_status_notifications_refuse_subscription_generations() -> None: +async def test_status_notifications_cover_subscription_generations() -> None: async with isolated_reporting_pool() as pool: store = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) await store.create_schema() world = await _world_with( - AGENCY, [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z")] + AGENCY, + [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z")], + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + "match": "all", + }, ) - # A store with status notifications on (its outbox schema aside). + # A store with status notifications on (its outbox schema aside): + # status checkpoints are keyed by offering, so every offering of one + # subscription version is admitted. store._notifications_enabled = True - with pytest.raises(LedgerConflictError) as refused: - await _plan(world, AGENCY, store, "2026-10-04T00:00:00Z") - assert refused.value.code == "UNSUPPORTED_FEATURE" + (plan,) = await _plan(world, AGENCY, store, "2026-10-04T00:00:00Z") + assert len({key.offering_id for key in plan.generations}) == len(plan.generations) > 1 + + +async def test_a_six_column_status_checkpoint_schema_is_keyed_by_offering() -> None: + """Migration 0006 widens status checkpoints and events published on the version key.""" + import json + from dataclasses import replace + + from adcp_reporting_storage.migrations import migrate, migrations, pending_migrations + from adcp_reporting_storage.outbox.status_schema import validate_status_schema + + from adcp.reporting.ledger import ReportingSubscriptionGeneration + from adcp.reporting.ledger.models import ledger_asdict + from adcp.reporting.ledger.notification_models import ( + ReportingDomainEvent, + ReportingStatusScope, + StatusChanged, + event_storage, + ) + from tests.conformance.reporting._generation_support import configuration + + async def checkpoint(connection, scope: ReportingStatusScope) -> None: + await connection.execute( + "INSERT INTO reporting_status_scope_checkpoints (account_id, consumer_namespace," + " delivery_config_id, version, scope_kind, obligation_namespace, offering_id, scope," + " fingerprint, snapshot, generation, source_sequence, baseline, publishable," + " initialized) VALUES (%s,%s,%s,%s,%s,%s,%s,%s::jsonb,%s,%s::jsonb,1,0,TRUE,TRUE,TRUE)", + ( + *scope.checkpoint_key, + json.dumps(ledger_asdict(scope)), + "a" * 64, + json.dumps({"health": "complete", "issues": []}), + ), + ) + + async def event(connection, scope: ReportingStatusScope, *, widened: bool) -> None: + value = ReportingDomainEvent( + "acct_a", + f"rpn_{scope.generation_key.offering_id or 'account'}", + NOW, + StatusChanged(scope, "complete", "a" * 64, 1), + ) + key = scope.checkpoint_key if widened else scope.checkpoint_key[:6] + await connection.execute( + "INSERT INTO reporting_status_notification_events (account_id, consumer_namespace," + " delivery_config_id, version, scope_kind, obligation_namespace," + + (" offering_id," if widened else "") + + " notification_id, notification_type, cause_kind, cause_id, cause_generation," + " fingerprint, fired_at, snapshot) VALUES (" + + ",".join(["%s"] * (len(key) + 7)) + + ",%s::jsonb)", + ( + *key, + value.notification_id, + value.notification_type, + "status_changed", + value.cause_id, + 1, + "a" * 64, + NOW, + json.dumps(event_storage(value)), + ), + ) + + def scope(config) -> ReportingStatusScope: + return ReportingStatusScope("acct_a", config.generation_key, feed_purpose="analytics") + + (widening,) = [item for item in migrations() if item.version == "0006"] + assert widening.stopped_worker + assert widening.includes == ("ledger/reporting_status_checkpoint_offerings.sql",) + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) # 0006 changes nothing without status notifications + store = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) + account = replace(configuration(), delivery_config_id="hourly") + offerings = { + offering: replace( + configuration(), subscription=ReportingSubscriptionGeneration("daily", offering) + ) + for offering in ("pkg-a", "pkg-b") + } + for config in (account, *offerings.values()): + await store.put_configuration(config) + ledger = files("adcp_reporting_storage.ledger") + async with pool.connection() as connection: + # The published status layout: checkpoints and their events keyed + # per subscription version, holding at most one offering of it. + for name in ( + "reporting_status_notifications.sql", + "reporting_status_selector_version.sql", + ): + await connection.execute(ledger.joinpath(name).read_text()) + await connection.execute("INSERT INTO reporting_status_accounts VALUES ('acct_a')") + for config in (account, offerings["pkg-a"]): + await checkpoint(connection, scope(config)) + await event(connection, scope(config), widened=False) + await connection.execute("DELETE FROM adcp_schema_migrations WHERE version = '0006'") + assert [item.version for item in await pending_migrations(pool)] == ["0006"] + assert [item.version for item in await migrate(pool)] == ["0006"] + assert await migrate(pool) == [] + async with pool.connection() as connection: + for statement in widening.statements(): # idempotent + await connection.execute(statement) + await validate_status_schema(connection) + primary = await ( + await connection.execute( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint WHERE conrelid =" + " 'reporting_status_scope_checkpoints'::regclass AND contype = 'p'" + ) + ).fetchone() + assert primary[0].endswith("obligation_namespace, offering_id)") + # Each event takes its checkpoint's offering. + rows = await ( + await connection.execute( + "SELECT notification_id, offering_id FROM reporting_status_notification_events" + " ORDER BY notification_id" + ) + ).fetchall() + assert rows == [("rpn_account", ""), ("rpn_pkg-a", "pkg-a")] + # A second offering of the version keeps its own checkpoint and events. + await checkpoint(connection, scope(offerings["pkg-b"])) + await event(connection, scope(offerings["pkg-b"]), widened=True) + keys = await ( + await connection.execute( + "SELECT delivery_config_id, scope_kind, offering_id" + " FROM reporting_status_scope_checkpoints ORDER BY 1, 3" + ) + ).fetchall() + assert keys == [ + ("daily", "configuration", "pkg-a"), + ("daily", "configuration", "pkg-b"), + ("hourly", "configuration", ""), + ] async def test_an_account_only_managed_delivery_schema_is_upgraded() -> None: diff --git a/tests/conformance/reporting/test_reporting_subscription_production.py b/tests/conformance/reporting/test_reporting_subscription_production.py index d3a6d218a..b2c96e0c3 100644 --- a/tests/conformance/reporting/test_reporting_subscription_production.py +++ b/tests/conformance/reporting/test_reporting_subscription_production.py @@ -542,8 +542,8 @@ async def test_owed_obligations_are_delivered_with_status_notifications(tmp_path 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: +async def test_a_second_offering_of_one_version_is_served(tmp_path, backend) -> None: + async with _seller(tmp_path, backend, notifications=True) 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") @@ -557,11 +557,12 @@ async def test_a_second_offering_of_one_version_is_refused(tmp_path, backend) -> (*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" + # Each resolved offering keeps its own status checkpoints, so a second + # offering of one subscription version is planned beside the first. + assert h.failures == [] (key,) = h.bound - assert await h.store.list_generation_obligations(key) == () + assert key.offering_id == OFFERING_ID + assert await h.store.list_generation_obligations(key) != () async def test_the_managed_path_serves_production_through_its_application( diff --git a/tests/test_reporting_storage_facade.py b/tests/test_reporting_storage_facade.py index da623f694..9702434cc 100644 --- a/tests/test_reporting_storage_facade.py +++ b/tests/test_reporting_storage_facade.py @@ -116,12 +116,15 @@ def test_migrations_are_numbered_files_with_their_upgrade_notes() -> None: from adcp_reporting_storage.migrations import MigrationStatus, describe, migrations shipped = migrations() - assert [item.version for item in shipped] == ["0001", "0002", "0003", "0004", "0005"] - widened = shipped[1] + assert [item.version for item in shipped] == ["0001", "0002", "0003", "0004", "0005", "0006"] + widened, offerings = shipped[1], shipped[5] assert widened.name == "reporting_subscription_generations" and widened.stopped_worker assert widened.includes == ("ledger/reporting_ledger_account_generations.sql",) - assert "stop every reporting worker" in widened.description - assert not any(item.stopped_worker for item in shipped if item is not widened) + assert offerings.name == "reporting_status_checkpoint_offerings" and offerings.stopped_worker + assert offerings.includes == ("ledger/reporting_status_checkpoint_offerings.sql",) + for item in (widened, offerings): + assert "stop every reporting worker" in item.description + assert not any(item.stopped_worker for item in shipped if item not in (widened, offerings)) for item in shipped: assert item.statements(), item.version # every include resolves lines = describe([MigrationStatus(item, None) for item in shipped]) From ed7a19fcffb6824f586273dfaf8d3c6d3a9d681a Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:08:08 +0000 Subject: [PATCH 2/2] test(reporting): run the row-upload lease test at the wall clock The producer row-upload test stamps its worker lease with the settling harness clock, fixed at 2026-11-01T02:10Z, while the row store checks the lease against the database's clock_timestamp. Once the wall clock passed that instant plus the lease, every run failed with STATE_UNAVAILABLE: no live worker lease covers this row upload, on next and on every pull request. The test now runs its turn at the wall clock, so the lease lies in the real future. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../reporting/test_reporting_row_storage_object_pg.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py index aa2410045..f4cab4017 100644 --- a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py +++ b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -442,13 +443,18 @@ async def test_the_producer_stages_object_uploads_outside_its_publication_transa from tests.test_reporting_settling import _capabilities, _harness, _revisions async with isolated_reporting_pool() as pool: - producer, store, _, _ = await _harness( + producer, store, _, clock = await _harness( _capabilities(restatement_window="P3D"), store_factory=lambda clock: PgReportingLedgerStore( pool=pool, clock=clock, row_storage=options(tmp_path) ), ) assert isinstance(producer, ReportingProducer) + # The row upload checks the worker lease against the database's + # clock_timestamp(), so the lease the injected clock stamps must lie in + # the real future: run the turn at the wall clock, not the harness's + # fixed instant, which is already in the past. + clock[0] = datetime.now(timezone.utc) turn = await producer.run_worker() assert turn.revisions_committed (revision,) = await _revisions(store)