feat(reporting-storage): deliver reporting subscriptions through the PostgreSQL production store

docs/reliable-reporting-service.md

@@ -522,14 +522,25 @@ Supply `bind_managed_generation(configuration, version)`.
   stays pending under the same identity and is retried.
 - **Requests.** The mounted production handler checks each task against the
   grant and adds `subscription_coverage`.
+- **Stores.** The memory and PostgreSQL production stores both serve managed
+  subscription delivery, with or without status notifications. PostgreSQL
+  keeps each owed set's enrollment, and each owed obligation's frozen
+  constituents, in the private `adcp_reporting_production_owed_sets` and
+  `adcp_reporting_production_owed_constituents` tables. Its reconciliation
+  records, materializer discovery and status checkpoints record the
+  generation's offering. Running `create_schema` with this release lifts
+  those tables' account-only checks and widens the discovery key in place.
 
 **Limits.** These are not served yet:
 
 - subscriptions that resolve a receipt-reconciled (`consumer_receipt`) offering;
-- PostgreSQL production stores for subscriptions (they are refused at startup);
-- status notifications for subscription generations: a PostgreSQL ledger store
-  with them enabled refuses an owed set. `PgReportingLedgerStore` otherwise holds
-  owed sets, after the generation-key widening described in
+- more than one offering of one subscription version for the same account and
+  owner under production. Status checkpoints hold one configuration scope per
+  version, so admitting a second offering fails that subscription's planning
+  with `reporting_production_offering_conflict`;
+- status notifications for subscription generations outside production: a
+  `PgReportingLedgerStore` with them enabled refuses an owed set. It otherwise
+  holds owed sets, after the generation-key widening described in
   [ledger migrations](reporting-ledger-migration.md).
 
 Keep the engine's `status_retention_days` equal to the service's.

docs/reporting-ledger-migration.md

@@ -68,10 +68,19 @@ In PostgreSQL this is a **stopped-worker maintenance upgrade**:
    - `reporting_consumer_statuses` gains `offering_id`;
    - the SDK's own references (reconciliation records, materializer discovery,
      status scope checkpoints, production generations) are re-pointed at the
-     widened key. They hold account-level generations only, enforced by a
-     `CHECK`, until each feature serves subscriptions.
+     widened key. Reconciliation records, materializer discovery and status
+     scope checkpoints also hold subscription generations, for managed
+     delivery. Production generations hold account-level generations only,
+     enforced by a `CHECK`.
 3. Restart the workers on the new release.
 
+A schema created by a release that widened the key but did not yet serve
+managed subscription delivery still carries an account-only `CHECK` on
+reconciliation records, materializer discovery and status scope checkpoints,
+and a four-column materializer discovery key. `create_schema()` drops those
+checks and widens the discovery key in place, losslessly and idempotently.
+Earlier workers refuse the upgraded schema, so stop them first here too.
+
 A foreign key outside the SDK that references `reporting_configurations`
 refuses the upgrade with a maintenance error, rather than being dropped.
 Re-point it at the widened key, then run the upgrade. The upgrade is

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/delivery_pg.py

@@ -75,6 +75,7 @@
     "receipt_chain_key",
     "receipt_status",
     "supersedes_receipt_id",
+    "offering_id",
 )
 _SELECT_IDENTITY = ", ".join("r." + column for column in _IDENTITY_COLUMNS)
 _FEED_JOIN = (
@@ -377,9 +378,9 @@ async def _insert(self, connection: Any, record: ReportingDeliveryRecord) -> Non
             " (account_id, consumer_id, namespace, record_id, record_kind, delivery_config_id,"
             " delivery_config_version, reporting_obligation_id, reporting_revision_id,"
             " reporting_materialization_id, reporting_adjustment_id, attempt_number,"
-            " receipt_chain_key, receipt_status, supersedes_receipt_id, payload,"
+            " receipt_chain_key, receipt_status, supersedes_receipt_id, offering_id, payload,"
             " content_sha256, change_id)"
-            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,"
+            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,"
             " %s::jsonb, %s, %s)",
             (
                 *storage_identity(record),

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py

@@ -929,11 +929,19 @@ async def _lock_account(connection: Any, account_id: str) -> None:
 
     # -- configurations ---------------------------------------------------
 
+    def _status_covers_owed_sets(self) -> bool:
+        """Whether status notifications may cover reporting-subscription generations.
+
+        A status checkpoint records its generation's offering but is keyed per
+        (account, consumer, delivery_config_id, version): two offerings of one
+        subscription version would share a configuration checkpoint. Only a
+        store that admits at most one offering per version may serve them.
+        """
+        return not self._notifications_enabled
+
     async def put_configuration(self, configuration: ReportingConfiguration) -> None:
         reject_reserved_authoritative_party(configuration)
-        if configuration.owed_set and self._notifications_enabled:
-            # Status checkpoints are keyed per account-level generation; they
-            # do not yet hold one per resolved offering.
+        if configuration.owed_set and not self._status_covers_owed_sets():
             raise LedgerConflictError(
                 "UNSUPPORTED_FEATURE",
                 "status notifications do not yet cover reporting-subscription generations",

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql

@@ -91,13 +91,13 @@ BEGIN
     ALTER TABLE reporting_consumer_statuses
         ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '';
 
-    -- These tables serve account-level generations only; a subscription
-    -- generation is refused by their CHECK until each feature serves it.
+    -- Managed delivery records subscription generations under their offering
+    -- (reconciliation records, materializer discovery and status checkpoints).
+    -- Production generations serve account-level generations only; a
+    -- subscription generation is refused by their CHECK.
     IF to_regclass('reporting_reconciliation_records') IS NOT NULL THEN
         ALTER TABLE reporting_reconciliation_records
-            ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''
-                CONSTRAINT reporting_reconciliation_records_account_generation
-                CHECK (offering_id = ''),
+            ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
             ADD CONSTRAINT reporting_reconciliation_records_generation_fkey
                 FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version,
                              offering_id)
@@ -106,9 +106,7 @@ BEGIN
     END IF;
     IF to_regclass('reporting_materializer_discovery') IS NOT NULL THEN
         ALTER TABLE reporting_materializer_discovery
-            ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''
-                CONSTRAINT reporting_materializer_discovery_account_generation
-                CHECK (offering_id = ''),
+            ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
             ADD CONSTRAINT reporting_materializer_discovery_generation_fkey
                 FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version,
                              offering_id)
@@ -117,9 +115,7 @@ BEGIN
     END IF;
     IF to_regclass('reporting_status_scope_checkpoints') IS NOT NULL THEN
         ALTER TABLE reporting_status_scope_checkpoints
-            ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''
-                CONSTRAINT reporting_status_scope_checkpoints_account_generation
-                CHECK (offering_id = ''),
+            ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
             ADD CONSTRAINT reporting_status_scope_checkpoints_generation_fkey
                 FOREIGN KEY (account_id, consumer_namespace, delivery_config_id, version,
                              offering_id)

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql

@@ -108,10 +108,9 @@ BEGIN
         CHECK ((receipt_status IS NOT NULL) = (receipt_chain_key IS NOT NULL)),
         CHECK ((record_kind = 'materialization_attempt') = (attempt_number IS NOT NULL)),
         CHECK (payload->>'kind' = record_kind),
-        -- Account-level generations only, until managed delivery serves
-        -- reporting subscriptions (see reporting_ledger_account_generations.sql).
-        offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''
-            CONSTRAINT reporting_reconciliation_records_account_generation CHECK (offering_id = ''),
+        -- '' for an account-level generation; a reporting subscription's
+        -- generation names its resolved offering (managed delivery).
+        offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
         CONSTRAINT reporting_reconciliation_records_generation_fkey
             FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id)
             REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id,
@@ -132,6 +131,14 @@ BEGIN
     CREATE UNIQUE INDEX IF NOT EXISTS reporting_receipt_one_successor
         ON reporting_reconciliation_records(account_id, consumer_id, supersedes_receipt_id)
         WHERE supersedes_receipt_id IS NOT NULL;
+    -- A schema created by an earlier release refused subscription generations
+    -- here; managed delivery now records them under their offering.
+    IF EXISTS (SELECT 1 FROM pg_constraint
+        WHERE conrelid = 'reporting_reconciliation_records'::regclass
+          AND conname = 'reporting_reconciliation_records_account_generation') THEN
+        ALTER TABLE reporting_reconciliation_records
+            DROP CONSTRAINT IF EXISTS reporting_reconciliation_records_account_generation;
+    END IF;
 
     CREATE TABLE IF NOT EXISTS reporting_receipt_heads (
         account_id TEXT COLLATE "C" NOT NULL,
@@ -432,8 +439,10 @@ BEGIN
             RETURN;
         END IF;
         SELECT b.payload INTO binding FROM reporting_reconciliation_records b
-        WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version)
-            = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version)
+        WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version,
+               b.offering_id)
+            = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version,
+               r.offering_id)
           AND b.record_kind = 'destination_binding';
         SELECT d.payload INTO delivery FROM reporting_reconciliation_records d
         WHERE (d.account_id, d.consumer_id, d.delivery_config_id, d.delivery_config_version,
@@ -527,12 +536,17 @@ BEGIN
             'verification_path', 'verification_profile', 'verified_at', 'verified_format'])) THEN
             RAISE EXCEPTION 'reporting payload is not closed retained evidence' USING ERRCODE = '23514';
         END IF;
+        -- An account-level generation key omits offering_id (ledger_asdict).
         generation := jsonb_build_object('account_id', r.account_id, 'consumer_id', r.consumer_id,
             'delivery_config_id', r.delivery_config_id,
-            'delivery_config_version', r.delivery_config_version);
+            'delivery_config_version', r.delivery_config_version)
+            || CASE WHEN r.offering_id = '' THEN '{}'::jsonb
+                    ELSE jsonb_build_object('offering_id', r.offering_id) END;
         IF NOT EXISTS (SELECT 1 FROM reporting_configurations c
-            WHERE (c.account_id, c.consumer_id, c.delivery_config_id, c.delivery_config_version)
-                = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version))
+            WHERE (c.account_id, c.consumer_id, c.delivery_config_id, c.delivery_config_version,
+                   c.offering_id)
+                = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version,
+                   r.offering_id))
            OR (r.reporting_obligation_id IS NOT NULL AND NOT EXISTS (
                SELECT 1 FROM reporting_obligations o
                WHERE (o.account_id, o.consumer_id, o.delivery_config_id, o.delivery_config_version, o.reporting_obligation_id)
@@ -552,7 +566,9 @@ BEGIN
             WHEN 'destination_binding' THEN reporting_identity_sha256(
                 '{"account_id":' || to_json(r.account_id)::text || ',"consumer_id":' || to_json(r.consumer_id)::text || ',"delivery_config_id":' ||
                 to_json(r.delivery_config_id)::text || ',"delivery_config_version":' ||
-                r.delivery_config_version::text || '}')
+                r.delivery_config_version::text ||
+                CASE WHEN r.offering_id = '' THEN ''
+                     ELSE ',"offering_id":' || to_json(r.offering_id)::text END || '}')
             WHEN 'obligation_delivery' THEN r.reporting_obligation_id
             WHEN 'materialization_check' THEN r.payload->>'check_id'
             WHEN 'revision_receipt' THEN r.payload->>'reporting_receipt_id'
@@ -594,8 +610,10 @@ BEGIN
                 'consumer_id', r.consumer_id, 'reporting_obligation_id', r.reporting_obligation_id)
            OR NOT EXISTS (
                 SELECT 1 FROM reporting_reconciliation_records b
-                WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version)
-                    = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version)
+                WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version,
+                       b.offering_id)
+                    = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version,
+                       r.offering_id)
                   AND b.record_kind = 'destination_binding') THEN
             RAISE EXCEPTION 'reporting frozen binding is unavailable' USING ERRCODE = '23514';
         END IF;
@@ -622,8 +640,10 @@ BEGIN
         END IF;
         IF r.record_kind = 'materialization' AND r.payload->>'status' IN ('available', 'delivered') THEN
             IF EXISTS (SELECT 1 FROM reporting_reconciliation_records b
-                WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version)
-                    = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version)
+                WHERE (b.account_id, b.consumer_id, b.delivery_config_id, b.delivery_config_version,
+                       b.offering_id)
+                    = (r.account_id, r.consumer_id, r.delivery_config_id, r.delivery_config_version,
+                       r.offering_id)
                   AND b.record_kind = 'destination_binding'
                   AND (r.payload->>'status' IS DISTINCT FROM b.payload->>'success_status'
                     OR r.payload#>>'{verification,verification_profile}' IS DISTINCT FROM b.payload->>'verification_profile')) THEN
@@ -740,7 +760,8 @@ BEGIN
             proposed := proposed - mutable;
             SELECT EXISTS (SELECT 1 FROM reporting_reconciliation_records r
                 WHERE r.account_id=OLD.account_id AND r.consumer_id=OLD.consumer_id AND r.delivery_config_id = OLD.delivery_config_id
-                    AND r.delivery_config_version = OLD.delivery_config_version) INTO referenced;
+                    AND r.delivery_config_version = OLD.delivery_config_version
+                    AND r.offering_id = OLD.offering_id) INTO referenced;
         ELSIF TG_TABLE_NAME = 'reporting_obligations' THEN
             SELECT EXISTS (SELECT 1 FROM reporting_reconciliation_records r
                 WHERE r.account_id = OLD.account_id AND r.reporting_obligation_id = OLD.reporting_obligation_id) INTO referenced;

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_materializer.sql

@@ -23,14 +23,32 @@ BEGIN
         delivery_config_version BIGINT NOT NULL CHECK (delivery_config_version > 0),
         after_obligation_id TEXT COLLATE "C" NOT NULL DEFAULT '',
         complete BOOLEAN NOT NULL DEFAULT FALSE,
-        offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''
-            CONSTRAINT reporting_materializer_discovery_account_generation CHECK (offering_id = ''),
-        PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version),
+        -- '' for an account-level generation; a reporting subscription's
+        -- generation names its resolved offering (managed delivery).
+        offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
+        PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version,
+            offering_id),
         CONSTRAINT reporting_materializer_discovery_generation_fkey
             FOREIGN KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id)
             REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id,
                                                 delivery_config_version, offering_id)
     );
+    -- A schema created by an earlier release keyed discovery without the
+    -- offering and refused subscription generations: widen it losslessly.
+    IF EXISTS (SELECT 1 FROM pg_constraint
+        WHERE conrelid = 'reporting_materializer_discovery'::regclass
+          AND conname = 'reporting_materializer_discovery_account_generation') THEN
+        ALTER TABLE reporting_materializer_discovery
+            DROP CONSTRAINT IF EXISTS reporting_materializer_discovery_account_generation;
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_constraint pk
+        WHERE pk.conrelid = 'reporting_materializer_discovery'::regclass AND pk.contype = 'p'
+          AND array_length(pk.conkey, 1) = 4) THEN
+        ALTER TABLE reporting_materializer_discovery
+            DROP CONSTRAINT reporting_materializer_discovery_pkey,
+            ADD CONSTRAINT reporting_materializer_discovery_pkey PRIMARY KEY
+                (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id);
+    END IF;
     CREATE INDEX IF NOT EXISTS reporting_materializer_discovery_pending
         ON reporting_materializer_discovery (account_id, consumer_id, delivery_config_id,
             delivery_config_version) WHERE NOT complete;
@@ -295,6 +313,7 @@ BEGIN
                 WHERE account_id = NEW.account_id AND consumer_id = NEW.consumer_id
                 AND delivery_config_id = NEW.delivery_config_id
                     AND delivery_config_version = NEW.delivery_config_version
+                    AND offering_id = NEW.offering_id
                 ORDER BY consumer_id
             LOOP
                 PERFORM reporting_materializer_dirty(NEW.account_id, r.consumer_id,
@@ -312,8 +331,9 @@ BEGIN
         IF NEW.record_kind = 'destination_binding' THEN
             PERFORM pg_advisory_xact_lock(hashtext('adcp.reporting:' || NEW.account_id));
             INSERT INTO reporting_materializer_discovery
-                (account_id, consumer_id, delivery_config_id, delivery_config_version)
-            VALUES (NEW.account_id, NEW.consumer_id, NEW.delivery_config_id, NEW.delivery_config_version)
+                (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id)
+            VALUES (NEW.account_id, NEW.consumer_id, NEW.delivery_config_id,
+                    NEW.delivery_config_version, NEW.offering_id)
             ON CONFLICT DO NOTHING;
             PERFORM reporting_materializer_wake(NEW.account_id);
         ELSIF NEW.record_kind = 'materialization_check' THEN
@@ -412,8 +432,8 @@ BEGIN
     -- One-time indexed backfill seeds bindings, not account enumeration or
     -- recurring whole-ledger scans. Reinstall never rewinds a discovery cursor.
     INSERT INTO reporting_materializer_discovery
-        (account_id, consumer_id, delivery_config_id, delivery_config_version)
-    SELECT account_id, consumer_id, delivery_config_id, delivery_config_version
+        (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id)
+    SELECT account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id
         FROM reporting_reconciliation_records WHERE record_kind='destination_binding'
     ON CONFLICT DO NOTHING;
     INSERT INTO reporting_materializer_accounts (account_id, due_at)