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_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) 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])