fix(reporting): key status checkpoints by resolved offering

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.
 

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

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
 

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)

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