From d0aadc37d5472344e251487a20cfb629438d340b Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 11:09:07 +0000 Subject: [PATCH 1/4] feat(reporting-storage): fence the orphan collector against a cloned or restored ledger database A clone or point-in-time restore shares the live database's installation identity, namespace key and configuration (including acknowledgeExclusiveNamespace), so its collector would delete the live database's newer row objects. A real collectOrphanedRowObjects pass now claims the namespace first: a create-only chain of epoch objects under {prefix}/{namespace_key}/_collector-fence/ plus a row in the new adcp_reporting_orphan_collector_fences table (migration 0009). It deletes only while the highest epoch in the bucket is exactly the claim the database recorded, re-checks before every deleting page, and reports a refused binding as fenced with a cause. dryRun checks without claiming. takeOver OrphanCollectorFence is the deliberate takeover. Closes #3292 Co-Authored-By: Claude Sonnet 5.5 --- .changeset/orphan-collector-clone-fence.md | 21 + docs/guides/REPORTING-OPERATIONS.md | 53 +- docs/migration-reporting-storage-package.md | 12 + packages/reporting-storage/README.md | 1 + .../reporting-storage/scripts/run-tests.mjs | 1 + packages/reporting-storage/src/index.ts | 1 + packages/reporting-storage/src/maintenance.ts | 24 +- packages/reporting-storage/src/migrations.ts | 8 +- .../reporting-storage/src/orphan-fence.ts | 276 ++++++++ packages/reporting-storage/src/postgres.ts | 319 ++++++++- .../src/row-storage-migration.ts | 23 + packages/reporting-storage/src/row-storage.ts | 224 +++++++ .../reporting-maintenance-warnings.test.js | 29 + ...eporting-orphan-collector-fence-pg.test.js | 634 ++++++++++++++++++ .../reporting-row-orphan-collector-pg.test.js | 5 + 15 files changed, 1615 insertions(+), 16 deletions(-) create mode 100644 .changeset/orphan-collector-clone-fence.md create mode 100644 packages/reporting-storage/src/orphan-fence.ts create mode 100644 packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js diff --git a/.changeset/orphan-collector-clone-fence.md b/.changeset/orphan-collector-clone-fence.md new file mode 100644 index 000000000..336125e39 --- /dev/null +++ b/.changeset/orphan-collector-clone-fence.md @@ -0,0 +1,21 @@ +--- +'@adcp/reporting-storage': minor +--- + +Reporting storage (experimental): the orphaned-object collector is fenced against a cloned or restored ledger database (#3292). + +A clone or point-in-time restore of the ledger database shares the live database's installation identity, namespace key and configuration, so it also shares `acknowledgeExclusiveNamespace: true`. Its collector saw an older set of references and would have deleted the live database's newer row objects. A real `collectOrphanedRowObjects` pass now claims the namespace first and refuses to delete when another database holds a newer claim. + +- **Fence.** A chain of create-only objects, one per epoch, under `{prefix}/{namespace_key}/_collector-fence/`, each holding a random token. The database records its last claim in `adcp_reporting_orphan_collector_fences`. A pass deletes only while the highest epoch in the bucket is exactly the claim the database recorded; it claims the next epoch before it deletes anything and re-reads the fence before every page that deletes. A copy taken before the original's latest claim sees a higher epoch and refuses. +- **Reporting.** A refused binding has `status: 'fenced'` and `fence: { cause, detail, fenceEpoch?, recordedEpoch? }`; the result has a new required `fenced: string[]`, and the maintenance task raises a warning. Causes: `fence_advanced`, `claimed_elsewhere`, `token_mismatch`, `fence_missing`, `fence_regressed`, `fence_unreadable`, `claim_lost`, `migration_required`. `dryRun` reads the fence without claiming it and reports fenced bindings too. +- **Takeover.** New `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, bindings?, includeReadOnly? })` for a database restored in place or whose predecessor is gone. It claims the next epoch and fences every other copy out. +- **Limit.** Nothing in the database tells a copy from the original until one of them claims, so the first claim wins. Run orphan collection only on the live database, never on a copy of the deployment. + +Migration note (minor, breaking only for orphan collection): + +1. Run `npx adcp-reporting-storage migrate`: new version `0009`, `reporting_orphan_collector_fence` (additive; the SQL is exported as `REPORTING_ORPHAN_FENCE_MIGRATION`). Until it is applied, a pass reports every binding as `fenced` with cause `migration_required` and deletes nothing, and `postgresStorage().checkSchema()` names `0009` as pending. +2. Run one real pass on the live database before any staging copy runs a collector. +3. A custom store that implements `collectOrphanedRowObjects` for the maintenance task must return `fenced: []`. `ReportingOrphanBindingResultV1.status` gains `'fenced'`. +4. Fence objects live in your object-store prefix. Provider contract is unchanged (create-only `putIfAbsent`, `get`, `list`, `delete`). + +See `docs/guides/REPORTING-OPERATIONS.md` and `docs/migration-reporting-storage-package.md`. diff --git a/docs/guides/REPORTING-OPERATIONS.md b/docs/guides/REPORTING-OPERATIONS.md index 4c613352d..257a49088 100644 --- a/docs/guides/REPORTING-OPERATIONS.md +++ b/docs/guides/REPORTING-OPERATIONS.md @@ -198,12 +198,51 @@ service runs for you. See of them. The age threshold does not protect against that, which is why a real (not dry) pass needs `acknowledgeExclusiveNamespace: true`: it is your statement that this database is the only owner. A clone made by copying the deployment - copies the flag too, so the flag does not protect it: remove it, or the whole - option, from every copy. Disable the option on clones, and after - restoring a database in place expect the collector to delete the objects of the - row sets the restore lost, which is its purpose. + copies the flag too, so the flag alone does not protect it, and a persisted + **ownership fence** backs it. It needs migration `0009` + (`npx adcp-reporting-storage migrate`); until then every real pass and dry run + reports each binding as fenced with cause `migration_required` and deletes + nothing. + + - **Claim.** The first real pass claims the namespace: it records + `(epoch, token)` in `adcp_reporting_orphan_collector_fences` and writes a + create-only object, `{prefix}/{namespace_key}/_collector-fence/{epoch}.json`, + for that epoch. Every later real pass claims the next epoch before it + deletes anything, and prunes the claims older than the previous one. + - **Check.** A pass deletes only while the highest epoch in the bucket is + exactly the claim this database recorded. It re-reads the fence before every + page that deletes. Anything else refuses, deletes nothing and reports the + binding as `fenced` with a `cause` (`fence_advanced`, `claimed_elsewhere`, + `token_mismatch`, `fence_missing`, `fence_regressed`, `fence_unreadable`, + `claim_lost`, `migration_required`), in the result's `fenced` list and as a + maintenance warning. Bindings that share a bucket, prefix and installation + share one fence. + - **Clones and restores.** A copy that still holds an older claim sees the + live database's newer epoch and refuses. So does a database restored to an + earlier state. A `dryRun` reads the fence without claiming it, so it + reports a fenced binding too and writes nothing. + - **First claim wins.** Nothing in the database distinguishes a copy from the + original until one of them claims. If a clone collects before the original + has run a pass since the copy was taken, the clone claims first and the + original is fenced, and the warning is the signal. The age threshold still + holds: the clone cannot delete an object the original wrote after the copy + until that object is `minAgeSeconds` old, so turn the clone's collector off + within that window. Run orphan collection only on the live database, and + never copy the deployment with the option on. + - **Takeover.** After restoring a database in place, or when the original is + gone, call `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true })` + once. It claims the next epoch for this database and fences every other + copy out. It refuses without the acknowledgement. Do not call it from a + clone while the original is running: that fences the original and lets the + clone delete its objects. + + After a restore in place, expect the collector to delete the objects of the + row sets the restore lost once you have taken the namespace over, which is its + purpose. The cost grows with bucket size: one listing request per page, plus one - database query per page and one transaction per deleted object. + database query per page and one transaction per deleted object. The fence + adds a small listing and a read per pass, and one more before each deleting + page. Alert on these row-storage signals: @@ -216,6 +255,10 @@ Alert on these row-storage signals: host does not configure, and `overdue` (expired more than 24 hours ago, still unswept) means no host is handling them and their objects are leaking. The maintenance helper reports both through `onError`. +- The `orphan object collection` task reporting `fenced` bindings: another + database holds the namespace's collector fence, or this one cannot prove it + does. The `fence.cause` says why; see the ownership fence above before you + take the namespace over. - The `orphan object collection` task reporting `failed` or `unsupported` bindings (their provider cannot list, so nothing there is collected). A listing that fails is retried from the page that failed (or from the start, if it was diff --git a/docs/migration-reporting-storage-package.md b/docs/migration-reporting-storage-package.md index 6c76f215d..237314d34 100644 --- a/docs/migration-reporting-storage-package.md +++ b/docs/migration-reporting-storage-package.md @@ -168,6 +168,18 @@ Reporting storage is now one decision. `postgresStorage({ pool })` (this package The migration constants (`REPORTING_LEDGER_MIGRATION`, `REPORTING_ROW_STORAGE_MIGRATION`, `REPORTING_LEDGER_CHANGES_MIGRATION`, `REPORTING_MANAGED_DELIVERY_MIGRATION`, the object-write and finality-fence constants) are deprecated re-exports for one beta. A schema they set up migrates cleanly: the first `migrate` re-runs them as no-ops and records versions `0001`–`0008`. See the [package README](https://github.com/adcontextprotocol/adcp-client/blob/main/packages/reporting-storage/README.md#apply-migrations-npx-adcp-reporting-storage-migrate) for the version table. +### Orphan collector ownership fence (storage `0.3`, migration `0009`) + +Only deployments that enable `orphanCollection` (or call `collectOrphanedRowObjects`) are affected. A clone or restore of the ledger database shares the live database's namespace key and configuration, including `acknowledgeExclusiveNamespace: true`, and would delete the live database's newer row objects. A real pass now claims the namespace first and refuses to delete when another database holds a newer claim. + +1. Run `npx adcp-reporting-storage migrate` (version `0009`, `reporting_orphan_collector_fence`, additive). Until it is applied, every pass, `dryRun` included, reports each binding as `fenced` with cause `migration_required` and deletes nothing. `postgresStorage().checkSchema()` names `0009` as pending. +2. Roll `0009` out and run one pass on the **live** database before any staging copy of it runs a collector. The first claim wins, and a copy has no way to tell it is a copy. +3. Handle the new result shape: `ReportingOrphanBindingResultV1.status` can be `'fenced'` (with `fence: { cause, detail, fenceEpoch?, recordedEpoch? }`), and `ReportingOrphanCollectionResultV1` has a required `fenced: string[]`. A custom store that implements `collectOrphanedRowObjects` for the maintenance task must return `fenced` (an empty array is fine). The maintenance task raises a warning for each fenced binding. +4. After restoring a database in place, or when the database that held the namespace is gone, call `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true })` once, then collection resumes. It refuses without the acknowledgement. +5. The fence chain lives under `{prefix}/{namespace_key}/_collector-fence/` in each object binding's bucket. Your provider needs nothing new (create-only `putIfAbsent`, `get`, `list` and `delete`, which the conformance suite already covers), and a bucket lifecycle rule must not expire objects under the owned prefix (the probe already refuses one that does). + +See [Orphaned object collection](guides/REPORTING-OPERATIONS.md#adopting-row-storage-retention-and-the-change-feed) for the claim, check and first-claim-wins rules. + ### `service.setup.migrations` comes from the store `createReliableReportingService` no longer names the PostgreSQL migration. `service.setup.migrations` is the store's own `migrations` list (`ReportingLedgerStore.migrations`, optional). `PostgresReportingLedgerStore` lists `[REPORTING_LEDGER_MIGRATION]`. A store with nothing to install (an in-memory test double), or a decorator over the PostgreSQL store that does not forward `migrations`, gives an empty list. diff --git a/packages/reporting-storage/README.md b/packages/reporting-storage/README.md index adf20778f..5562bc2ab 100644 --- a/packages/reporting-storage/README.md +++ b/packages/reporting-storage/README.md @@ -93,6 +93,7 @@ SDK's `python -m adcp migrate` uses), so concurrent runners serialize. `migrate( | 0006 | `reporting_managed_delivery` | managed delivery (`REPORTING_MANAGED_DELIVERY_MIGRATION`) | | 0007 | `reporting_object_writes` | the GCS write fence (`REPORTING_OBJECT_WRITE_MIGRATION`, `REPORTING_OBJECT_WRITE_AUTHORITY_MIGRATION`) | | 0008 | `reporting_consumer` | the buyer consumer (`REPORTING_CONSUMER_POSTGRES_MIGRATION` from `@adcp/sdk/reporting/consumer`) | +| 0009 | `reporting_orphan_collector_fence` | the orphan collector's ownership fence (`REPORTING_ORPHAN_FENCE_MIGRATION`); needed before orphan collection can delete | A schema set up by applying those constants by hand migrates cleanly: the statements are idempotent, re-run as no-ops, and the versions are recorded. The constants stay exported as deprecated re-exports for one beta. diff --git a/packages/reporting-storage/scripts/run-tests.mjs b/packages/reporting-storage/scripts/run-tests.mjs index 6b12b0795..053404495 100644 --- a/packages/reporting-storage/scripts/run-tests.mjs +++ b/packages/reporting-storage/scripts/run-tests.mjs @@ -40,6 +40,7 @@ export const SERIAL_TESTS = [ 'reporting-gcp-validation-pg.test.js', 'reporting-head-of-line-pg.test.js', 'reporting-host-reads-pg.test.js', + 'reporting-orphan-collector-fence-pg.test.js', 'reporting-row-orphan-collector-pg.test.js', 'reporting-row-retention-pg.test.js', ]; diff --git a/packages/reporting-storage/src/index.ts b/packages/reporting-storage/src/index.ts index 89c2c7077..77c6b21c8 100644 --- a/packages/reporting-storage/src/index.ts +++ b/packages/reporting-storage/src/index.ts @@ -15,6 +15,7 @@ export * from './postgres'; export * from './row-storage-migration'; export * from './row-storage'; export * from './row-storage-object'; +export type { ReportingOrphanFenceCauseV1, ReportingOrphanFenceRefusalV1 } from './orphan-fence'; export * from './object-writes'; export * from './change-feed'; export * from './maintenance'; diff --git a/packages/reporting-storage/src/maintenance.ts b/packages/reporting-storage/src/maintenance.ts index e60ee2c4b..0e77beb1e 100644 --- a/packages/reporting-storage/src/maintenance.ts +++ b/packages/reporting-storage/src/maintenance.ts @@ -43,8 +43,11 @@ export interface ReportingCompactionOptionsV1 { * * Enable it only on the ledger database that owns the namespace. A clone or restore of that * database that still points at the same bucket shares its namespace and would delete the objects - * of every row set the original wrote after the clone was taken. Start with `dryRun: true` and read - * `candidates` in the task result. See `docs/guides/REPORTING-OPERATIONS.md`. + * of every row set the original wrote after the clone was taken, so a real pass is fenced: it + * claims the namespace (migration `0009`) and refuses, reporting the binding as `fenced`, when the + * object store holds a newer claim than this database recorded. After a restore, or when the + * original is gone, call `takeOverOrphanCollectorFence` on the store. Start with `dryRun: true` and + * read `candidates` in the task result. See `docs/guides/REPORTING-OPERATIONS.md`. * * @experimental Reliable Reporting storage surface: breaking changes may ship in minor releases. See https://github.com/adcontextprotocol/adcp-client/blob/main/docs/guides/REPORTING-STABILITY.md */ @@ -54,7 +57,8 @@ export interface ReportingOrphanCollectionOptionsV1 { dryRun?: boolean; /** * Required unless `dryRun`: your statement that this ledger database is the only one that owns the - * namespace, and not a clone or restore pointed at the same bucket. + * namespace, and not a clone or restore pointed at the same bucket. A copy of the configuration carries + * this flag too; the ownership fence is what stops such a copy. */ acknowledgeExclusiveNamespace?: boolean; /** @@ -307,6 +311,7 @@ export async function runReportingLedgerMaintenance( failed: 0, unsupported: [], skippedReadOnly: [], + fenced: [], candidates: [], cursors: {}, completed: [], @@ -372,7 +377,12 @@ export async function runReportingLedgerMaintenance( overdue?: number; // Orphan object collection: bindings it could not cover. unsupported?: readonly string[]; - bindings?: readonly { bindingId: string; status: string; error?: string }[]; + bindings?: readonly { + bindingId: string; + status: string; + error?: string; + fence?: { cause: string; detail: string }; + }[]; } | undefined; const failed = Array.isArray(value?.failed) @@ -411,6 +421,12 @@ export async function runReportingLedgerMaintenance( if (binding.status === 'failed') { warnings.push(`orphan collection failed for row binding ${binding.bindingId} (${binding.error ?? 'unknown'})`); } + if (binding.status === 'fenced') { + warnings.push( + `orphan collection refused to delete for row binding ${binding.bindingId}: its namespace is fenced ` + + `(${binding.fence?.cause ?? 'unknown'}). ${binding.fence?.detail ?? ''}`.trim() + ); + } } let partialFailure: ReportingMaintenancePartialFailureError | undefined; if (failed > 0 || warnings.length > 0) { diff --git a/packages/reporting-storage/src/migrations.ts b/packages/reporting-storage/src/migrations.ts index 86a621ccc..ede7f90cf 100644 --- a/packages/reporting-storage/src/migrations.ts +++ b/packages/reporting-storage/src/migrations.ts @@ -25,7 +25,7 @@ import { REPORTING_LEDGER_MIGRATION, type ReportingPgPool, } from './postgres'; -import { REPORTING_ROW_STORAGE_MIGRATION } from './row-storage-migration'; +import { REPORTING_ORPHAN_FENCE_MIGRATION, REPORTING_ROW_STORAGE_MIGRATION } from './row-storage-migration'; import { REPORTING_MANAGED_DELIVERY_MIGRATION } from './managed/postgres'; /** @@ -124,6 +124,12 @@ const MIGRATIONS: readonly Migration[] = Object.freeze([ "The buyer consumer's checkpoints, pending statuses, change cursors, leases and processed notifications. Online-safe.", [REPORTING_CONSUMER_POSTGRES_MIGRATION] ), + migration( + '0009', + 'reporting_orphan_collector_fence', + "The orphan collector's ownership fence: one row per fenced namespace. Additive. Until applied, orphan collection refuses to delete.", + [REPORTING_ORPHAN_FENCE_MIGRATION] + ), ]); const TABLE = `CREATE TABLE IF NOT EXISTS adcp_schema_migrations ( diff --git a/packages/reporting-storage/src/orphan-fence.ts b/packages/reporting-storage/src/orphan-fence.ts new file mode 100644 index 000000000..419a30064 --- /dev/null +++ b/packages/reporting-storage/src/orphan-fence.ts @@ -0,0 +1,276 @@ +/** + * The orphan collector's ownership fence (adcontextprotocol/adcp-client#3292). + * + * The collector deletes objects under `{prefix}/{namespace_key}/` that no ledger record names. A clone or + * point-in-time restore of the ledger database carries the same configuration, installation identity and + * namespace key, but an older set of references, so its collector would delete the live database's newer + * objects. Nothing inside the database can tell the two apart; the object store they share can. + * + * The fence is a chain of create-only objects, one per epoch, under the binding's owned prefix: + * + * {prefix}/{namespace_key}/_collector-fence/0000000000000007.json + * + * Each holds `{ v, namespace_key, epoch, token, previous_token, claimed_at }`. The highest epoch is the + * current claim. The database records the `(epoch, token)` it last claimed in + * `adcp_reporting_orphan_collector_fences`. A real pass deletes only when the highest epoch in the store is + * exactly the claim this database recorded, then claims the next epoch (create-only, so exactly one claimant + * wins) before it deletes anything. Every other state refuses with a cause. A database that was copied + * before the original's last claim sees a higher epoch and refuses; so does a restore, until an operator + * calls `takeOverOrphanCollectorFence`. + */ +import { createHash } from 'crypto'; + +import { canonicalize } from '@adcp/sdk'; + +/** + * Object-store directory (under the owned prefix) that holds the fence chain. + * + * @internal + */ +export const ORPHAN_FENCE_DIRECTORY = '_collector-fence'; + +/** Fence objects are tiny; anything larger is not one. */ +export const ORPHAN_FENCE_MAX_BYTES = 4096; + +const EPOCH_DIGITS = 16; +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; +const KEY_NAME = new RegExp(`^(\\d{${EPOCH_DIGITS}})\\.json$`); + +/** + * Why a binding's collector refused to delete. + * + * - `claimed_elsewhere`: this database never claimed the namespace, but the object store holds a claim. A + * clone that has not collected before, a database whose fence table was lost, or another installation. + * - `fence_advanced`: the object store holds a newer claim than this database recorded. Another database + * (usually the live one this was cloned from) has collected since this database's state was taken. + * - `token_mismatch`: the object store holds this database's epoch under a different token. Two databases + * claimed the same epoch; the other one won. + * - `fence_missing`: this database recorded a claim and the object store holds none: the bucket or prefix + * was reset or is a different one. + * - `fence_regressed`: the object store holds an older claim than this database recorded: a database + * restored from a newer state, or a bucket restored from an older one. + * - `fence_unreadable`: the fence objects are malformed, oversized, duplicated or name another namespace. + * - `claim_lost`: this pass tried to claim the next epoch and another database got there first. + * - `migration_required`: the fence table does not exist; apply the pending reporting-storage migration. + * + * @experimental Reliable Reporting storage surface: breaking changes may ship in minor releases. See https://github.com/adcontextprotocol/adcp-client/blob/main/docs/guides/REPORTING-STABILITY.md + */ +export type ReportingOrphanFenceCauseV1 = + | 'claimed_elsewhere' + | 'fence_advanced' + | 'token_mismatch' + | 'fence_missing' + | 'fence_regressed' + | 'fence_unreadable' + | 'claim_lost' + | 'migration_required'; + +/** + * Why a binding was fenced, reported on its collection outcome. + * + * @experimental Reliable Reporting storage surface: breaking changes may ship in minor releases. See https://github.com/adcontextprotocol/adcp-client/blob/main/docs/guides/REPORTING-STABILITY.md + */ +export interface ReportingOrphanFenceRefusalV1 { + cause: ReportingOrphanFenceCauseV1; + /** What to check, and how to take the namespace over if this database really is its only owner. */ + detail: string; + /** Highest epoch found in the object store, when readable. */ + fenceEpoch?: number; + /** Epoch this database last recorded, when it has one. */ + recordedEpoch?: number; +} + +/** One fence object found by a listing. */ +export interface OrphanFenceEntry { + epoch: number; + objectRef: string; + nativeVersionRef: string; +} + +/** The contents of one fence object. */ +export interface OrphanFenceRecord { + epoch: number; + token: string; + previousToken: string | null; + claimedAt: string; +} + +/** What this database recorded for a fence scope. */ +export interface OrphanFenceRow { + epoch: number; + token: string; + previousToken: string | null; +} + +export type OrphanFenceDecision = + /** + * This database may collect. A real pass claims `epoch` first (the next epoch after its own claim, or 1 when the + * namespace was never claimed); a read-only check just proceeds. + */ + | { action: 'claim'; epoch: number; previousToken: string | null } + /** This database recorded a claim and crashed or failed before its object was written: write it now. */ + | { action: 'complete' } + | { action: 'fenced'; refusal: ReportingOrphanFenceRefusalV1 }; + +const TAKEOVER_HINT = + 'If this database is the only one that owns this namespace (for example after restoring it from a backup, and ' + + 'the original is gone), call takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }). Otherwise leave ' + + 'orphan collection off on this copy.'; + +/** + * Decide, from what this database recorded and what the object store holds, whether it may collect. + * Pure: the caller does the I/O and the transaction. + */ +export function decideOrphanFence( + row: OrphanFenceRow | null, + latest: { epoch: number; token: string } | null +): OrphanFenceDecision { + const refuse = ( + cause: ReportingOrphanFenceCauseV1, + detail: string + ): { action: 'fenced'; refusal: ReportingOrphanFenceRefusalV1 } => ({ + action: 'fenced', + refusal: { + cause, + detail: `${detail}. ${TAKEOVER_HINT}`, + ...(latest ? { fenceEpoch: latest.epoch } : {}), + ...(row ? { recordedEpoch: row.epoch } : {}), + }, + }); + if (!row) { + if (!latest) return { action: 'claim', epoch: 1, previousToken: null }; + return refuse( + 'claimed_elsewhere', + `the namespace is already claimed (epoch ${latest.epoch}) by a database this one has no record of: ` + + 'it is a clone or restore of that database, or another installation that shares the namespace' + ); + } + if (!latest) { + // An unconfirmed first claim: this database recorded epoch 1 and did not write the object. + if (row.epoch === 1 && row.previousToken === null) return { action: 'complete' }; + return refuse( + 'fence_missing', + `this database recorded epoch ${row.epoch} and the object store holds no fence: the bucket or prefix was ` + + 'reset or is a different one' + ); + } + if (latest.epoch === row.epoch) { + if (latest.token === row.token) return { action: 'claim', epoch: row.epoch + 1, previousToken: row.token }; + return refuse( + 'token_mismatch', + `epoch ${row.epoch} in the object store belongs to another claim: two databases claimed the same epoch and ` + + 'the other one won' + ); + } + if (latest.epoch > row.epoch) { + return refuse( + 'fence_advanced', + `the object store holds epoch ${latest.epoch} and this database recorded epoch ${row.epoch}: another ` + + 'database that shares this namespace has collected since this database was copied or restored' + ); + } + // latest.epoch < row.epoch. One behind with the recorded previous token is an unconfirmed claim of this + // database (it wrote the intent and not the object); anything else is a regression. + if (latest.epoch === row.epoch - 1 && row.previousToken !== null && latest.token === row.previousToken) { + return { action: 'complete' }; + } + return refuse( + 'fence_regressed', + `the object store holds epoch ${latest.epoch}, older than the epoch ${row.epoch} this database recorded: the ` + + 'database was restored from a newer state than the bucket, or the bucket from an older one' + ); +} + +/** Object key of one epoch's fence object under a binding's owned prefix (`{prefix}/{namespace_key}/`). */ +export function orphanFenceKey(ownedPrefix: string, epoch: number): string { + return `${orphanFenceDirectory(ownedPrefix)}${String(epoch).padStart(EPOCH_DIGITS, '0')}.json`; +} + +/** The listing prefix of the fence chain. */ +export function orphanFenceDirectory(ownedPrefix: string): string { + return `${ownedPrefix}${ORPHAN_FENCE_DIRECTORY}/`; +} + +/** Whether a listed key is part of the fence chain. */ +export function isOrphanFenceKey(ownedPrefix: string, key: string): boolean { + return key.startsWith(orphanFenceDirectory(ownedPrefix)); +} + +/** The epoch a fence key names, or `undefined` for any other key under the fence directory. */ +export function orphanFenceEpochOfKey(ownedPrefix: string, key: string): number | undefined { + const directory = orphanFenceDirectory(ownedPrefix); + if (!key.startsWith(directory)) return undefined; + const match = KEY_NAME.exec(key.slice(directory.length)); + if (!match) return undefined; + const epoch = Number(match[1]); + return Number.isSafeInteger(epoch) && epoch >= 1 ? epoch : undefined; +} + +export function encodeOrphanFenceRecord(namespaceKey: string, record: OrphanFenceRecord): Uint8Array { + return Buffer.from( + `${JSON.stringify({ + v: 1, + namespace_key: namespaceKey, + epoch: record.epoch, + token: record.token, + previous_token: record.previousToken, + claimed_at: record.claimedAt, + })}\n` + ); +} + +/** Parse a fence object strictly; returns `undefined` for anything that is not exactly a fence of this namespace and epoch. */ +export function parseOrphanFenceRecord( + bytes: Uint8Array, + expected: { namespaceKey: string; epoch: number } +): OrphanFenceRecord | undefined { + if (bytes.byteLength > ORPHAN_FENCE_MAX_BYTES) return undefined; + let value: unknown; + try { + value = JSON.parse(Buffer.from(bytes).toString('utf8')); + } catch { + return undefined; + } + if (value === null || typeof value !== 'object' || Array.isArray(value)) return undefined; + const record = value as Record; + const keys = Object.keys(record).sort().join(','); + if (keys !== 'claimed_at,epoch,namespace_key,previous_token,token,v') return undefined; + if ( + record.v !== 1 || + record.namespace_key !== expected.namespaceKey || + record.epoch !== expected.epoch || + typeof record.token !== 'string' || + !UUID.test(record.token) || + !( + record.previous_token === null || + (typeof record.previous_token === 'string' && UUID.test(record.previous_token)) + ) || + typeof record.claimed_at !== 'string' || + !Number.isFinite(Date.parse(record.claimed_at)) + ) { + return undefined; + } + return { + epoch: expected.epoch, + token: record.token, + previousToken: record.previous_token as string | null, + claimedAt: record.claimed_at, + }; +} + +/** + * Identity of one physical fence: the provider, its location, the prefix and the namespace key. Bindings that + * share all four share a fence, because a collector listing through either one reaches the same objects. + */ +export function orphanFenceScope(input: { + provider: string; + location: Readonly>; + prefix: string; + namespaceKey: string; +}): string { + return createHash('sha256') + .update( + canonicalize(['adcp.orphan_fence.v1', input.provider, { ...input.location }, input.prefix, input.namespaceKey]) + ) + .digest('hex'); +} diff --git a/packages/reporting-storage/src/postgres.ts b/packages/reporting-storage/src/postgres.ts index f9a96ec62..78fde5d53 100644 --- a/packages/reporting-storage/src/postgres.ts +++ b/packages/reporting-storage/src/postgres.ts @@ -90,6 +90,12 @@ import { type ReportingRowChunkLocatorV1, type ReportingRowStorageOptionsV1, } from './row-storage'; +import { + decideOrphanFence, + type OrphanFenceEntry, + type OrphanFenceRow, + type ReportingOrphanFenceRefusalV1, +} from './orphan-fence'; type QueryResultRow = Record; interface ReportingPgResult { @@ -2696,9 +2702,17 @@ ${managedDueArm} ) * Deleting needs `acknowledgeExclusiveNamespace: true`: enable it only on the ledger database that * owns the namespace. A clone or restore of that database that still points at the same bucket * shares its `namespace_key`, and would delete the objects of every row set the original wrote after - * the clone was taken. Everything the listing finds under the owned prefix with a row-object key is - * treated as written by this SDK, so do not store other `.jsonl` files there. `deleted` counts delete - * requests that found nothing referenced; an object another host removed first still counts. + * the clone was taken. That flag is copied with the configuration, so it is backed by a fence: a real + * pass first claims the namespace (a create-only object under `{prefix}/{namespace_key}/_collector-fence/` + * and a row in `adcp_reporting_orphan_collector_fences`) and deletes only while the object store's + * highest claim is exactly the one this database recorded. A database copied before the original's last + * pass sees a newer claim and refuses: its binding is reported `fenced` with a `cause`, in `fenced`, and + * in the maintenance warnings, and nothing is deleted. A `dryRun` checks the fence without claiming it + * and reports fenced bindings the same way. After a restore, or when the original is gone, take the + * namespace over deliberately with `takeOverOrphanCollectorFence`. Everything the listing finds under + * the owned prefix with a row-object key is treated as written by this SDK, so do not store other + * `.jsonl` files there. `deleted` counts delete requests that found nothing referenced; an object + * another host removed first still counts. */ async collectOrphanedRowObjects( input: { @@ -2766,6 +2780,7 @@ ${managedDueArm} ) failed: 0, unsupported: [], skippedReadOnly: [], + fenced: [], candidates: [], cursors: {}, completed: [], @@ -2773,6 +2788,24 @@ ${managedDueArm} ) }; const tables = await this.optionalTables(); if (!tables.has('adcp_reporting_row_sets') || !tables.has('adcp_reporting_row_write_intents')) return result; + const hasFenceTable = tables.has('adcp_reporting_orphan_collector_fences'); + // One fence check or claim per physical fence per pass: bindings that share a bucket and prefix share it. + const gates = new Map(); + const gate = async (bindingId: string, mode: 'claim' | 'check') => { + const { scope } = await this.rowStorage.collectorFenceLocation(bindingId); + if (mode === 'claim') { + const known = gates.get(scope); + if (known) return known; + } + const outcome = await this.orphanFenceGate( + bindingId, + mode === 'claim' && dryRun ? 'check' : mode, + hasFenceTable, + input.signal + ); + if (mode === 'claim') gates.set(scope, outcome); + return outcome; + }; const cutoff = clock.getTime() - minAgeSeconds * 1000; for (const summary of summaries) { if (input.bindings && !input.bindings.includes(summary.id)) continue; @@ -2801,6 +2834,24 @@ ${managedDueArm} ) continue; } let token = input.cursors?.[summary.id]; + // The ownership fence comes before any listing: a clone or restore must not even enumerate for deletion. + try { + const claim = await gate(summary.id, 'claim'); + if (!claim.ok) { + outcome.status = 'fenced'; + outcome.fence = claim.refusal; + result.fenced.push(summary.id); + continue; + } + } catch (error) { + input.signal?.throwIfAborted(); + outcome.status = 'failed'; + outcome.error = isReportingRowStoreError(error) ? error.code : 'INTERNAL_ERROR'; + outcome.failed += 1; + // The listing never started, so the position it had stays where it was. + if (token !== undefined) result.cursors[summary.id] = token; + continue; + } // A failure on the first page of a resumed listing drops the cursor: a stale or invalid token must // not wedge the binding, and restarting from the beginning is always safe. let resumed = token !== undefined; @@ -2838,6 +2889,17 @@ ${managedDueArm} ) old.push(entry); } const named = await this.rowObjectKeysNamedByRecords(old.map(entry => entry.objectRef)); + if (!dryRun && old.some(entry => !named.has(entry.objectRef))) { + // Another database may have claimed the namespace since this pass began: read the fence again + // before every page that deletes. + const again = await gate(summary.id, 'check'); + if (!again.ok) { + outcome.status = 'fenced'; + outcome.fence = again.refusal; + result.fenced.push(summary.id); + break; + } + } for (const entry of old) { input.signal?.throwIfAborted(); if (named.has(entry.objectRef)) { @@ -2909,6 +2971,244 @@ ${managedDueArm} ) return result; } + /** + * Whether this database may collect orphans under a binding's namespace, and for a real pass the claim that + * makes it so (see `orphan-fence.ts`). `check` only reads: it is what `dryRun` and the per-page recheck use. + * `claim` runs under a per-fence advisory lock: it verifies the object store's highest epoch is exactly the + * claim this database recorded, records the next epoch and a fresh token in the database, and only then + * writes that epoch's object create-only. The intent is committed before the object is written, so a crash + * between the two is finished by the next pass instead of looking like another database's claim. + */ + private async orphanFenceGate( + bindingId: string, + mode: 'claim' | 'check', + fencesTable: boolean, + signal?: AbortSignal + ): Promise<{ ok: true } | { ok: false; refusal: ReportingOrphanFenceRefusalV1 }> { + if (!fencesTable) { + return { + ok: false, + refusal: { + cause: 'migration_required', + detail: + 'the orphan collector fence table does not exist: apply the pending reporting-storage migration ' + + '(npx adcp-reporting-storage migrate), then orphan collection can claim its namespace', + }, + }; + } + const { scope } = await this.rowStorage.collectorFenceLocation(bindingId); + try { + if (mode === 'check') { + const row = await this.orphanFenceRow(scope); + const { latest } = await this.latestOrphanFence(bindingId, signal); + const decision = decideOrphanFence(row?.row ?? null, latest); + return decision.action === 'fenced' ? { ok: false, refusal: decision.refusal } : { ok: true }; + } + const plan = await this.transaction(async client => { + await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [`adcp.orphan_fence:${scope}`]); + const row = await this.orphanFenceRow(scope, client); + const { entries, latest } = await this.latestOrphanFence(bindingId, signal); + const decision = decideOrphanFence(row?.row ?? null, latest); + if (decision.action === 'fenced') return { kind: 'fenced', refusal: decision.refusal } as const; + if (decision.action === 'complete') { + return { + kind: 'write', + record: { + epoch: row!.row.epoch, + token: row!.row.token, + previousToken: row!.row.previousToken, + claimedAt: row!.claimedAt, + }, + entries, + } as const; + } + const record = { + epoch: decision.epoch, + token: randomUUID(), + previousToken: decision.previousToken, + claimedAt: new Date().toISOString(), + }; + await client.query( + `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token, claimed_at) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (fence_scope) DO UPDATE + SET epoch = EXCLUDED.epoch, token = EXCLUDED.token, + previous_token = EXCLUDED.previous_token, claimed_at = EXCLUDED.claimed_at`, + [scope, record.epoch, record.token, record.previousToken, record.claimedAt] + ); + return { kind: 'write', record, entries } as const; + }); + if (plan.kind === 'fenced') return { ok: false, refusal: plan.refusal }; + const written = await this.rowStorage.putCollectorFence(bindingId, plan.record, signal); + if (written.record.token !== plan.record.token) { + return { + ok: false, + refusal: { + cause: 'claim_lost', + detail: + `another database claimed epoch ${plan.record.epoch} first. If this database is the only one that ` + + 'owns this namespace, call takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); ' + + 'otherwise leave orphan collection off on this copy.', + fenceEpoch: written.record.epoch, + recordedEpoch: plan.record.epoch, + }, + }; + } + // Keep the claim and its predecessor; everything older is superseded. + await this.rowStorage.deleteCollectorFence( + bindingId, + plan.entries.filter(entry => entry.epoch < plan.record.epoch - 1), + signal + ); + return { ok: true }; + } catch (error) { + // Fence objects that are not exactly what this collector writes fail closed. + if (isReportingRowStoreError(error) && error.code === 'ROWS_INTEGRITY_FAILED') { + return { + ok: false, + refusal: { + cause: 'fence_unreadable', + detail: + 'the collector fence objects under this namespace are malformed, oversized, duplicated or name ' + + 'another namespace. Inspect them; once the namespace is known to be owned by this database alone, ' + + 'call takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }).', + }, + }; + } + throw error; + } + } + + private async orphanFenceRow( + scope: string, + client?: ReportingPgClient + ): Promise<{ row: OrphanFenceRow; claimedAt: string } | null> { + const sql = `SELECT epoch::text AS epoch, token, previous_token, claimed_at + FROM adcp_reporting_orphan_collector_fences WHERE fence_scope = $1`; + const result = client + ? await client.query<{ epoch: string; token: string; previous_token: string | null; claimed_at: Date }>(sql, [ + scope, + ]) + : await this.query<{ epoch: string; token: string; previous_token: string | null; claimed_at: Date }>(sql, [ + scope, + ]); + const found = result.rows[0]; + if (!found) return null; + return { + row: { epoch: Number(found.epoch), token: found.token, previousToken: found.previous_token }, + claimedAt: new Date(found.claimed_at).toISOString(), + }; + } + + /** The object store's current claim: the highest epoch listed, read and verified. */ + private async latestOrphanFence( + bindingId: string, + signal?: AbortSignal + ): Promise<{ entries: OrphanFenceEntry[]; latest: { epoch: number; token: string } | null }> { + const entries = await this.rowStorage.listCollectorFence(bindingId, signal); + const newest = entries[entries.length - 1]; + if (!newest) return { entries, latest: null }; + const record = await this.rowStorage.readCollectorFence(bindingId, newest, signal); + return { entries, latest: { epoch: record.epoch, token: record.token } }; + } + + /** + * Take over the orphan collector's namespace fence: the deliberate step after a restore, or when the database + * that held the fence is gone for good. It claims the next epoch for this database, so the previous holder + * (and every other copy of its state) is fenced out, and this database's collector can delete again. + * + * Do this only when this database is the one that should own the namespace. If the original is still running, + * taking over from a clone fences the original and lets the clone delete its objects: the acknowledgement + * exists because nothing else can tell them apart. `acknowledgePreviousOwnerGone: true` is required. + * + * Takes over every writable (or, with `includeReadOnly`, every) object binding whose provider can list, or + * only the `bindings` named; bindings that share a bucket, prefix and installation share one fence and are + * reported together. Requires the `0009` migration. + */ + async takeOverOrphanCollectorFence(input: { + acknowledgePreviousOwnerGone: boolean; + bindings?: readonly string[]; + includeReadOnly?: boolean; + signal?: AbortSignal; + }): Promise<{ bindingIds: string[]; epoch: number; previousEpoch: number | null }[]> { + if (input?.acknowledgePreviousOwnerGone !== true) { + throw new TypeError( + 'takeOverOrphanCollectorFence fences out the namespace’s current owner: pass acknowledgePreviousOwnerGone: ' + + 'true to confirm that database is gone or must never collect again' + ); + } + const summaries = this.rowStorage.objectBindingSummaries(); + for (const id of input.bindings ?? []) { + const summary = summaries.find(item => item.id === id); + if (!summary) throw new RangeError(`bindings names ${id}, which is not a configured object binding`); + if (!summary.listable) throw new RangeError(`bindings names ${id}, whose provider cannot list objects`); + } + const tables = await this.query<{ present: boolean }>( + `SELECT to_regclass('adcp_reporting_orphan_collector_fences') IS NOT NULL AS present` + ); + if (!tables.rows[0]?.present) { + throw new Error('the orphan collector fence table does not exist: run npx adcp-reporting-storage migrate'); + } + const selected = summaries.filter( + summary => + summary.listable && + (input.bindings ? input.bindings.includes(summary.id) : input.includeReadOnly === true || !summary.readOnly) + ); + const byScope = new Map(); + for (const summary of selected) { + const { scope } = await this.rowStorage.collectorFenceLocation(summary.id); + byScope.set(scope, [...(byScope.get(scope) ?? []), summary.id]); + } + const taken: { bindingIds: string[]; epoch: number; previousEpoch: number | null }[] = []; + for (const [scope, bindingIds] of byScope) { + input.signal?.throwIfAborted(); + const bindingId = bindingIds[0]!; + const plan = await this.transaction(async client => { + await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [`adcp.orphan_fence:${scope}`]); + const stored = await this.orphanFenceRow(scope, client); + const entries = await this.rowStorage.listCollectorFence(bindingId, input.signal); + const newest = entries[entries.length - 1]; + // A malformed newest object is exactly what a takeover is for: its key still gives the epoch to supersede. + const previous = newest + ? await this.rowStorage.readCollectorFence(bindingId, newest, input.signal).catch(error => { + if (isReportingRowStoreError(error) && error.code === 'ROWS_INTEGRITY_FAILED') return null; + throw error; + }) + : null; + const epoch = Math.max(newest?.epoch ?? 0, stored?.row.epoch ?? 0) + 1; + const record = { + epoch, + token: randomUUID(), + previousToken: previous?.token ?? null, + claimedAt: new Date().toISOString(), + }; + await client.query( + `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token, claimed_at) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (fence_scope) DO UPDATE + SET epoch = EXCLUDED.epoch, token = EXCLUDED.token, + previous_token = EXCLUDED.previous_token, claimed_at = EXCLUDED.claimed_at`, + [scope, record.epoch, record.token, record.previousToken, record.claimedAt] + ); + return { record, entries, previousEpoch: newest?.epoch ?? null }; + }); + const written = await this.rowStorage.putCollectorFence(bindingId, plan.record, input.signal); + if (written.record.token !== plan.record.token) { + throw new ReportingRowStoreError( + 'STATE_UNAVAILABLE', + `another database claimed epoch ${plan.record.epoch} while this takeover ran; run it again` + ); + } + await this.rowStorage.deleteCollectorFence( + bindingId, + plan.entries.filter(entry => entry.epoch < plan.record.epoch - 1), + input.signal + ); + taken.push({ bindingIds, epoch: plan.record.epoch, previousEpoch: plan.previousEpoch }); + } + return taken; + } + /** * Of these object keys, the ones a ledger record names at any version, under any binding: a chunk of a row * set that is not yet `pruned`, or an object any write intent lists. Version- and binding-blind on purpose: @@ -3023,7 +3323,8 @@ ${managedDueArm} ) WHERE table_schema = current_schema() AND table_name IN ('adcp_reporting_row_sets', 'adcp_reporting_row_write_intents', 'adcp_reporting_materializations', 'adcp_reporting_changes', - 'adcp_reporting_feed_consumers', 'adcp_reporting_receipts')` + 'adcp_reporting_feed_consumers', 'adcp_reporting_receipts', + 'adcp_reporting_orphan_collector_fences')` ); return new Set(result.rows.map(row => row.name)); } @@ -6244,9 +6545,11 @@ export interface ReportingOrphanBindingResultV1 { /** * `complete`: the listing reached its end. `partial`: stopped at `limit`; resume from `cursors`. * `failed`: the listing or its provider failed (`error` is the stable code). `unsupported`: the - * binding's provider has no `list`. `read_only`: skipped because the binding is `readOnly`. + * binding's provider has no `list`. `read_only`: skipped because the binding is `readOnly`. `fenced`: + * another database holds the namespace's collector fence (or this database cannot prove it holds it), so + * nothing was deleted; `fence` says why. */ - status: 'complete' | 'partial' | 'failed' | 'unsupported' | 'read_only'; + status: 'complete' | 'partial' | 'failed' | 'unsupported' | 'read_only' | 'fenced'; /** Objects the provider listed, including ones dropped as foreign. */ listed: number; /** Old enough, but named by a ledger record or open upload; left alone. */ @@ -6260,6 +6563,8 @@ export interface ReportingOrphanBindingResultV1 { wouldDelete: number; failed: number; error?: string; + /** Set when `status` is `fenced`. */ + fence?: ReportingOrphanFenceRefusalV1; } /** @@ -6278,6 +6583,8 @@ export interface ReportingOrphanCollectionResultV1 { unsupported: string[]; /** `readOnly` bindings skipped. */ skippedReadOnly: string[]; + /** Bindings whose collector fence another database holds: nothing was deleted there. See each binding's `fence`. */ + fenced: string[]; /** Dry run only: the orphans a real pass would delete, at most 1000. */ candidates: { bindingId: string; diff --git a/packages/reporting-storage/src/row-storage-migration.ts b/packages/reporting-storage/src/row-storage-migration.ts index 1c6c5a403..137bfdfd1 100644 --- a/packages/reporting-storage/src/row-storage-migration.ts +++ b/packages/reporting-storage/src/row-storage-migration.ts @@ -392,3 +392,26 @@ export async function probeReportingRowStorageSchemaV1(pool: QueryablePoolV1): P if (!installationId) throw new Error('Reporting row storage installation identity is missing'); return { installationId }; } + +/** + * The orphan collector's ownership fence (adcontextprotocol/adcp-client#3292): one row per fence, recording the + * `(epoch, token)` this database last claimed in the object store. A collector deletes only while the object + * store's highest fence epoch is exactly the claim recorded here, so a clone or restore of this database that + * the original has since out-claimed refuses to delete. Applied as migration `0009`. Idempotent. + * + * Until it is applied, `collectOrphanedRowObjects` reports every binding as fenced with cause + * `migration_required` (a `dryRun` included) and deletes nothing. + * + * @experimental Reliable Reporting storage surface: breaking changes may ship in minor releases. See https://github.com/adcontextprotocol/adcp-client/blob/main/docs/guides/REPORTING-STABILITY.md + */ +export const REPORTING_ORPHAN_FENCE_MIGRATION = ` +CREATE TABLE IF NOT EXISTS adcp_reporting_orphan_collector_fences ( + fence_scope TEXT COLLATE "C" PRIMARY KEY CHECK (fence_scope ~ '^[0-9a-f]{64}$'), + epoch BIGINT NOT NULL CHECK (epoch >= 1), + token TEXT COLLATE "C" NOT NULL + CHECK (token ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'), + previous_token TEXT COLLATE "C" + CHECK (previous_token IS NULL OR previous_token ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'), + claimed_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() +); +`.trim(); diff --git a/packages/reporting-storage/src/row-storage.ts b/packages/reporting-storage/src/row-storage.ts index 867410ec5..d98da1da9 100644 --- a/packages/reporting-storage/src/row-storage.ts +++ b/packages/reporting-storage/src/row-storage.ts @@ -19,6 +19,18 @@ import { ReportingRowStoreError, isReportingRowStoreError, } from '@adcp/sdk/reporting/ledger'; +import { + ORPHAN_FENCE_MAX_BYTES, + encodeOrphanFenceRecord, + isOrphanFenceKey, + orphanFenceDirectory, + orphanFenceEpochOfKey, + orphanFenceKey, + orphanFenceScope, + parseOrphanFenceRecord, + type OrphanFenceEntry, + type OrphanFenceRecord, +} from './orphan-fence'; import { probeReportingRowStorageSchemaV1 } from './row-storage-migration'; import { isReportingRowInstant, isReportingRowObjectKeyShape } from './row-storage-provider-io'; import { @@ -440,6 +452,8 @@ export class ReportingRowStorageV1 { const objects: ReportingRowObjectListEntryV1[] = []; let foreign = Number.isSafeInteger(page.skipped) && page.skipped! > 0 ? page.skipped! : 0; for (const entry of page.objects) { + // The collector's own fence chain lives under the owned prefix; it is not a row object and not foreign. + if (entry && typeof entry.objectRef === 'string' && isOrphanFenceKey(ownedPrefix, entry.objectRef)) continue; if ( !entry || typeof entry.objectRef !== 'string' || @@ -460,6 +474,216 @@ export class ReportingRowStorageV1 { return { ownedPrefix, objects, foreign, ...(page.nextPageToken ? { nextPageToken: page.nextPageToken } : {}) }; } + /** + * The physical location of an object binding's collector fence: the scope its database row is keyed by, and + * the object-store directory that holds its chain. Bindings that share provider, location, prefix and + * installation share a fence. Throws `STATE_UNAVAILABLE` for a binding that is not an object binding, and + * `INVALID_INPUT` when its provider cannot list (the fence is read by listing). + */ + async collectorFenceLocation(bindingId: string): Promise<{ scope: string; directory: string; namespaceKey: string }> { + await this.ready(); + const binding = this.bindings[bindingId]; + if (!binding || binding.kind !== 'object') { + throw new ReportingRowStoreError('STATE_UNAVAILABLE', `row binding ${bindingId} is not configured for objects`); + } + const provider = this.providers[binding.provider]!; + if (typeof provider.list !== 'function') { + throw new ReportingRowStoreError('INVALID_INPUT', `row binding ${bindingId}'s provider cannot list objects`); + } + const namespaceKey = this.installation!.namespaceKey; + return { + scope: orphanFenceScope({ + provider: binding.provider, + location: binding.location, + prefix: binding.prefix, + namespaceKey, + }), + directory: orphanFenceDirectory(`${binding.prefix}/${namespaceKey}/`), + namespaceKey, + }; + } + + /** + * Every fence object of a binding's namespace, ordered by epoch. Keys under the fence directory that are + * not an epoch name are ignored. Two live versions of one epoch's key (a versioned store after a manual + * delete and re-create) are ambiguous and fail closed with `ROWS_INTEGRITY_FAILED`, as does a chain too + * long to be one this collector wrote (it prunes to two objects). + */ + async listCollectorFence(bindingId: string, signal?: AbortSignal): Promise { + const { directory } = await this.collectorFenceLocation(bindingId); + const binding = this.bindings[bindingId] as ReportingObjectRowBindingV1; + const provider = this.providers[binding.provider]!; + const ownedPrefix = `${binding.prefix}/${this.installation!.namespaceKey}/`; + const entries = new Map(); + const pageSize = 50; + let token: string | undefined; + for (let page = 0; ; page += 1) { + if (page >= 20) { + throw new ReportingRowStoreError( + 'ROWS_INTEGRITY_FAILED', + 'the collector fence directory holds too many objects' + ); + } + const context = { signal: deadlineSignal(this.readDeadline, signal) }; + const listed = await this.provider(() => + provider.list!( + { + ...providerLocation(binding), + prefix: directory, + pageSize, + ...(token !== undefined ? { pageToken: token } : {}), + }, + context + ) + ); + if (!listed || !Array.isArray(listed.objects) || listed.objects.length > pageSize) { + throw new ReportingRowStoreError('PROVIDER_UNAVAILABLE', 'provider returned an invalid listing'); + } + for (const object of listed.objects) { + if (!object || typeof object.objectRef !== 'string' || typeof object.nativeVersionRef !== 'string') continue; + if ( + object.nativeVersionRef.length < 1 || + object.nativeVersionRef.length > REPORTING_ROW_NATIVE_VERSION_REF_MAX_LENGTH + ) { + continue; + } + const epoch = orphanFenceEpochOfKey(ownedPrefix, object.objectRef); + if (epoch === undefined || object.objectRef !== orphanFenceKey(ownedPrefix, epoch)) continue; + if (entries.has(epoch)) { + throw new ReportingRowStoreError( + 'ROWS_INTEGRITY_FAILED', + 'the collector fence holds two versions of one epoch' + ); + } + entries.set(epoch, { epoch, objectRef: object.objectRef, nativeVersionRef: object.nativeVersionRef }); + } + if (listed.nextPageToken === undefined) break; + if (typeof listed.nextPageToken !== 'string' || listed.nextPageToken === '' || listed.nextPageToken === token) { + throw new ReportingRowStoreError('PROVIDER_UNAVAILABLE', 'provider returned an invalid page token'); + } + token = listed.nextPageToken; + } + return [...entries.values()].sort((a, b) => a.epoch - b.epoch); + } + + /** + * Read and verify one fence object. A missing version is `PROVIDER_UNAVAILABLE` (it was listed a moment ago, + * so this is a race to retry); content that is not exactly a fence of this namespace and epoch is + * `ROWS_INTEGRITY_FAILED`. + */ + async readCollectorFence( + bindingId: string, + entry: OrphanFenceEntry, + signal?: AbortSignal + ): Promise { + await this.ready(); + const binding = this.bindings[bindingId]; + if (!binding || binding.kind !== 'object') { + throw new ReportingRowStoreError('STATE_UNAVAILABLE', `row binding ${bindingId} is not configured for objects`); + } + const provider = this.providers[binding.provider]!; + let bytes: Uint8Array | null; + try { + bytes = await this.provider(() => + provider.get( + { + ...providerLocation(binding), + objectRef: entry.objectRef, + nativeVersionRef: entry.nativeVersionRef, + maxBytes: ORPHAN_FENCE_MAX_BYTES, + }, + { signal: deadlineSignal(this.readDeadline, signal) } + ) + ); + } catch (error) { + // An object over the read cap is the same finding as bad content. + if (isReportingRowStoreError(error) && error.code === 'ROWS_INTEGRITY_FAILED') { + throw new ReportingRowStoreError('ROWS_INTEGRITY_FAILED', 'the collector fence object is malformed'); + } + throw error; + } + if (bytes === null) { + throw new ReportingRowStoreError( + 'PROVIDER_UNAVAILABLE', + 'a collector fence object disappeared while it was read' + ); + } + const record = parseOrphanFenceRecord(bytes, { + namespaceKey: this.installation!.namespaceKey, + epoch: entry.epoch, + }); + if (!record) { + throw new ReportingRowStoreError('ROWS_INTEGRITY_FAILED', 'the collector fence object is malformed'); + } + return record; + } + + /** + * Create-only write of one epoch's fence object. When the key is taken, returns the record already there + * (verified), so the caller can tell its own earlier write from another database's claim. + */ + async putCollectorFence( + bindingId: string, + record: OrphanFenceRecord, + signal?: AbortSignal + ): Promise<{ created: boolean; record: OrphanFenceRecord }> { + await this.ready(); + const binding = this.bindings[bindingId]; + if (!binding || binding.kind !== 'object') { + throw new ReportingRowStoreError('STATE_UNAVAILABLE', `row binding ${bindingId} is not configured for objects`); + } + const provider = this.providers[binding.provider]!; + const { namespaceKey, installationId } = this.installation!; + const objectRef = orphanFenceKey(`${binding.prefix}/${namespaceKey}/`, record.epoch); + const put = await this.provider(() => + provider.putIfAbsent( + { + ...providerLocation(binding), + objectRef, + bytes: encodeOrphanFenceRecord(namespaceKey, record), + contentType: 'application/x-ndjson', + metadata: { 'adcp-installation': installationId, 'adcp-collector-fence': String(record.epoch) }, + }, + { signal: deadlineSignal(this.writeDeadline, signal) } + ) + ); + assertNativeVersionRef(put?.nativeVersionRef); + if (put.created) return { created: true, record }; + return { + created: false, + record: await this.readCollectorFence( + bindingId, + { epoch: record.epoch, objectRef, nativeVersionRef: put.nativeVersionRef }, + signal + ), + }; + } + + /** Delete superseded fence objects. Best effort: a failure leaves a stale object that the next claim prunes. */ + async deleteCollectorFence( + bindingId: string, + entries: readonly OrphanFenceEntry[], + signal?: AbortSignal + ): Promise { + await this.ready(); + const binding = this.bindings[bindingId]; + if (!binding || binding.kind !== 'object') return; + const provider = this.providers[binding.provider]!; + for (const entry of entries) { + signal?.throwIfAborted(); + try { + await this.provider(() => + provider.delete( + { ...providerLocation(binding), objectRef: entry.objectRef, nativeVersionRef: entry.nativeVersionRef }, + { signal: deadlineSignal(this.writeDeadline, signal) } + ) + ); + } catch { + signal?.throwIfAborted(); + } + } + } + /** Probe the schema and every provider, and register (or verify) every configured binding. Idempotent. */ ready(): Promise { this.readied ??= this.register().catch(error => { diff --git a/packages/reporting-storage/test/reporting-maintenance-warnings.test.js b/packages/reporting-storage/test/reporting-maintenance-warnings.test.js index c3de8fe09..fcc6ed0ca 100644 --- a/packages/reporting-storage/test/reporting-maintenance-warnings.test.js +++ b/packages/reporting-storage/test/reporting-maintenance-warnings.test.js @@ -50,6 +50,7 @@ describe('orphan object collection task', () => { failed: 0, unsupported: [], skippedReadOnly: [], + fenced: [], candidates: [], cursors: {}, completed: [], @@ -137,6 +138,34 @@ describe('orphan object collection task', () => { assert.ok(orphanState.nextCycleAt > Date.now()); }); + test('a fenced binding is a warning that names the cause and the takeover, and the cycle still rests', async () => { + const orphanState = { cursors: {} }; + const fenced = async () => + orphanResult({ + fenced: ['objects'], + bindings: [ + { + bindingId: 'objects', + status: 'fenced', + fence: { + cause: 'fence_advanced', + detail: 'another database has collected. call takeOverOrphanCollectorFence', + fenceEpoch: 3, + recordedEpoch: 1, + }, + }, + ], + }); + const { task } = await runWith(fenced, { + orphanCollection: { enabled: true, acknowledgeExclusiveNamespace: true }, + orphanState, + }); + assert.match(task.partialFailure.message, /refused to delete for row binding objects/); + assert.match(task.partialFailure.message, /fence_advanced/); + assert.match(task.partialFailure.message, /takeOverOrphanCollectorFence/); + assert.ok(orphanState.nextCycleAt > Date.now(), 'a refusal is not retried every pass'); + }); + test('a failed binding keeps the task retrying every pass and names its cause; a rejected collector is a failed task', async () => { const orphanState = { cursors: {} }; const failing = async () => diff --git a/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js b/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js new file mode 100644 index 000000000..001ab8f0c --- /dev/null +++ b/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js @@ -0,0 +1,634 @@ +/** + * PostgreSQL integration tests for the orphan collector's ownership fence (adcp-client#3292): a clone or + * restore of the ledger database that shares the live one's bucket, configuration, installation identity and + * namespace key must not delete the live database's newer objects. + * + * REPORTING_LEDGER_PG_URL=postgres://localhost/test npm test --workspace=@adcp/reporting-storage -- test/reporting-orphan-collector-fence-pg.test.js + */ +const assert = require('node:assert/strict'); +const { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } = require('node:fs'); +const { tmpdir } = require('node:os'); +const path = require('node:path'); +const { after, before, beforeEach, describe, test } = require('node:test'); + +const DATABASE_URL = process.env.REPORTING_LEDGER_PG_URL; +const DAY = 24 * 3600 * 1000; + +describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQL URL not set' }, () => { + const live = `adcp_fence_live_${process.pid}`; + const clone = `adcp_fence_clone_${process.pid}`; + const bare = `adcp_fence_bare_${process.pid}`; + const schemas = [live, clone, bare]; + const later = () => new Date(Date.now() + 8 * DAY); + let ledger; + let bootstrapPool; + let root; + let provider; + let namespaceKey; + let liveStore; + let livePool; + let clonePool; + let cloneStore; + let barePool; + let bareStore; + const pools = []; + const base = { location: { root: 'rows' } }; + const providerCalls = { delete: [], put: [] }; + + const key = name => `orph/${namespaceKey}/${name}`; + const fenceDirectory = () => path.join(root, 'orph', namespaceKey, '_collector-fence'); + const fenceFiles = () => (existsSync(fenceDirectory()) ? readdirSync(fenceDirectory()).sort() : []); + const fenceName = epoch => `${String(epoch).padStart(16, '0')}.json`; + const onDisk = name => existsSync(path.join(root, name)); + const context = () => ({ signal: AbortSignal.timeout(5_000) }); + async function putObject(objectRef, body = '{"a":1}\n') { + const created = await provider.putIfAbsent( + { + ...base, + objectRef, + bytes: Buffer.from(body), + contentType: 'application/x-ndjson', + metadata: { 'adcp-installation': 'x', 'adcp-intent': 'y' }, + }, + context() + ); + return created.nativeVersionRef; + } + const collect = (target, overrides = {}) => + target.collectOrphanedRowObjects({ now: later(), acknowledgeExclusiveNamespace: true, ...overrides }); + const objectsBinding = result => result.bindings.find(item => item.bindingId === 'objects'); + const fenceRow = async pool => + (await pool.query('SELECT epoch::int AS epoch, token, previous_token FROM adcp_reporting_orphan_collector_fences')) + .rows[0]; + const addIntent = (pool, rowSetId, objectRef, nativeVersionRef) => + pool.query( + `INSERT INTO adcp_reporting_row_write_intents + (row_binding_id, account_id, row_set_id, content_sha256, state, owner_token, created_objects, expires_at) + VALUES ('objects', 'acct_fence', $1, $2, 'open', 'owner', $3::jsonb, clock_timestamp() + INTERVAL '1 hour')`, + [rowSetId, 'e'.repeat(64), JSON.stringify([{ object_ref: objectRef, native_version_ref: nativeVersionRef }])] + ); + const fenceRecord = epoch => JSON.parse(readFileSync(path.join(fenceDirectory(), fenceName(epoch)), 'utf8')); + + async function newPool(schema) { + const { Pool } = require('pg'); + const pool = new Pool({ connectionString: DATABASE_URL, options: `-c search_path="${schema}"` }); + pools.push(pool); + return pool; + } + + function newStore(pool) { + const rowStorage = { + bindings: { + objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' }, + // Shares `objects`' bucket and prefix, as a renamed or successor binding does. + twin: { + kind: 'object', + provider: 'filesystem', + location: { root: 'rows' }, + prefix: 'orph', + keyTemplate: '{prefix}/{namespace_key}/twin/{revision_key}/{content_sha256}/{chunk_index}', + readOnly: true, + }, + }, + providers: { filesystem: provider }, + selectRowBinding: () => 'objects', + }; + return new ledger.PostgresReportingLedgerStore(pool, { acknowledgeIsolatedDatabase: true, rowStorage }); + } + + /** A point-in-time copy of every table in `from`: what a staging clone or a restore carries. */ + async function copySchema(from, to) { + await bootstrapPool.query(`DROP SCHEMA IF EXISTS "${to}" CASCADE`); + await bootstrapPool.query(`CREATE SCHEMA "${to}"`); + const tables = ( + await bootstrapPool.query( + `SELECT table_name FROM information_schema.tables WHERE table_schema = $1 AND table_type = 'BASE TABLE'`, + [from] + ) + ).rows.map(row => row.table_name); + for (const table of tables) { + await bootstrapPool.query( + `CREATE TABLE "${to}"."${table}" (LIKE "${from}"."${table}" INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING INDEXES)` + ); + await bootstrapPool.query(`INSERT INTO "${to}"."${table}" SELECT * FROM "${from}"."${table}"`); + } + } + + async function freshClone() { + if (clonePool) await clonePool.end(); + pools.splice(pools.indexOf(clonePool), 1); + await copySchema(live, clone); + clonePool = await newPool(clone); + cloneStore = newStore(clonePool); + await cloneStore.readyRowStorage(); + return cloneStore; + } + + before(async () => { + const { Pool } = require('pg'); + ledger = require('./helpers/ledger.js'); + root = mkdtempSync(path.join(tmpdir(), 'adcp-fence-')); + bootstrapPool = new Pool({ connectionString: DATABASE_URL }); + for (const schema of schemas) await bootstrapPool.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`); + await bootstrapPool.query(`CREATE SCHEMA "${live}"`); + await bootstrapPool.query(`CREATE SCHEMA "${bare}"`); + livePool = await newPool(live); + await livePool.query(ledger.REPORTING_LEDGER_MIGRATION); + await livePool.query(ledger.REPORTING_ROW_STORAGE_MIGRATION); + await livePool.query(ledger.REPORTING_ORPHAN_FENCE_MIGRATION); + // A database that has not applied the fence migration. + barePool = await newPool(bare); + await barePool.query(ledger.REPORTING_LEDGER_MIGRATION); + await barePool.query(ledger.REPORTING_ROW_STORAGE_MIGRATION); + const filesystem = ledger.createFilesystemReportingRowObjectProviderV1({ roots: { rows: root } }); + provider = new Proxy(filesystem, { + get(target, property) { + if (property === 'delete') { + return async (input, ctx) => { + providerCalls.delete.push(input.objectRef); + return target.delete(input, ctx); + }; + } + if (property === 'putIfAbsent') { + return async (input, ctx) => { + providerCalls.put.push(input.objectRef); + return target.putIfAbsent(input, ctx); + }; + } + const value = target[property]; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + liveStore = newStore(livePool); + await liveStore.readyRowStorage(); + const engine = new ledger.ReportingRowStorageV1(livePool, { + bindings: { objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' } }, + providers: { filesystem: provider }, + }); + await engine.ready(); + namespaceKey = engine.installationIdentity.namespaceKey; + bareStore = newStore(barePool); + await bareStore.readyRowStorage(); + }); + + beforeEach(async () => { + providerCalls.delete.length = 0; + providerCalls.put.length = 0; + await livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); + await livePool.query('DELETE FROM adcp_reporting_row_write_intents'); + rmSync(path.join(root, 'orph'), { recursive: true, force: true }); + if (clonePool) { + await clonePool.end(); + pools.splice(pools.indexOf(clonePool), 1); + clonePool = undefined; + } + }); + + after(async () => { + for (const pool of pools) await pool.end().catch(() => undefined); + if (bootstrapPool) { + for (const schema of schemas) await bootstrapPool.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`); + await bootstrapPool.end(); + } + if (root) rmSync(root, { recursive: true, force: true }); + }); + + test('the first real pass claims epoch 1, each later pass claims the next epoch and prunes the old ones', async () => { + const orphan = key('acct/a.jsonl'); + await putObject(orphan); + const first = await collect(liveStore); + assert.equal(objectsBinding(first).status, 'complete'); + assert.equal(objectsBinding(first).deleted, 1); + assert.deepEqual(first.fenced, []); + assert.deepEqual(fenceFiles(), [fenceName(1)]); + assert.deepEqual(await fenceRow(livePool), { epoch: 1, token: fenceRecord(1).token, previous_token: null }); + assert.equal(fenceRecord(1).namespace_key, namespaceKey); + assert.equal(onDisk(orphan), false); + + await collect(liveStore); + assert.deepEqual(fenceFiles(), [fenceName(1), fenceName(2)]); + assert.equal(fenceRecord(2).previous_token, fenceRecord(1).token); + assert.notEqual(fenceRecord(2).token, fenceRecord(1).token, 'every claim carries a fresh token'); + await collect(liveStore); + await collect(liveStore); + assert.deepEqual(fenceFiles(), [fenceName(3), fenceName(4)], 'older claims are superseded and pruned'); + assert.deepEqual((await fenceRow(livePool)).epoch, 4); + }); + + test('the fence chain is not a row object: it is neither listed, counted as foreign nor deleted', async () => { + await putObject(key('acct/a.jsonl')); + await collect(liveStore); + const second = await collect(liveStore, { dryRun: true }); + assert.equal(objectsBinding(second).foreign, 0); + assert.equal(objectsBinding(second).listed, 0); + assert.ok(fenceFiles().length >= 1); + }); + + test('a dry run checks the fence without claiming it, and reports what a real pass would delete', async () => { + const orphan = key('acct/dry.jsonl'); + await putObject(orphan); + providerCalls.put.length = 0; + const dry = await collect(liveStore, { dryRun: true }); + assert.equal(objectsBinding(dry).wouldDelete, 1); + assert.deepEqual(dry.fenced, []); + assert.deepEqual(fenceFiles(), [], 'a dry run writes no fence object'); + assert.equal(await fenceRow(livePool), undefined, 'and records no claim'); + assert.deepEqual(providerCalls.put, [], 'and writes nothing at all'); + }); + + test('a clone taken before the original last collected is fenced and deletes nothing', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); // epoch 1: the state a clone is taken from + await freshClone(); + // The live database goes on writing after the clone was taken. The clone's tables know nothing of it. + const newer = key('acct/written-after-the-clone.jsonl'); + await addIntent(livePool, 'rrev_after_the_clone', newer, await putObject(newer)); + await collect(liveStore); // epoch 2: the live database out-claims every copy of epoch 1 + assert.ok(onDisk(newer), 'the live database references it'); + + const refused = await collect(cloneStore); + assert.deepEqual(refused.fenced, ['objects']); + const outcome = objectsBinding(refused); + assert.equal(outcome.status, 'fenced'); + assert.equal(outcome.fence.cause, 'fence_advanced'); + assert.equal(outcome.fence.fenceEpoch, 2); + assert.equal(outcome.fence.recordedEpoch, 1); + assert.match(outcome.fence.detail, /takeOverOrphanCollectorFence/); + assert.equal(refused.deleted, 0); + assert.equal(refused.listed, 0, 'a fenced binding is not even listed'); + assert.ok(onDisk(newer), "the live database's newer object survives its clone"); + assert.equal((await fenceRow(clonePool)).epoch, 1, 'the clone recorded nothing'); + + const dry = await collect(cloneStore, { dryRun: true }); + assert.deepEqual(dry.fenced, ['objects'], 'a dry run reports the fenced binding too'); + assert.equal(dry.wouldDelete, 0); + assert.deepEqual(dry.candidates, []); + + // The live database is unaffected and still collects its own orphans. + const own = key('acct/live-orphan.jsonl'); + await putObject(own); + const result = await collect(liveStore); + assert.deepEqual(result.fenced, []); + assert.equal(onDisk(own), false); + assert.ok(onDisk(newer)); + + // Control: this is what the fence prevents. A clone that has taken the namespace over deletes the object + // its own tables know nothing about. + await cloneStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + await collect(cloneStore); + assert.equal(onDisk(newer), false); + }); + + test('a clone that collects first claims the namespace and fences the original: first claim wins', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await freshClone(); + const claimed = await collect(cloneStore); + assert.deepEqual(claimed.fenced, [], 'nothing distinguishes the copy from the original until one claims'); + assert.equal((await fenceRow(clonePool)).epoch, 2); + const original = await collect(liveStore); + assert.deepEqual(original.fenced, ['objects']); + assert.equal(objectsBinding(original).fence.cause, 'fence_advanced'); + assert.match( + require('node:util').inspect(objectsBinding(original).fence.detail), + /another database that shares this namespace has collected/ + ); + }); + + test('a restore to an older database state is fenced until an operator takes over', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); // epoch 1 + const snapshot = (await livePool.query('SELECT * FROM adcp_reporting_orphan_collector_fences')).rows; + await collect(liveStore); // epoch 2 + await collect(liveStore); // epoch 3 + // Point-in-time restore: the database goes back to its state at epoch 1; the bucket is at epoch 3. + await livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); + for (const row of snapshot) { + await livePool.query( + 'INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token, claimed_at) VALUES ($1, $2, $3, $4, $5)', + [row.fence_scope, row.epoch, row.token, row.previous_token, row.claimed_at] + ); + } + const newer = key('acct/lost-with-the-restore.jsonl'); + await putObject(newer); + const refused = await collect(liveStore); + assert.equal(objectsBinding(refused).fence.cause, 'fence_advanced'); + assert.ok(onDisk(newer)); + + await assert.rejects(() => liveStore.takeOverOrphanCollectorFence({}), TypeError); + await assert.rejects( + () => liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: false }), + TypeError + ); + assert.ok(onDisk(newer), 'a refused takeover changes nothing'); + + const taken = await liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + assert.deepEqual(taken, [{ bindingIds: ['objects'], epoch: 4, previousEpoch: 3 }]); + assert.equal(fenceRecord(4).previous_token, fenceRecord(3).token); + const resumed = await collect(liveStore); + assert.deepEqual(resumed.fenced, []); + assert.equal(objectsBinding(resumed).status, 'complete'); + assert.equal(onDisk(newer), false, 'after a takeover the restored database collects again'); + }); + + test('a takeover fences out the previous owner', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await freshClone(); + const taken = await cloneStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + assert.equal(taken[0].epoch, 2); + const original = await collect(liveStore); + assert.equal(objectsBinding(original).fence.cause, 'fence_advanced'); + assert.deepEqual((await collect(cloneStore)).fenced, []); + }); + + test('takeOver validates its bindings and covers a binding that shares a fence once', async () => { + await assert.rejects( + () => liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, bindings: ['missing'] }), + RangeError + ); + const taken = await liveStore.takeOverOrphanCollectorFence({ + acknowledgePreviousOwnerGone: true, + includeReadOnly: true, + }); + assert.deepEqual(taken, [{ bindingIds: ['objects', 'twin'], epoch: 1, previousEpoch: null }]); + assert.deepEqual(fenceFiles(), [fenceName(1)]); + }); + + test('bindings that share a bucket and prefix share one fence and one claim per pass', async () => { + await putObject(key('acct/a.jsonl')); + const result = await collect(liveStore, { includeReadOnly: true }); + assert.deepEqual(result.fenced, []); + assert.deepEqual( + result.bindings.map(item => item.status), + ['complete', 'complete'] + ); + assert.deepEqual(fenceFiles(), [fenceName(1)]); + assert.equal((await fenceRow(livePool)).epoch, 1); + }); + + test('a database with no recorded claim refuses a namespace that already has one', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); + const refused = await collect(liveStore); + assert.equal(objectsBinding(refused).fence.cause, 'claimed_elsewhere'); + assert.equal(objectsBinding(refused).fence.recordedEpoch, undefined); + }); + + test('a recorded claim with no fence in the bucket is refused as missing, never silently re-claimed', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await collect(liveStore); + rmSync(fenceDirectory(), { recursive: true, force: true }); + const survivor = key('acct/survivor.jsonl'); + await putObject(survivor); + const refused = await collect(liveStore); + assert.equal(objectsBinding(refused).fence.cause, 'fence_missing'); + assert.ok(onDisk(survivor)); + }); + + test('a database newer than its bucket is refused as regressed', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await collect(liveStore); + // The database comes back from a newer backup than the bucket: it records epoch 5, the bucket holds 2. + await livePool.query(`UPDATE adcp_reporting_orphan_collector_fences SET epoch = 5, previous_token = $1`, [ + '55555555-5555-4555-8555-555555555555', + ]); + const refused = await collect(liveStore); + assert.equal(objectsBinding(refused).fence.cause, 'fence_regressed'); + assert.equal(objectsBinding(refused).fence.fenceEpoch, 2); + assert.equal(objectsBinding(refused).fence.recordedEpoch, 5); + }); + + test('two claims of one epoch: the loser is fenced as a token mismatch, and a lost race as claim_lost', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); // epoch 1, recorded + // Another database wrote epoch 1 under a different token (the bucket was replaced). + const file = path.join(fenceDirectory(), fenceName(1)); + const record = fenceRecord(1); + writeFileSync(file, `${JSON.stringify({ ...record, token: '11111111-1111-4111-8111-111111111111' })}\n`); + const mismatch = await collect(liveStore); + assert.equal(objectsBinding(mismatch).fence.cause, 'token_mismatch'); + + // A claim of the next epoch that another database wrote between the check and the write. + await livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); + rmSync(fenceDirectory(), { recursive: true, force: true }); + await collect(liveStore); // epoch 1 again, recorded + const intruder = { + ...fenceRecord(1), + epoch: 2, + token: '22222222-2222-4222-8222-222222222222', + previous_token: fenceRecord(1).token, + }; + const racing = new Proxy(provider, { + get(target, property) { + if (property === 'putIfAbsent') { + return async (input, ctx) => { + // The intruder's epoch 2 lands first. + writeFileSync(path.join(fenceDirectory(), fenceName(2)), `${JSON.stringify(intruder)}\n`); + return target.putIfAbsent(input, ctx); + }; + } + const value = target[property]; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + const racingStore = new ledger.PostgresReportingLedgerStore(livePool, { + acknowledgeIsolatedDatabase: true, + rowStorage: { + bindings: { objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' } }, + providers: { filesystem: racing }, + selectRowBinding: () => 'objects', + }, + }); + await racingStore.readyRowStorage(); + const lost = await collect(racingStore); + assert.equal(objectsBinding(lost).fence.cause, 'claim_lost'); + assert.equal(lost.deleted, 0); + // The next pass sees the other database's epoch. + assert.equal(objectsBinding(await collect(liveStore)).fence.cause, 'token_mismatch'); + }); + + test('malformed, foreign-namespace or oversized fence objects fail closed', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + const file = path.join(fenceDirectory(), fenceName(1)); + const survivor = key('acct/survivor.jsonl'); + await putObject(survivor); + for (const content of [ + 'not json', + JSON.stringify({ ...fenceRecord(1), namespace_key: 'f'.repeat(64) }), + JSON.stringify({ ...fenceRecord(1), extra: true }), + JSON.stringify({ ...fenceRecord(1), epoch: 9 }), + JSON.stringify({ ...fenceRecord(1), token: 'nope' }), + `${JSON.stringify(fenceRecord(1))}${' '.repeat(5000)}`, + ]) { + writeFileSync(file, content); + const refused = await collect(liveStore); + assert.equal(objectsBinding(refused).fence.cause, 'fence_unreadable', content.slice(0, 40)); + assert.ok(onDisk(survivor)); + } + // A takeover supersedes even an unreadable chain. + const taken = await liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + assert.equal(taken[0].epoch, 2); + assert.equal(taken[0].previousEpoch, 1); + assert.deepEqual((await collect(liveStore)).fenced, []); + }); + + test('two versions of one epoch are ambiguous and fail closed', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + const duplicating = new Proxy(provider, { + get(target, property) { + if (property === 'list') { + return async (input, ctx) => { + const listed = await target.list(input, ctx); + return input.prefix.endsWith('/_collector-fence/') + ? { + ...listed, + objects: [...listed.objects, ...listed.objects.map(entry => ({ ...entry, nativeVersionRef: 'v2' }))], + } + : listed; + }; + } + const value = target[property]; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + const store = new ledger.PostgresReportingLedgerStore(livePool, { + acknowledgeIsolatedDatabase: true, + rowStorage: { + bindings: { objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' } }, + providers: { filesystem: duplicating }, + selectRowBinding: () => 'objects', + }, + }); + await store.readyRowStorage(); + assert.equal(objectsBinding(await collect(store)).fence.cause, 'fence_unreadable'); + }); + + test('an unfinished claim (intent recorded, object never written) is completed by the next pass', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); // epoch 1 + const orphan = key('acct/after-crash.jsonl'); + await putObject(orphan); + // A crash after the intent for epoch 2 committed and before its object was written. + const intent = '33333333-3333-4333-8333-333333333333'; + await livePool.query( + `UPDATE adcp_reporting_orphan_collector_fences SET epoch = 2, token = $1, previous_token = $2`, + [intent, fenceRecord(1).token] + ); + const result = await collect(liveStore); + assert.deepEqual(result.fenced, []); + assert.equal(onDisk(orphan), false); + assert.equal(fenceRecord(2).token, intent, 'the recorded intent was written as the claim'); + assert.equal((await fenceRow(livePool)).epoch, 2); + + // The same for a crash during the very first claim. + await livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); + rmSync(fenceDirectory(), { recursive: true, force: true }); + await livePool.query( + `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token) + SELECT $1, 1, $2, NULL`, + [await scopeOf(liveStore), intent] + ); + assert.deepEqual((await collect(liveStore)).fenced, []); + assert.equal(fenceRecord(1).token, intent); + }); + + async function scopeOf(store) { + return (await store.rowStorage.collectorFenceLocation('objects')).scope; + } + + test('two hosts of the same database collecting at once do not fence each other', async () => { + await putObject(key('acct/a.jsonl')); + await putObject(key('acct/b.jsonl')); + for (let round = 0; round < 3; round += 1) { + const [first, second] = await Promise.all([collect(liveStore), collect(liveStore)]); + assert.deepEqual([first.fenced, second.fenced], [[], []]); + assert.equal(first.failed + second.failed, 0); + } + // A host that finds its sibling's claim still unwritten finishes that claim instead of claiming again, + // so the epoch advances once or twice per round, never backwards. + const epoch = (await fenceRow(livePool)).epoch; + assert.ok(epoch >= 3 && epoch <= 6, `epoch ${epoch}`); + assert.ok(fenceFiles().includes(fenceName(epoch))); + }); + + test('another database claiming mid-pass stops the deletes on the next page', async () => { + for (const name of ['a', 'b', 'c']) await putObject(key(`acct/${name}.jsonl`)); + await collect(liveStore, { dryRun: true }); + let lists = 0; + const intruding = new Proxy(provider, { + get(target, property) { + if (property === 'list') { + return async (input, ctx) => { + const listed = await target.list(input, ctx); + if (!input.prefix.endsWith('/_collector-fence/')) { + lists += 1; + if (input.pageToken?.endsWith('/acct/a.jsonl')) { + // After the first object's page, another database claims the next epoch. + const record = fenceRecord(1); + writeFileSync( + path.join(fenceDirectory(), fenceName(2)), + `${JSON.stringify({ ...record, epoch: 2, token: '44444444-4444-4444-8444-444444444444', previous_token: record.token })}\n` + ); + } + } + return listed; + }; + } + const value = target[property]; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + const store = new ledger.PostgresReportingLedgerStore(livePool, { + acknowledgeIsolatedDatabase: true, + rowStorage: { + bindings: { objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' } }, + providers: { filesystem: intruding }, + selectRowBinding: () => 'objects', + }, + }); + await store.readyRowStorage(); + const result = await collect(store, { pageSize: 1 }); + assert.equal(objectsBinding(result).status, 'fenced'); + assert.equal(objectsBinding(result).fence.cause, 'fence_advanced'); + assert.equal(objectsBinding(result).deleted, 1, 'the page before the claim was deleted, no page after it'); + assert.deepEqual(result.fenced, ['objects']); + }); + + test('without the fence migration a real pass and a dry run both refuse, and nothing is deleted', async () => { + const orphan = key('acct/needs-migration.jsonl'); + await putObject(orphan); + for (const dryRun of [false, true]) { + const result = await collect(bareStore, { dryRun }); + assert.deepEqual(result.fenced, ['objects']); + assert.equal(objectsBinding(result).fence.cause, 'migration_required'); + assert.match(objectsBinding(result).fence.detail, /npx adcp-reporting-storage migrate/); + } + assert.ok(onDisk(orphan)); + await assert.rejects( + () => bareStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }), + /migrate/ + ); + }); + + test('the versioned migrations apply the fence table, and reapplying it changes nothing', async () => { + const versions = ledger.migrations().map(item => item.name); + assert.ok(versions.includes('reporting_orphan_collector_fence')); + const fence = ledger.migrations().find(item => item.name === 'reporting_orphan_collector_fence'); + assert.deepEqual(fence.statements(), [ledger.REPORTING_ORPHAN_FENCE_MIGRATION]); + await livePool.query(ledger.REPORTING_ORPHAN_FENCE_MIGRATION); + await assert.rejects( + () => + livePool.query( + `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token) VALUES ($1, 0, $2)`, + ['a'.repeat(64), '33333333-3333-4333-8333-333333333333'] + ), + /check/i + ); + }); +}); diff --git a/packages/reporting-storage/test/reporting-row-orphan-collector-pg.test.js b/packages/reporting-storage/test/reporting-row-orphan-collector-pg.test.js index ab5ed5dae..fef12f2cb 100644 --- a/packages/reporting-storage/test/reporting-row-orphan-collector-pg.test.js +++ b/packages/reporting-storage/test/reporting-row-orphan-collector-pg.test.js @@ -74,6 +74,7 @@ describe('orphaned row object collector', { skip: !DATABASE_URL && 'PostgreSQL U pool = new Pool({ connectionString: DATABASE_URL, options: `-c search_path="${schema}"` }); await pool.query(ledger.REPORTING_LEDGER_MIGRATION); await pool.query(ledger.REPORTING_ROW_STORAGE_MIGRATION); + await pool.query(ledger.REPORTING_ORPHAN_FENCE_MIGRATION); const filesystem = ledger.createFilesystemReportingRowObjectProviderV1({ roots: { rows: root } }); provider = new Proxy(filesystem, { get(target, property) { @@ -86,6 +87,9 @@ describe('orphaned row object collector', { skip: !DATABASE_URL && 'PostgreSQL U } if (property === 'list') { return async (input, context) => { + // The collector's ownership fence is listed through the same provider; these hooks target the + // object listing, so the fence passes through (reporting-orphan-collector-fence-pg covers it). + if (input.prefix.endsWith('/_collector-fence/')) return target.list(input, context); const failure = listFailures.shift(); if (failure) throw failure; const real = async page => { @@ -148,6 +152,7 @@ describe('orphaned row object collector', { skip: !DATABASE_URL && 'PostgreSQL U listOverride = null; ageShiftMs = 0; await pool.query('DELETE FROM adcp_reporting_row_write_intents'); + await pool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); rmSync(path.join(root, 'orph'), { recursive: true, force: true }); rmSync(path.join(root, 'legacy'), { recursive: true, force: true }); }); From 9a1a0220b505be75c3a70330960485060e7c00c8 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 11:18:22 +0000 Subject: [PATCH 2/4] fix(reporting-storage): address review of the orphan collector fence Claim under a session lock held through the object write, finish an unfinished claim under a fresh token, recheck the fence every 25 deletes, read the bucket before the row in read-only checks, scope a fence by provider kind, and widen takeover to a damaged chain. Adds non-database tests for the decision table. Co-Authored-By: Claude Sonnet 5.5 --- docs/guides/REPORTING-OPERATIONS.md | 13 +- docs/migration-reporting-storage-package.md | 4 +- packages/reporting-storage/README.md | 20 +- packages/reporting-storage/src/maintenance.ts | 6 +- packages/reporting-storage/src/postgres.ts | 250 ++++++++++----- packages/reporting-storage/src/row-storage.ts | 30 +- ...eporting-orphan-collector-fence-pg.test.js | 302 +++++++++++++++++- .../test/reporting-orphan-fence.test.js | 131 ++++++++ .../test/reporting-storage-migrations.test.js | 1 + 9 files changed, 636 insertions(+), 121 deletions(-) create mode 100644 packages/reporting-storage/test/reporting-orphan-fence.test.js diff --git a/docs/guides/REPORTING-OPERATIONS.md b/docs/guides/REPORTING-OPERATIONS.md index 257a49088..faf57a06b 100644 --- a/docs/guides/REPORTING-OPERATIONS.md +++ b/docs/guides/REPORTING-OPERATIONS.md @@ -222,7 +222,9 @@ service runs for you. See earlier state. A `dryRun` reads the fence without claiming it, so it reports a fenced binding too and writes nothing. - **First claim wins.** Nothing in the database distinguishes a copy from the - original until one of them claims. If a clone collects before the original + original until one of them claims. When you roll the fence out, claim from the + live database first: one real pass, or `takeOverOrphanCollectorFence` (on a + namespace nobody has claimed it is a claim and nothing more). If a clone collects before the original has run a pass since the copy was taken, the clone claims first and the original is fenced, and the warning is the signal. The age threshold still holds: the clone cannot delete an object the original wrote after the copy @@ -236,6 +238,15 @@ service runs for you. See clone while the original is running: that fences the original and lets the clone delete its objects. + A fenced binding is checked again within the hour (never a whole `intervalSeconds` + later), so applying `0009` or taking the namespace over resumes collection + without a restart. A pass writes one small fence object per pass (and prunes + the older ones), so the binding's credentials need write access, including a + `readOnly` binding collected with `includeReadOnly`. Keep `_collector-fence/` out + of bucket lifecycle rules and manual cleanup: deleting the newest fence object + makes an older copy look current again. The claim holds a database connection + and an advisory lock across the object-store listing and read. + After a restore in place, expect the collector to delete the objects of the row sets the restore lost once you have taken the namespace over, which is its purpose. diff --git a/docs/migration-reporting-storage-package.md b/docs/migration-reporting-storage-package.md index 237314d34..ba17c7db1 100644 --- a/docs/migration-reporting-storage-package.md +++ b/docs/migration-reporting-storage-package.md @@ -166,14 +166,14 @@ Reporting storage is now one decision. `postgresStorage({ pool })` (this package | `subscriptions: { principals: createPrincipalLifecycle({ store, notifications, agentUrl, resolvePrincipal }), stateStore }` | `subscriptions: { principal: { notifications, agentUrl, resolvePrincipal } }` with `storage`; serve `protocol: reporting.principals.protocol` | | `createReliableReportingConsumerV1({ persistence: createPostgresReportingConsumerRuntimeV1({ db, namespace }) })` | `createReliableReportingConsumerV1({ storage })` | -The migration constants (`REPORTING_LEDGER_MIGRATION`, `REPORTING_ROW_STORAGE_MIGRATION`, `REPORTING_LEDGER_CHANGES_MIGRATION`, `REPORTING_MANAGED_DELIVERY_MIGRATION`, the object-write and finality-fence constants) are deprecated re-exports for one beta. A schema they set up migrates cleanly: the first `migrate` re-runs them as no-ops and records versions `0001`–`0008`. See the [package README](https://github.com/adcontextprotocol/adcp-client/blob/main/packages/reporting-storage/README.md#apply-migrations-npx-adcp-reporting-storage-migrate) for the version table. +The migration constants (`REPORTING_LEDGER_MIGRATION`, `REPORTING_ROW_STORAGE_MIGRATION`, `REPORTING_LEDGER_CHANGES_MIGRATION`, `REPORTING_MANAGED_DELIVERY_MIGRATION`, the object-write and finality-fence constants) are deprecated re-exports for one beta. A schema they set up migrates cleanly: the first `migrate` re-runs them as no-ops and records versions `0001`–`0009`. See the [package README](https://github.com/adcontextprotocol/adcp-client/blob/main/packages/reporting-storage/README.md#apply-migrations-npx-adcp-reporting-storage-migrate) for the version table. ### Orphan collector ownership fence (storage `0.3`, migration `0009`) Only deployments that enable `orphanCollection` (or call `collectOrphanedRowObjects`) are affected. A clone or restore of the ledger database shares the live database's namespace key and configuration, including `acknowledgeExclusiveNamespace: true`, and would delete the live database's newer row objects. A real pass now claims the namespace first and refuses to delete when another database holds a newer claim. 1. Run `npx adcp-reporting-storage migrate` (version `0009`, `reporting_orphan_collector_fence`, additive). Until it is applied, every pass, `dryRun` included, reports each binding as `fenced` with cause `migration_required` and deletes nothing. `postgresStorage().checkSchema()` names `0009` as pending. -2. Roll `0009` out and run one pass on the **live** database before any staging copy of it runs a collector. The first claim wins, and a copy has no way to tell it is a copy. +2. Roll `0009` out and claim the namespace from the **live** database before any staging copy of it runs a collector: run one real pass there, or call `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true })`, which on a namespace nobody has claimed is a claim and nothing more. The first claim wins, and a copy has no way to tell it is a copy; an already-running copy with `acknowledgeExclusiveNamespace: true` that claims first fences the live database out. 3. Handle the new result shape: `ReportingOrphanBindingResultV1.status` can be `'fenced'` (with `fence: { cause, detail, fenceEpoch?, recordedEpoch? }`), and `ReportingOrphanCollectionResultV1` has a required `fenced: string[]`. A custom store that implements `collectOrphanedRowObjects` for the maintenance task must return `fenced` (an empty array is fine). The maintenance task raises a warning for each fenced binding. 4. After restoring a database in place, or when the database that held the namespace is gone, call `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true })` once, then collection resumes. It refuses without the acknowledgement. 5. The fence chain lives under `{prefix}/{namespace_key}/_collector-fence/` in each object binding's bucket. Your provider needs nothing new (create-only `putIfAbsent`, `get`, `list` and `delete`, which the conformance suite already covers), and a bucket lifecycle rule must not expire objects under the owned prefix (the probe already refuses one that does). diff --git a/packages/reporting-storage/README.md b/packages/reporting-storage/README.md index 5562bc2ab..c22d5a45b 100644 --- a/packages/reporting-storage/README.md +++ b/packages/reporting-storage/README.md @@ -83,16 +83,16 @@ SDK's `python -m adcp migrate` uses), so concurrent runners serialize. `migrate( `pendingMigrations(pool)` and `migrations()` are the same operations in code. Startup never changes the schema: `storage.checkSchema()` names the pending versions and this command. -| Version | Name | Contents (the deprecated constant that holds the same SQL) | -| ------- | --------------------------------- | ---------------------------------------------------------------------------------------------------------- | -| 0001 | `adcp_state` | principal records and reporting-subscription engine state (`ADCP_STATE_MIGRATION` from `@adcp/sdk/server`) | -| 0002 | `reporting_ledger` | the ledger (`REPORTING_LEDGER_MIGRATION`) | -| 0003 | `reporting_ledger_finality_fence` | **stopped-worker:** fences pre-SDK-14 writers (`REPORTING_LEDGER_FINALITY_WRITER_FENCE_MIGRATION`) | -| 0004 | `reporting_row_storage` | row storage, PostgreSQL 13+ (`REPORTING_ROW_STORAGE_MIGRATION`) | -| 0005 | `reporting_ledger_changes` | the host change feed, PostgreSQL 13+ (`REPORTING_LEDGER_CHANGES_MIGRATION`) | -| 0006 | `reporting_managed_delivery` | managed delivery (`REPORTING_MANAGED_DELIVERY_MIGRATION`) | -| 0007 | `reporting_object_writes` | the GCS write fence (`REPORTING_OBJECT_WRITE_MIGRATION`, `REPORTING_OBJECT_WRITE_AUTHORITY_MIGRATION`) | -| 0008 | `reporting_consumer` | the buyer consumer (`REPORTING_CONSUMER_POSTGRES_MIGRATION` from `@adcp/sdk/reporting/consumer`) | +| Version | Name | Contents (the deprecated constant that holds the same SQL) | +| ------- | ---------------------------------- | ----------------------------------------------------------------------------------------------------------------------- | +| 0001 | `adcp_state` | principal records and reporting-subscription engine state (`ADCP_STATE_MIGRATION` from `@adcp/sdk/server`) | +| 0002 | `reporting_ledger` | the ledger (`REPORTING_LEDGER_MIGRATION`) | +| 0003 | `reporting_ledger_finality_fence` | **stopped-worker:** fences pre-SDK-14 writers (`REPORTING_LEDGER_FINALITY_WRITER_FENCE_MIGRATION`) | +| 0004 | `reporting_row_storage` | row storage, PostgreSQL 13+ (`REPORTING_ROW_STORAGE_MIGRATION`) | +| 0005 | `reporting_ledger_changes` | the host change feed, PostgreSQL 13+ (`REPORTING_LEDGER_CHANGES_MIGRATION`) | +| 0006 | `reporting_managed_delivery` | managed delivery (`REPORTING_MANAGED_DELIVERY_MIGRATION`) | +| 0007 | `reporting_object_writes` | the GCS write fence (`REPORTING_OBJECT_WRITE_MIGRATION`, `REPORTING_OBJECT_WRITE_AUTHORITY_MIGRATION`) | +| 0008 | `reporting_consumer` | the buyer consumer (`REPORTING_CONSUMER_POSTGRES_MIGRATION` from `@adcp/sdk/reporting/consumer`) | | 0009 | `reporting_orphan_collector_fence` | the orphan collector's ownership fence (`REPORTING_ORPHAN_FENCE_MIGRATION`); needed before orphan collection can delete | A schema set up by applying those constants by hand migrates cleanly: the statements are idempotent, re-run as no-ops, diff --git a/packages/reporting-storage/src/maintenance.ts b/packages/reporting-storage/src/maintenance.ts index 0e77beb1e..083495b3f 100644 --- a/packages/reporting-storage/src/maintenance.ts +++ b/packages/reporting-storage/src/maintenance.ts @@ -343,7 +343,11 @@ export async function runReportingLedgerMaintenance( !outcome.bindings.some(item => item.status === 'failed') ) { state.completed = []; - state.nextCycleAt = Date.now() + (orphanCollection.intervalSeconds ?? 86_400) * 1000; + // A fenced binding looks again within the hour, so applying the migration or taking the + // namespace over does not leave collection idle for the rest of a day. + const fenced = outcome.bindings.some(item => item.status === 'fenced'); + const rest = orphanCollection.intervalSeconds ?? 86_400; + state.nextCycleAt = Date.now() + (fenced ? Math.min(rest, 3_600) : rest) * 1000; } else { state.completed = [...new Set([...(state.completed ?? []), ...outcome.completed])]; delete state.nextCycleAt; diff --git a/packages/reporting-storage/src/postgres.ts b/packages/reporting-storage/src/postgres.ts index 78fde5d53..18a361e61 100644 --- a/packages/reporting-storage/src/postgres.ts +++ b/packages/reporting-storage/src/postgres.ts @@ -93,6 +93,7 @@ import { import { decideOrphanFence, type OrphanFenceEntry, + type OrphanFenceRecord, type OrphanFenceRow, type ReportingOrphanFenceRefusalV1, } from './orphan-fence'; @@ -2799,6 +2800,7 @@ ${managedDueArm} ) } const outcome = await this.orphanFenceGate( bindingId, + scope, mode === 'claim' && dryRun ? 'check' : mode, hasFenceTable, input.signal @@ -2889,17 +2891,10 @@ ${managedDueArm} ) old.push(entry); } const named = await this.rowObjectKeysNamedByRecords(old.map(entry => entry.objectRef)); - if (!dryRun && old.some(entry => !named.has(entry.objectRef))) { - // Another database may have claimed the namespace since this pass began: read the fence again - // before every page that deletes. - const again = await gate(summary.id, 'check'); - if (!again.ok) { - outcome.status = 'fenced'; - outcome.fence = again.refusal; - result.fenced.push(summary.id); - break; - } - } + // Another database may claim the namespace while this pass runs: read the fence again before the + // first delete of every page and then every ORPHAN_FENCE_RECHECK_DELETES deletes after it. + let sinceFenceCheck = ORPHAN_FENCE_RECHECK_DELETES; + let fencedMidPass = false; for (const entry of old) { input.signal?.throwIfAborted(); if (named.has(entry.objectRef)) { @@ -2919,6 +2914,18 @@ ${managedDueArm} ) } continue; } + if (sinceFenceCheck >= ORPHAN_FENCE_RECHECK_DELETES) { + const again = await gate(summary.id, 'check'); + if (!again.ok) { + outcome.status = 'fenced'; + outcome.fence = again.refusal; + result.fenced.push(summary.id); + fencedMidPass = true; + break; + } + sinceFenceCheck = 0; + } + sinceFenceCheck += 1; try { // The per-object lock and exact-version reference check every other delete makes. const removed = await this.deleteUnreferencedObjects( @@ -2934,6 +2941,7 @@ ${managedDueArm} ) outcome.failed += 1; } } + if (fencedMidPass) break; if (page.nextPageToken !== undefined && page.nextPageToken === token) { dropCursor = true; throw new ReportingRowStoreError('PROVIDER_UNAVAILABLE', 'provider returned the same page token twice'); @@ -2978,9 +2986,14 @@ ${managedDueArm} ) * claim this database recorded, records the next epoch and a fresh token in the database, and only then * writes that epoch's object create-only. The intent is committed before the object is written, so a crash * between the two is finished by the next pass instead of looking like another database's claim. + * + * Each real pass claims a new epoch (the `gates` cache is per call), so a cycle spread over many partial passes + * writes and prunes one tiny object per pass. The claim holds a pool connection and the advisory lock across the + * object-store listing and read, so a slow bucket pins one connection for up to the read deadline. */ private async orphanFenceGate( bindingId: string, + scope: string, mode: 'claim' | 'check', fencesTable: boolean, signal?: AbortSignal @@ -2996,70 +3009,49 @@ ${managedDueArm} ) }, }; } - const { scope } = await this.rowStorage.collectorFenceLocation(bindingId); try { if (mode === 'check') { - const row = await this.orphanFenceRow(scope); + // The bucket first, then the row. A claim commits its intent to the row before it writes the object, so + // a row read after the bucket is never behind a claim of this database that the bucket already shows + // (a sibling host claiming between the two reads would otherwise look like another database's claim). const { latest } = await this.latestOrphanFence(bindingId, signal); + const row = await this.orphanFenceRow(scope); const decision = decideOrphanFence(row?.row ?? null, latest); return decision.action === 'fenced' ? { ok: false, refusal: decision.refusal } : { ok: true }; } - const plan = await this.transaction(async client => { - await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [`adcp.orphan_fence:${scope}`]); - const row = await this.orphanFenceRow(scope, client); + const claimed = await this.claimOrphanFence(bindingId, scope, signal, async stored => { const { entries, latest } = await this.latestOrphanFence(bindingId, signal); - const decision = decideOrphanFence(row?.row ?? null, latest); - if (decision.action === 'fenced') return { kind: 'fenced', refusal: decision.refusal } as const; - if (decision.action === 'complete') { - return { - kind: 'write', - record: { - epoch: row!.row.epoch, - token: row!.row.token, - previousToken: row!.row.previousToken, - claimedAt: row!.claimedAt, - }, - entries, - } as const; - } - const record = { - epoch: decision.epoch, - token: randomUUID(), - previousToken: decision.previousToken, - claimedAt: new Date().toISOString(), + const decision = decideOrphanFence(stored?.row ?? null, latest); + if (decision.action === 'fenced') return { refusal: decision.refusal }; + // An unfinished claim is finished under a fresh token, never the recorded one: a copy of this database + // taken while the claim was unfinished carries the same token, and both could then "adopt" the same + // object. With distinct tokens, create-only leaves exactly one of them the owner. + return { + record: { + epoch: decision.action === 'complete' ? stored!.row.epoch : decision.epoch, + token: randomUUID(), + previousToken: decision.action === 'complete' ? stored!.row.previousToken : decision.previousToken, + claimedAt: new Date().toISOString(), + }, + entries, + previousEpoch: latest?.epoch ?? null, }; - await client.query( - `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token, claimed_at) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (fence_scope) DO UPDATE - SET epoch = EXCLUDED.epoch, token = EXCLUDED.token, - previous_token = EXCLUDED.previous_token, claimed_at = EXCLUDED.claimed_at`, - [scope, record.epoch, record.token, record.previousToken, record.claimedAt] - ); - return { kind: 'write', record, entries } as const; }); - if (plan.kind === 'fenced') return { ok: false, refusal: plan.refusal }; - const written = await this.rowStorage.putCollectorFence(bindingId, plan.record, signal); - if (written.record.token !== plan.record.token) { + if (claimed.kind === 'fenced') return { ok: false, refusal: claimed.refusal }; + if (claimed.kind === 'lost') { return { ok: false, refusal: { cause: 'claim_lost', detail: - `another database claimed epoch ${plan.record.epoch} first. If this database is the only one that ` + + `another database claimed epoch ${claimed.record.epoch} first. If this database is the only one that ` + 'owns this namespace, call takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); ' + 'otherwise leave orphan collection off on this copy.', - fenceEpoch: written.record.epoch, - recordedEpoch: plan.record.epoch, + fenceEpoch: claimed.found.epoch, + recordedEpoch: claimed.record.epoch, }, }; } - // Keep the claim and its predecessor; everything older is superseded. - await this.rowStorage.deleteCollectorFence( - bindingId, - plan.entries.filter(entry => entry.epoch < plan.record.epoch - 1), - signal - ); return { ok: true }; } catch (error) { // Fence objects that are not exactly what this collector writes fail closed. @@ -3079,6 +3071,90 @@ ${managedDueArm} ) } } + /** + * Claim one fence epoch for this database, under a session-level advisory lock held from the decision to the + * object write. `plan` runs inside the transaction with the stored row, reads the bucket and returns either a + * refusal or the claim to make (and the entries it supersedes). The claim's intent (epoch, token) is committed + * before the object is written, so a crash between the two is finished by the next pass instead of looking like + * another database's claim; holding the lock through the write keeps a sibling host of this database from + * finishing that claim under another token while the first write is still in flight. A clone has its own + * database and its own lock, which is exactly why the object store's create-only write decides between them. + */ + private async claimOrphanFence( + bindingId: string, + scope: string, + signal: AbortSignal | undefined, + plan: ( + stored: { row: OrphanFenceRow; claimedAt: string } | null + ) => Promise< + | { refusal: ReportingOrphanFenceRefusalV1 } + | { record: OrphanFenceRecord; entries: OrphanFenceEntry[]; previousEpoch: number | null } + > + ): Promise< + | { kind: 'fenced'; refusal: ReportingOrphanFenceRefusalV1 } + | { kind: 'lost'; record: OrphanFenceRecord; found: OrphanFenceRecord } + | { kind: 'claimed'; record: OrphanFenceRecord; previousEpoch: number | null } + > { + let client: ReportingPgClient; + try { + client = await this.pool.connect(); + } catch (cause) { + throw new Error('PostgresReportingLedgerStore database connection failed', { cause }); + } + let locked = false; + let timeoutSet = false; + let releaseError: Error | undefined; + try { + // A sibling holds the lock across its object-store write, which has its own deadline. + await client.query("SET lock_timeout = '120s'"); + timeoutSet = true; + await client.query('SELECT pg_advisory_lock(hashtextextended($1, 0))', [`adcp.orphan_fence:${scope}`]); + locked = true; + await client.query('RESET lock_timeout'); + timeoutSet = false; + await client.query('BEGIN'); + let planned: Awaited>; + try { + planned = await plan(await this.orphanFenceRow(scope, client)); + if ('record' in planned) { + await client.query( + `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token, claimed_at) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (fence_scope) DO UPDATE + SET epoch = EXCLUDED.epoch, token = EXCLUDED.token, + previous_token = EXCLUDED.previous_token, claimed_at = EXCLUDED.claimed_at`, + [scope, planned.record.epoch, planned.record.token, planned.record.previousToken, planned.record.claimedAt] + ); + } + await client.query('COMMIT'); + } catch (error) { + await client.query('ROLLBACK').catch(() => undefined); + throw error; + } + if ('refusal' in planned) return { kind: 'fenced', refusal: planned.refusal }; + const written = await this.rowStorage.putCollectorFence(bindingId, planned.record, signal); + if (written.record.token !== planned.record.token) { + return { kind: 'lost', record: planned.record, found: written.record }; + } + // Keep the claim and its predecessor; everything older is superseded. + await this.rowStorage.deleteCollectorFence( + bindingId, + planned.entries.filter(entry => entry.epoch < planned.record.epoch - 1), + signal + ); + return { kind: 'claimed', record: planned.record, previousEpoch: planned.previousEpoch }; + } finally { + try { + if (timeoutSet) await client.query('RESET lock_timeout'); + if (locked) + await client.query('SELECT pg_advisory_unlock(hashtextextended($1, 0))', [`adcp.orphan_fence:${scope}`]); + } catch (cleanupCause) { + releaseError = cleanupCause instanceof Error ? cleanupCause : new Error('Orphan fence lock release failed'); + } + client.release(releaseError); + } + } + private async orphanFenceRow( scope: string, client?: ReportingPgClient @@ -3130,7 +3206,7 @@ ${managedDueArm} ) bindings?: readonly string[]; includeReadOnly?: boolean; signal?: AbortSignal; - }): Promise<{ bindingIds: string[]; epoch: number; previousEpoch: number | null }[]> { + }): Promise { if (input?.acknowledgePreviousOwnerGone !== true) { throw new TypeError( 'takeOverOrphanCollectorFence fences out the namespace’s current owner: pass acknowledgePreviousOwnerGone: ' + @@ -3159,14 +3235,12 @@ ${managedDueArm} ) const { scope } = await this.rowStorage.collectorFenceLocation(summary.id); byScope.set(scope, [...(byScope.get(scope) ?? []), summary.id]); } - const taken: { bindingIds: string[]; epoch: number; previousEpoch: number | null }[] = []; + const taken: ReportingOrphanFenceTakeoverV1[] = []; for (const [scope, bindingIds] of byScope) { input.signal?.throwIfAborted(); const bindingId = bindingIds[0]!; - const plan = await this.transaction(async client => { - await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [`adcp.orphan_fence:${scope}`]); - const stored = await this.orphanFenceRow(scope, client); - const entries = await this.rowStorage.listCollectorFence(bindingId, input.signal); + const claimed = await this.claimOrphanFence(bindingId, scope, input.signal, async stored => { + const entries = await this.rowStorage.listCollectorFence(bindingId, input.signal, { tolerant: true }); const newest = entries[entries.length - 1]; // A malformed newest object is exactly what a takeover is for: its key still gives the epoch to supersede. const previous = newest @@ -3175,36 +3249,26 @@ ${managedDueArm} ) throw error; }) : null; - const epoch = Math.max(newest?.epoch ?? 0, stored?.row.epoch ?? 0) + 1; - const record = { - epoch, - token: randomUUID(), - previousToken: previous?.token ?? null, - claimedAt: new Date().toISOString(), + return { + record: { + epoch: Math.max(newest?.epoch ?? 0, stored?.row.epoch ?? 0) + 1, + token: randomUUID(), + previousToken: previous?.token ?? null, + claimedAt: new Date().toISOString(), + }, + entries, + previousEpoch: newest?.epoch ?? null, }; - await client.query( - `INSERT INTO adcp_reporting_orphan_collector_fences (fence_scope, epoch, token, previous_token, claimed_at) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (fence_scope) DO UPDATE - SET epoch = EXCLUDED.epoch, token = EXCLUDED.token, - previous_token = EXCLUDED.previous_token, claimed_at = EXCLUDED.claimed_at`, - [scope, record.epoch, record.token, record.previousToken, record.claimedAt] - ); - return { record, entries, previousEpoch: newest?.epoch ?? null }; }); - const written = await this.rowStorage.putCollectorFence(bindingId, plan.record, input.signal); - if (written.record.token !== plan.record.token) { + if (claimed.kind !== 'claimed') { throw new ReportingRowStoreError( 'STATE_UNAVAILABLE', - `another database claimed epoch ${plan.record.epoch} while this takeover ran; run it again` + claimed.kind === 'lost' + ? `another database claimed epoch ${claimed.record.epoch} while this takeover ran; run it again` + : 'the takeover was refused' ); } - await this.rowStorage.deleteCollectorFence( - bindingId, - plan.entries.filter(entry => entry.epoch < plan.record.epoch - 1), - input.signal - ); - taken.push({ bindingIds, epoch: plan.record.epoch, previousEpoch: plan.previousEpoch }); + taken.push({ bindingIds, epoch: claimed.record.epoch, previousEpoch: claimed.previousEpoch }); } return taken; } @@ -6534,9 +6598,23 @@ function feedConsumerBehindChange(consumer: string, change: string): string { const DEFAULT_ORPHAN_MIN_AGE_SECONDS = 7 * 24 * 3600; const MIN_ORPHAN_MIN_AGE_SECONDS = 24 * 3600; const MAX_ORPHAN_CANDIDATES = 1_000; +/** Deletes between reads of the collector fence within one listing page. */ +const ORPHAN_FENCE_RECHECK_DELETES = 25; /** The suffixes `renderReportingRowObjectKeyV1` gives every key: nothing else under the prefix is a row object. */ const ROW_OBJECT_KEY_SUFFIX = /\.jsonl(\.gz)?$/; +/** + * One fence a takeover claimed: the bindings that share it, the epoch now held and the highest epoch it replaced. + * + * @experimental Reliable Reporting storage surface: breaking changes may ship in minor releases. See https://github.com/adcontextprotocol/adcp-client/blob/main/docs/guides/REPORTING-STABILITY.md + */ +export interface ReportingOrphanFenceTakeoverV1 { + bindingIds: string[]; + epoch: number; + /** Highest epoch found in the object store before the takeover; `null` when the namespace was never claimed. */ + previousEpoch: number | null; +} + /** * @experimental Reliable Reporting storage surface: breaking changes may ship in minor releases. See https://github.com/adcontextprotocol/adcp-client/blob/main/docs/guides/REPORTING-STABILITY.md */ diff --git a/packages/reporting-storage/src/row-storage.ts b/packages/reporting-storage/src/row-storage.ts index d98da1da9..e1e71b17a 100644 --- a/packages/reporting-storage/src/row-storage.ts +++ b/packages/reporting-storage/src/row-storage.ts @@ -479,6 +479,8 @@ export class ReportingRowStorageV1 { * the object-store directory that holds its chain. Bindings that share provider, location, prefix and * installation share a fence. Throws `STATE_UNAVAILABLE` for a binding that is not an object binding, and * `INVALID_INPUT` when its provider cannot list (the fence is read by listing). + * + * @internal */ async collectorFenceLocation(bindingId: string): Promise<{ scope: string; directory: string; namespaceKey: string }> { await this.ready(); @@ -493,7 +495,9 @@ export class ReportingRowStorageV1 { const namespaceKey = this.installation!.namespaceKey; return { scope: orphanFenceScope({ - provider: binding.provider, + // The provider's own name, not this host's registry key: two registry entries of one kind that reach + // the same bucket and prefix are one physical fence. + provider: provider.name, location: binding.location, prefix: binding.prefix, namespaceKey, @@ -507,9 +511,16 @@ export class ReportingRowStorageV1 { * Every fence object of a binding's namespace, ordered by epoch. Keys under the fence directory that are * not an epoch name are ignored. Two live versions of one epoch's key (a versioned store after a manual * delete and re-create) are ambiguous and fail closed with `ROWS_INTEGRITY_FAILED`, as does a chain too - * long to be one this collector wrote (it prunes to two objects). + * long to be one this collector wrote (it prunes to two objects). `tolerant` is for a takeover, which must + * be able to supersede a damaged chain: duplicates keep their first version and the page bound is wider. + * + * @internal */ - async listCollectorFence(bindingId: string, signal?: AbortSignal): Promise { + async listCollectorFence( + bindingId: string, + signal?: AbortSignal, + options: { tolerant?: boolean } = {} + ): Promise { const { directory } = await this.collectorFenceLocation(bindingId); const binding = this.bindings[bindingId] as ReportingObjectRowBindingV1; const provider = this.providers[binding.provider]!; @@ -518,7 +529,7 @@ export class ReportingRowStorageV1 { const pageSize = 50; let token: string | undefined; for (let page = 0; ; page += 1) { - if (page >= 20) { + if (page >= (options.tolerant ? 200 : 20)) { throw new ReportingRowStoreError( 'ROWS_INTEGRITY_FAILED', 'the collector fence directory holds too many objects' @@ -550,6 +561,7 @@ export class ReportingRowStorageV1 { const epoch = orphanFenceEpochOfKey(ownedPrefix, object.objectRef); if (epoch === undefined || object.objectRef !== orphanFenceKey(ownedPrefix, epoch)) continue; if (entries.has(epoch)) { + if (options.tolerant) continue; throw new ReportingRowStoreError( 'ROWS_INTEGRITY_FAILED', 'the collector fence holds two versions of one epoch' @@ -570,6 +582,8 @@ export class ReportingRowStorageV1 { * Read and verify one fence object. A missing version is `PROVIDER_UNAVAILABLE` (it was listed a moment ago, * so this is a race to retry); content that is not exactly a fence of this namespace and epoch is * `ROWS_INTEGRITY_FAILED`. + * + * @internal */ async readCollectorFence( bindingId: string, @@ -621,6 +635,8 @@ export class ReportingRowStorageV1 { /** * Create-only write of one epoch's fence object. When the key is taken, returns the record already there * (verified), so the caller can tell its own earlier write from another database's claim. + * + * @internal */ async putCollectorFence( bindingId: string, @@ -659,7 +675,11 @@ export class ReportingRowStorageV1 { }; } - /** Delete superseded fence objects. Best effort: a failure leaves a stale object that the next claim prunes. */ + /** + * Delete superseded fence objects. Best effort: a failure leaves a stale object that the next claim prunes. + * + * @internal + */ async deleteCollectorFence( bindingId: string, entries: readonly OrphanFenceEntry[], diff --git a/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js b/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js index 001ab8f0c..306f38259 100644 --- a/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js +++ b/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js @@ -115,8 +115,11 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ } async function freshClone() { - if (clonePool) await clonePool.end(); - pools.splice(pools.indexOf(clonePool), 1); + if (clonePool) { + await clonePool.end(); + pools.splice(pools.indexOf(clonePool), 1); + clonePool = undefined; + } await copySchema(live, clone); clonePool = await newPool(clone); cloneStore = newStore(clonePool); @@ -289,10 +292,7 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ const original = await collect(liveStore); assert.deepEqual(original.fenced, ['objects']); assert.equal(objectsBinding(original).fence.cause, 'fence_advanced'); - assert.match( - require('node:util').inspect(objectsBinding(original).fence.detail), - /another database that shares this namespace has collected/ - ); + assert.match(objectsBinding(original).fence.detail, /another database that shares this namespace has collected/); }); test('a restore to an older database state is fenced until an operator takes over', async () => { @@ -523,7 +523,9 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ const result = await collect(liveStore); assert.deepEqual(result.fenced, []); assert.equal(onDisk(orphan), false); - assert.equal(fenceRecord(2).token, intent, 'the recorded intent was written as the claim'); + assert.notEqual(fenceRecord(2).token, intent, 'a finished claim carries a fresh token'); + assert.equal(fenceRecord(2).token, (await fenceRow(livePool)).token); + assert.equal(fenceRecord(2).previous_token, fenceRecord(1).token); assert.equal((await fenceRow(livePool)).epoch, 2); // The same for a crash during the very first claim. @@ -535,7 +537,8 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ [await scopeOf(liveStore), intent] ); assert.deepEqual((await collect(liveStore)).fenced, []); - assert.equal(fenceRecord(1).token, intent); + assert.notEqual(fenceRecord(1).token, intent); + assert.equal(fenceRecord(1).token, (await fenceRow(livePool)).token); }); async function scopeOf(store) { @@ -550,24 +553,19 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ assert.deepEqual([first.fenced, second.fenced], [[], []]); assert.equal(first.failed + second.failed, 0); } - // A host that finds its sibling's claim still unwritten finishes that claim instead of claiming again, - // so the epoch advances once or twice per round, never backwards. - const epoch = (await fenceRow(livePool)).epoch; - assert.ok(epoch >= 3 && epoch <= 6, `epoch ${epoch}`); - assert.ok(fenceFiles().includes(fenceName(epoch))); + // The fence lock serializes the claims of one database, so every pass claims exactly one epoch. + assert.equal((await fenceRow(livePool)).epoch, 6); + assert.deepEqual(fenceFiles(), [fenceName(5), fenceName(6)]); }); test('another database claiming mid-pass stops the deletes on the next page', async () => { for (const name of ['a', 'b', 'c']) await putObject(key(`acct/${name}.jsonl`)); - await collect(liveStore, { dryRun: true }); - let lists = 0; const intruding = new Proxy(provider, { get(target, property) { if (property === 'list') { return async (input, ctx) => { const listed = await target.list(input, ctx); if (!input.prefix.endsWith('/_collector-fence/')) { - lists += 1; if (input.pageToken?.endsWith('/acct/a.jsonl')) { // After the first object's page, another database claims the next epoch. const record = fenceRecord(1); @@ -631,4 +629,276 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ /check/i ); }); + + /** A store over `pool` whose provider can be intercepted per operation. */ + function storeWith(pool, hooks) { + const wrapped = new Proxy(provider, { + get(target, property) { + const hook = hooks[property]; + if (hook) return (input, ctx) => hook(input, ctx, () => target[property](input, ctx)); + const value = target[property]; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + return new ledger.PostgresReportingLedgerStore(pool, { + acknowledgeIsolatedDatabase: true, + rowStorage: { + bindings: { objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' } }, + providers: { filesystem: wrapped }, + selectRowBinding: () => 'objects', + }, + }); + } + const isFenceKey = input => typeof input.objectRef === 'string' && input.objectRef.includes('/_collector-fence/'); + const isFencePrefix = input => typeof input.prefix === 'string' && input.prefix.endsWith('/_collector-fence/'); + + test('a provider failure while reading or writing the fence is a failed binding, not a fenced one', async () => { + const orphan = key('acct/provider-down.jsonl'); + await putObject(orphan); + const down = () => Object.assign(new Error('provider down'), { name: 'Error' }); + const listing = storeWith(livePool, { + list: (input, ctx, next) => { + if (isFencePrefix(input)) throw down(); + return next(); + }, + }); + await listing.readyRowStorage(); + const failedList = await collect(listing, { cursors: { objects: 'orph/some-token' } }); + assert.equal(objectsBinding(failedList).status, 'failed'); + assert.equal(objectsBinding(failedList).error, 'PROVIDER_UNAVAILABLE'); + assert.deepEqual(failedList.fenced, []); + assert.equal(failedList.failed, 1); + assert.deepEqual(failedList.cursors, { objects: 'orph/some-token' }, 'the listing position is kept'); + assert.ok(onDisk(orphan)); + assert.equal(await fenceRow(livePool), undefined, 'a claim that could not read the bucket records nothing'); + + // The write of the claim fails after its intent committed. + const writing = storeWith(livePool, { + putIfAbsent: (input, ctx, next) => { + if (isFenceKey(input)) throw down(); + return next(); + }, + }); + await writing.readyRowStorage(); + const failedPut = await collect(writing); + assert.equal(objectsBinding(failedPut).status, 'failed'); + assert.deepEqual(failedPut.fenced, []); + assert.equal(failedPut.deleted, 0); + assert.ok(onDisk(orphan), 'a claim that could not be written deletes nothing'); + assert.equal((await fenceRow(livePool)).epoch, 1, 'the intent stays recorded'); + assert.deepEqual(fenceFiles(), []); + + // The next healthy pass finishes that claim, with a fresh token, and then collects. + const intent = (await fenceRow(livePool)).token; + const recovered = await collect(liveStore); + assert.deepEqual(recovered.fenced, []); + assert.equal(onDisk(orphan), false); + assert.equal(fenceRecord(1).namespace_key, namespaceKey); + assert.notEqual(fenceRecord(1).token, intent, 'a finished claim never reuses the token a copy could hold'); + assert.equal((await fenceRow(livePool)).token, fenceRecord(1).token); + + // A failure while re-reading the fence before a delete is a failed binding too. + await putObject(key('acct/second.jsonl')); + let fenceLists = 0; + const rechecking = storeWith(livePool, { + list: (input, ctx, next) => { + if (isFencePrefix(input) && (fenceLists += 1) > 1) throw down(); + return next(); + }, + }); + await rechecking.readyRowStorage(); + const failedRecheck = await collect(rechecking); + assert.equal(objectsBinding(failedRecheck).status, 'failed'); + assert.deepEqual(failedRecheck.fenced, []); + assert.equal(objectsBinding(failedRecheck).deleted, 0); + + // An aborted signal is an abort, not a refusal. + await assert.rejects(() => collect(liveStore, { signal: AbortSignal.abort() })); + }); + + test('a copy taken while a claim was unfinished cannot adopt the same token as the original', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); // epoch 1 + // The original wrote its intent for epoch 2 and never got the object out; a copy is taken in that window. + const intent = '66666666-6666-4666-8666-666666666666'; + await livePool.query( + `UPDATE adcp_reporting_orphan_collector_fences SET epoch = 2, token = $1, previous_token = $2`, + [intent, fenceRecord(1).token] + ); + await freshClone(); + const newer = key('acct/live-only.jsonl'); + await addIntent(livePool, 'rrev_live_only', newer, await putObject(newer)); + const [original, copy] = [await collect(liveStore), await collect(cloneStore)]; + assert.deepEqual(original.fenced, [], 'the first to finish the claim owns it'); + assert.deepEqual(copy.fenced, ['objects']); + assert.equal(objectsBinding(copy).fence.cause, 'token_mismatch', 'the copy does not adopt the original token'); + assert.ok(onDisk(newer)); + assert.deepEqual(fenceRecord(2).token, (await fenceRow(livePool)).token); + assert.notEqual(fenceRecord(2).token, intent); + }); + + test('a sibling host waits for a claim in flight instead of finishing it under another token', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); // epoch 1 + await putObject(key('acct/a.jsonl')); + let reached; + const atPut = new Promise(resolve => (reached = resolve)); + let release; + const held = new Promise(resolve => (release = resolve)); + const slow = storeWith(livePool, { + putIfAbsent: async (input, ctx, next) => { + if (isFenceKey(input)) { + reached(); + await held; + } + return next(); + }, + }); + await slow.readyRowStorage(); + const first = collect(slow); // commits its intent for epoch 2, then waits to write it + await atPut; + let siblingDone = false; + const sibling = collect(liveStore).then(result => { + siblingDone = true; + return result; + }); + await new Promise(resolve => setTimeout(resolve, 150)); + assert.equal(siblingDone, false, 'the sibling waits on the fence lock while the claim is in flight'); + assert.equal((await fenceRow(livePool)).epoch, 2, 'the intent is already recorded'); + release(); + const [early, late] = await Promise.all([first, sibling]); + assert.deepEqual([early.fenced, late.fenced], [[], []]); + // The first claim is epoch 2 under its own token; the sibling then claims epoch 3 on top of it. + assert.equal(fenceRecord(3).previous_token, fenceRecord(2).token); + assert.equal((await fenceRow(livePool)).epoch, 3); + assert.equal((await fenceRow(livePool)).token, fenceRecord(3).token); + }); + + test('the fence is claimed before the first delete of a pass', async () => { + const events = []; + const ordered = storeWith(livePool, { + putIfAbsent: (input, ctx, next) => { + if (isFenceKey(input)) events.push('claim'); + return next(); + }, + delete: (input, ctx, next) => { + events.push(isFenceKey(input) ? 'prune' : 'delete'); + return next(); + }, + }); + await ordered.readyRowStorage(); + for (const name of ['a', 'b']) await putObject(key(`acct/${name}.jsonl`)); + await collect(ordered); + assert.deepEqual(events, ['claim', 'delete', 'delete']); + }); + + test('a takeover recovers every refusal, and moves past a database that is ahead of its bucket', async () => { + const seed = async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await collect(liveStore); + }; + const recipes = { + claimed_elsewhere: async () => livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'), + fence_missing: async () => rmSync(fenceDirectory(), { recursive: true, force: true }), + token_mismatch: async () => + writeFileSync( + path.join(fenceDirectory(), fenceName(2)), + `${JSON.stringify({ ...fenceRecord(2), token: '77777777-7777-4777-8777-777777777777' })}\n` + ), + fence_regressed: async () => + livePool.query(`UPDATE adcp_reporting_orphan_collector_fences SET epoch = 5, previous_token = $1`, [ + '88888888-8888-4888-8888-888888888888', + ]), + fence_unreadable: async () => writeFileSync(path.join(fenceDirectory(), fenceName(2)), 'garbage'), + }; + for (const [cause, recipe] of Object.entries(recipes)) { + await beforeEachReset(); + await seed(); + await recipe(); + const refused = await collect(liveStore); + assert.equal(objectsBinding(refused).fence.cause, cause); + const taken = await liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + const expected = cause === 'fence_regressed' ? 6 : 3; + assert.equal(taken[0].epoch, expected, cause); + assert.deepEqual((await collect(liveStore)).fenced, [], `${cause} is recovered by a takeover`); + } + }); + + async function beforeEachReset() { + await livePool.query('DELETE FROM adcp_reporting_orphan_collector_fences'); + await livePool.query('DELETE FROM adcp_reporting_row_write_intents'); + rmSync(path.join(root, 'orph'), { recursive: true, force: true }); + } + + test('a takeover can be limited to named bindings, skips readOnly ones, and refuses a provider that cannot list', async () => { + const taken = await liveStore.takeOverOrphanCollectorFence({ + acknowledgePreviousOwnerGone: true, + bindings: ['objects'], + }); + assert.deepEqual(taken, [{ bindingIds: ['objects'], epoch: 1, previousEpoch: null }]); + const defaults = await liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + assert.deepEqual(defaults, [{ bindingIds: ['objects'], epoch: 2, previousEpoch: 1 }], 'readOnly twin is left out'); + + const noList = new Proxy(provider, { + get(target, property) { + if (property === 'list') return undefined; + const value = target[property]; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + const store = new ledger.PostgresReportingLedgerStore(livePool, { + acknowledgeIsolatedDatabase: true, + rowStorage: { + bindings: { objects: { kind: 'object', provider: 'filesystem', location: { root: 'rows' }, prefix: 'orph' } }, + providers: { filesystem: noList }, + selectRowBinding: () => 'objects', + }, + }); + await assert.rejects( + () => store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, bindings: ['objects'] }), + RangeError + ); + await assert.rejects( + () => liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, signal: AbortSignal.abort() }), + /abort/i + ); + }); + + test('a fenced clone refuses every binding that shares the fence, with one check and no cursor kept', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await freshClone(); + await collect(liveStore); + const refused = await collect(cloneStore, { + includeReadOnly: true, + cursors: { objects: 'orph/stale-token' }, + }); + assert.deepEqual(refused.fenced, ['objects', 'twin']); + assert.deepEqual(refused.cursors, {}, 'a refused binding does not keep a listing position'); + assert.deepEqual(refused.completed, []); + assert.equal(refused.listed, 0); + }); + + test('a fenced namespace never rests a whole interval in the maintenance cycle, and applying the fix resumes it', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await freshClone(); + await collect(liveStore); + const orphanState = { cursors: {} }; + const run = () => + ledger.runReportingLedgerMaintenance({ + store: cloneStore, + statusRetentionDays: 30, + orphanCollection: { enabled: true, acknowledgeExclusiveNamespace: true, intervalSeconds: 7 * 24 * 3600 }, + orphanState, + }); + const results = await run(); + const task = results.find(item => item.task === 'orphan object collection'); + assert.equal(task.status, 'completed'); + assert.deepEqual(task.result.fenced, ['objects']); + assert.match(task.partialFailure.message, /fence_advanced/); + const rest = orphanState.nextCycleAt - Date.now(); + assert.ok(rest > 0 && rest <= 3_600_000, `a fenced cycle looks again within the hour (${rest} ms)`); + }); }); diff --git a/packages/reporting-storage/test/reporting-orphan-fence.test.js b/packages/reporting-storage/test/reporting-orphan-fence.test.js new file mode 100644 index 000000000..55adfc794 --- /dev/null +++ b/packages/reporting-storage/test/reporting-orphan-fence.test.js @@ -0,0 +1,131 @@ +/** + * The orphan collector's fence rules without a database (adcp-client#3292): the decision table, the strict + * parse of a fence object, and the key helpers. + */ +const assert = require('node:assert/strict'); +const { describe, test } = require('node:test'); + +const { + decideOrphanFence: decide, + encodeOrphanFenceRecord: encode, + isOrphanFenceKey, + orphanFenceEpochOfKey, + orphanFenceKey, + orphanFenceScope, + parseOrphanFenceRecord: parse, +} = require('../dist/orphan-fence'); + +const A = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'; +const B = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'; +const NS = 'c'.repeat(64); +const row = (epoch, token, previousToken = null) => ({ epoch, token, previousToken }); +const cause = decision => { + assert.equal(decision.action, 'fenced'); + return decision.refusal.cause; +}; + +describe('decideOrphanFence', () => { + test('claims epoch 1 when neither side has a claim, and the next epoch when the bucket matches the record', () => { + assert.deepEqual(decide(null, null), { action: 'claim', epoch: 1, previousToken: null }); + assert.deepEqual(decide(row(3, B, A), { epoch: 3, token: B }), { action: 'claim', epoch: 4, previousToken: B }); + }); + + test('refuses a namespace claimed by a database it has no record of', () => { + assert.equal(cause(decide(null, { epoch: 1, token: A })), 'claimed_elsewhere'); + }); + + test('finishes its own unfinished claim: the object is missing, or one epoch behind with the recorded previous token', () => { + assert.deepEqual(decide(row(1, A), null), { action: 'complete' }); + assert.deepEqual(decide(row(3, B, A), { epoch: 2, token: A }), { action: 'complete' }); + }); + + test('refuses every other state, with the epochs in the refusal', () => { + assert.equal(cause(decide(row(2, B, A), null)), 'fence_missing'); + assert.equal(cause(decide(row(1, A, B), null)), 'fence_missing'); + assert.equal(cause(decide(row(3, B, A), { epoch: 3, token: A })), 'token_mismatch'); + assert.equal(cause(decide(row(1, A), { epoch: 2, token: B })), 'fence_advanced'); + // Behind its record, but not by one claim of its own. + assert.equal(cause(decide(row(3, B, A), { epoch: 2, token: B })), 'fence_regressed'); + assert.equal(cause(decide(row(3, B, null), { epoch: 2, token: A })), 'fence_regressed'); + assert.equal(cause(decide(row(5, B, A), { epoch: 2, token: A })), 'fence_regressed'); + const advanced = decide(row(1, A), { epoch: 7, token: B }); + assert.equal(advanced.refusal.fenceEpoch, 7); + assert.equal(advanced.refusal.recordedEpoch, 1); + assert.match(advanced.refusal.detail, /takeOverOrphanCollectorFence/); + }); + + test('no input makes it delete when the bucket is ahead of the record', () => { + for (const ahead of [2, 3, 10, 1_000_000]) { + assert.equal(decide(row(1, A), { epoch: ahead, token: A }).action, 'fenced'); + } + }); +}); + +describe('fence objects and keys', () => { + const record = { epoch: 4, token: A, previousToken: B, claimedAt: '2026-11-02T10:00:00.000Z' }; + const good = () => JSON.parse(Buffer.from(encode(NS, record)).toString('utf8')); + const parseObject = (value, expected = { namespaceKey: NS, epoch: 4 }) => + parse(Buffer.from(typeof value === 'string' ? value : JSON.stringify(value)), expected); + + test('round-trips, and a first claim carries a null previous token', () => { + assert.deepEqual(parse(encode(NS, record), { namespaceKey: NS, epoch: 4 }), record); + const first = { ...record, epoch: 1, previousToken: null }; + assert.deepEqual(parse(encode(NS, first), { namespaceKey: NS, epoch: 1 }), first); + }); + + test('parses strictly: anything that is not exactly a fence of this namespace and epoch is rejected', () => { + assert.ok(parseObject(good())); + const mutations = { + 'another namespace': { namespace_key: 'd'.repeat(64) }, + 'another epoch': { epoch: 5 }, + 'another version': { v: 2 }, + 'upper-case token': { token: A.toUpperCase() }, + 'token that is not a uuid': { token: 'nope' }, + 'previous token of the wrong type': { previous_token: 7 }, + 'previous token that is not a uuid': { previous_token: 'x' }, + 'an unreadable instant': { claimed_at: 'yesterday' }, + 'an extra field': { extra: true }, + 'a missing field': { token: undefined }, + }; + for (const [name, change] of Object.entries(mutations)) { + assert.equal(parseObject({ ...good(), ...change }), undefined, name); + } + for (const bytes of ['not json', 'null', '[]', '"text"', '7', '', `${JSON.stringify(good())}${' '.repeat(5000)}`]) { + assert.equal(parseObject(bytes), undefined, bytes.slice(0, 20)); + } + }); + + test('keys sort by epoch, and only exact epoch names under the fence directory count', () => { + const owned = `orph/${NS}/`; + assert.equal(orphanFenceKey(owned, 7), `${owned}_collector-fence/0000000000000007.json`); + assert.ok(orphanFenceKey(owned, 9) < orphanFenceKey(owned, 10), 'zero padding keeps them in epoch order'); + assert.equal(orphanFenceEpochOfKey(owned, orphanFenceKey(owned, 12)), 12); + for (const key of [ + `${owned}_collector-fence/000000000000007.json`, + `${owned}_collector-fence/0000000000000000.json`, + `${owned}_collector-fence/0000000000000007.jsonl`, + `${owned}_collector-fence/nested/0000000000000007.json`, + `${owned}_collector-fence/0000000000000007.json.tmp`, + `${owned}other/0000000000000007.json`, + `orph/other/_collector-fence/0000000000000007.json`, + ]) { + assert.equal(orphanFenceEpochOfKey(owned, key), undefined, key); + } + assert.ok(isOrphanFenceKey(owned, `${owned}_collector-fence/anything`)); + assert.equal(isOrphanFenceKey(owned, `${owned}acct/a.jsonl`), false); + }); + + test('a fence is one physical place: provider, location, prefix and namespace', () => { + const base = { provider: 's3', location: { bucket: 'b' }, prefix: 'rows', namespaceKey: NS }; + assert.equal(orphanFenceScope(base), orphanFenceScope({ ...base, location: { bucket: 'b' } })); + assert.match(orphanFenceScope(base), /^[0-9a-f]{64}$/); + for (const change of [ + { provider: 'gcs' }, + { location: { bucket: 'c' } }, + { prefix: 'other' }, + { namespaceKey: 'd'.repeat(64) }, + ]) { + assert.notEqual(orphanFenceScope({ ...base, ...change }), orphanFenceScope(base)); + } + }); +}); diff --git a/packages/reporting-storage/test/reporting-storage-migrations.test.js b/packages/reporting-storage/test/reporting-storage-migrations.test.js index a38e5a5f2..6ab9cb460 100644 --- a/packages/reporting-storage/test/reporting-storage-migrations.test.js +++ b/packages/reporting-storage/test/reporting-storage-migrations.test.js @@ -53,6 +53,7 @@ describe('versioned migrations', () => { require('@adcp/reporting-storage/managed').REPORTING_MANAGED_DELIVERY_MIGRATION, ]); assert.deepEqual(sql('adcp_state'), [require('@adcp/sdk/server').ADCP_STATE_MIGRATION]); + assert.deepEqual(sql('reporting_orphan_collector_fence'), [lib.REPORTING_ORPHAN_FENCE_MIGRATION]); assert.deepEqual(sql('reporting_consumer'), [ require('@adcp/sdk/reporting/consumer').REPORTING_CONSUMER_POSTGRES_MIGRATION, ]); From b18c1263972652da46e0553e83d07d942d882637 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 11:26:38 +0000 Subject: [PATCH 3/4] fix(reporting-storage): takeover removes a damaged chain, document the fence limits A takeover now supersedes every object and version it saw. Documents that a fence that advances mid-pass stops the pass at the next check (at most 25 deletes), that a takeover does not stop a running collector, and that list must be read-after-write consistent. Co-Authored-By: Claude Sonnet 5.5 --- .changeset/orphan-collector-clone-fence.md | 8 ++--- docs/guides/REPORTING-OPERATIONS.md | 7 ++-- packages/reporting-storage/src/postgres.ts | 34 ++++++++++++------- .../src/row-storage-object.ts | 4 ++- packages/reporting-storage/src/row-storage.ts | 8 +++-- ...eporting-orphan-collector-fence-pg.test.js | 27 ++++++++++++++- 6 files changed, 65 insertions(+), 23 deletions(-) diff --git a/.changeset/orphan-collector-clone-fence.md b/.changeset/orphan-collector-clone-fence.md index 336125e39..1618a2c9f 100644 --- a/.changeset/orphan-collector-clone-fence.md +++ b/.changeset/orphan-collector-clone-fence.md @@ -6,16 +6,16 @@ Reporting storage (experimental): the orphaned-object collector is fenced agains A clone or point-in-time restore of the ledger database shares the live database's installation identity, namespace key and configuration, so it also shares `acknowledgeExclusiveNamespace: true`. Its collector saw an older set of references and would have deleted the live database's newer row objects. A real `collectOrphanedRowObjects` pass now claims the namespace first and refuses to delete when another database holds a newer claim. -- **Fence.** A chain of create-only objects, one per epoch, under `{prefix}/{namespace_key}/_collector-fence/`, each holding a random token. The database records its last claim in `adcp_reporting_orphan_collector_fences`. A pass deletes only while the highest epoch in the bucket is exactly the claim the database recorded; it claims the next epoch before it deletes anything and re-reads the fence before every page that deletes. A copy taken before the original's latest claim sees a higher epoch and refuses. +- **Fence.** A chain of create-only objects, one per epoch, under `{prefix}/{namespace_key}/_collector-fence/`, each holding a random token. The database records its last claim in `adcp_reporting_orphan_collector_fences`. A pass deletes only while the highest epoch in the bucket is exactly the claim the database recorded; it claims the next epoch before it deletes anything and re-reads the fence before every page that deletes and every 25 deletes within one. A copy taken before the original's latest claim sees a higher epoch and refuses. - **Reporting.** A refused binding has `status: 'fenced'` and `fence: { cause, detail, fenceEpoch?, recordedEpoch? }`; the result has a new required `fenced: string[]`, and the maintenance task raises a warning. Causes: `fence_advanced`, `claimed_elsewhere`, `token_mismatch`, `fence_missing`, `fence_regressed`, `fence_unreadable`, `claim_lost`, `migration_required`. `dryRun` reads the fence without claiming it and reports fenced bindings too. -- **Takeover.** New `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, bindings?, includeReadOnly? })` for a database restored in place or whose predecessor is gone. It claims the next epoch and fences every other copy out. -- **Limit.** Nothing in the database tells a copy from the original until one of them claims, so the first claim wins. Run orphan collection only on the live database, never on a copy of the deployment. +- **Takeover.** New `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, bindings?, includeReadOnly? })` for a database restored in place or whose predecessor is gone. It claims the next epoch and fences every other copy out; stop the previous owner's collector first, because a pass already running notices the new epoch at its next fence check (at most 25 deletes later). +- **Limit.** Nothing in the database tells a copy from the original until one of them claims, so the first claim wins. Run orphan collection only on the live database, never on a copy of the deployment, and claim from the live database first when you roll this out. A fenced binding is checked again within the hour. Migration note (minor, breaking only for orphan collection): 1. Run `npx adcp-reporting-storage migrate`: new version `0009`, `reporting_orphan_collector_fence` (additive; the SQL is exported as `REPORTING_ORPHAN_FENCE_MIGRATION`). Until it is applied, a pass reports every binding as `fenced` with cause `migration_required` and deletes nothing, and `postgresStorage().checkSchema()` names `0009` as pending. 2. Run one real pass on the live database before any staging copy runs a collector. 3. A custom store that implements `collectOrphanedRowObjects` for the maintenance task must return `fenced: []`. `ReportingOrphanBindingResultV1.status` gains `'fenced'`. -4. Fence objects live in your object-store prefix. Provider contract is unchanged (create-only `putIfAbsent`, `get`, `list`, `delete`). +4. Fence objects live in your object-store prefix. The provider contract is unchanged (create-only `putIfAbsent`, `get`, `list`, `delete`), except that `list` must now reflect keys `putIfAbsent` has already returned (read-after-write), as the bundled providers do. Run `runReportingRowObjectProviderConformanceV1` on a custom provider. See `docs/guides/REPORTING-OPERATIONS.md` and `docs/migration-reporting-storage-package.md`. diff --git a/docs/guides/REPORTING-OPERATIONS.md b/docs/guides/REPORTING-OPERATIONS.md index faf57a06b..fe3743e91 100644 --- a/docs/guides/REPORTING-OPERATIONS.md +++ b/docs/guides/REPORTING-OPERATIONS.md @@ -211,7 +211,7 @@ service runs for you. See deletes anything, and prunes the claims older than the previous one. - **Check.** A pass deletes only while the highest epoch in the bucket is exactly the claim this database recorded. It re-reads the fence before every - page that deletes. Anything else refuses, deletes nothing and reports the + page that deletes (and every 25 deletes within one). Anything else refuses, deletes nothing more and reports the binding as `fenced` with a `cause` (`fence_advanced`, `claimed_elsewhere`, `token_mismatch`, `fence_missing`, `fence_regressed`, `fence_unreadable`, `claim_lost`, `migration_required`), in the result's `fenced` list and as a @@ -234,7 +234,10 @@ service runs for you. See - **Takeover.** After restoring a database in place, or when the original is gone, call `store.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true })` once. It claims the next epoch for this database and fences every other - copy out. It refuses without the acknowledgement. Do not call it from a + copy out. It refuses without the acknowledgement. Stop the previous owner's + collector first: a pass already running notices the new epoch at its next + fence check, up to 25 deletes later, and the provider cannot refuse a delete + on the fence's behalf. Do not call it from a clone while the original is running: that fences the original and lets the clone delete its objects. diff --git a/packages/reporting-storage/src/postgres.ts b/packages/reporting-storage/src/postgres.ts index 18a361e61..af41ef468 100644 --- a/packages/reporting-storage/src/postgres.ts +++ b/packages/reporting-storage/src/postgres.ts @@ -2708,7 +2708,8 @@ ${managedDueArm} ) * and a row in `adcp_reporting_orphan_collector_fences`) and deletes only while the object store's * highest claim is exactly the one this database recorded. A database copied before the original's last * pass sees a newer claim and refuses: its binding is reported `fenced` with a `cause`, in `fenced`, and - * in the maintenance warnings, and nothing is deleted. A `dryRun` checks the fence without claiming it + * in the maintenance warnings, and nothing more is deleted (a fence that advances mid-pass stops the pass + * at the next check, within 25 deletes, and `deleted` keeps what went before). A `dryRun` checks the fence without claiming it * and reports fenced bindings the same way. After a restore, or when the original is gone, take the * namespace over deliberately with `takeOverOrphanCollectorFence`. Everything the listing finds under * the owned prefix with a row-object key is treated as written by this SDK, so do not store other @@ -3019,7 +3020,7 @@ ${managedDueArm} ) const decision = decideOrphanFence(row?.row ?? null, latest); return decision.action === 'fenced' ? { ok: false, refusal: decision.refusal } : { ok: true }; } - const claimed = await this.claimOrphanFence(bindingId, scope, signal, async stored => { + const claimed = await this.claimOrphanFence(bindingId, scope, signal, true, async stored => { const { entries, latest } = await this.latestOrphanFence(bindingId, signal); const decision = decideOrphanFence(stored?.row ?? null, latest); if (decision.action === 'fenced') return { refusal: decision.refusal }; @@ -3084,6 +3085,7 @@ ${managedDueArm} ) bindingId: string, scope: string, signal: AbortSignal | undefined, + keepPredecessor: boolean, plan: ( stored: { row: OrphanFenceRow; claimedAt: string } | null ) => Promise< @@ -3136,10 +3138,11 @@ ${managedDueArm} ) if (written.record.token !== planned.record.token) { return { kind: 'lost', record: planned.record, found: written.record }; } - // Keep the claim and its predecessor; everything older is superseded. + // A claim keeps its predecessor (an unfinished claim is recognised by it) and prunes everything older. A + // takeover has just written and confirmed its object, so it supersedes every object and version it saw. await this.rowStorage.deleteCollectorFence( bindingId, - planned.entries.filter(entry => entry.epoch < planned.record.epoch - 1), + planned.entries.filter(entry => entry.epoch < planned.record.epoch - (keepPredecessor ? 1 : 0)), signal ); return { kind: 'claimed', record: planned.record, previousEpoch: planned.previousEpoch }; @@ -3197,6 +3200,9 @@ ${managedDueArm} ) * taking over from a clone fences the original and lets the clone delete its objects: the acknowledgement * exists because nothing else can tell them apart. `acknowledgePreviousOwnerGone: true` is required. * + * A takeover does not stop a collector that is already mid-pass: that pass notices the new epoch at its next + * fence check, which is at most 25 deletes later. Stop the previous owner's collector before you take over. + * * Takes over every writable (or, with `includeReadOnly`, every) object binding whose provider can list, or * only the `bindings` named; bindings that share a bucket, prefix and installation share one fence and are * reported together. Requires the `0009` migration. @@ -3239,16 +3245,18 @@ ${managedDueArm} ) for (const [scope, bindingIds] of byScope) { input.signal?.throwIfAborted(); const bindingId = bindingIds[0]!; - const claimed = await this.claimOrphanFence(bindingId, scope, input.signal, async stored => { + const claimed = await this.claimOrphanFence(bindingId, scope, input.signal, false, async stored => { const entries = await this.rowStorage.listCollectorFence(bindingId, input.signal, { tolerant: true }); const newest = entries[entries.length - 1]; // A malformed newest object is exactly what a takeover is for: its key still gives the epoch to supersede. - const previous = newest - ? await this.rowStorage.readCollectorFence(bindingId, newest, input.signal).catch(error => { - if (isReportingRowStoreError(error) && error.code === 'ROWS_INTEGRITY_FAILED') return null; - throw error; - }) - : null; + // Several versions of the newest epoch are ambiguous, so none of them names the previous token. + const previous = + newest && entries.filter(entry => entry.epoch === newest.epoch).length === 1 + ? await this.rowStorage.readCollectorFence(bindingId, newest, input.signal).catch(error => { + if (isReportingRowStoreError(error) && error.code === 'ROWS_INTEGRITY_FAILED') return null; + throw error; + }) + : null; return { record: { epoch: Math.max(newest?.epoch ?? 0, stored?.row.epoch ?? 0) + 1, @@ -6625,7 +6633,7 @@ export interface ReportingOrphanBindingResultV1 { * `failed`: the listing or its provider failed (`error` is the stable code). `unsupported`: the * binding's provider has no `list`. `read_only`: skipped because the binding is `readOnly`. `fenced`: * another database holds the namespace's collector fence (or this database cannot prove it holds it), so - * nothing was deleted; `fence` says why. + * no further object was deleted (`deleted` counts any that went before a mid-pass refusal); `fence` says why. */ status: 'complete' | 'partial' | 'failed' | 'unsupported' | 'read_only' | 'fenced'; /** Objects the provider listed, including ones dropped as foreign. */ @@ -6661,7 +6669,7 @@ export interface ReportingOrphanCollectionResultV1 { unsupported: string[]; /** `readOnly` bindings skipped. */ skippedReadOnly: string[]; - /** Bindings whose collector fence another database holds: nothing was deleted there. See each binding's `fence`. */ + /** Bindings whose collector fence another database holds: no further object was deleted there. See each binding's `fence`. */ fenced: string[]; /** Dry run only: the orphans a real pass would delete, at most 1000. */ candidates: { diff --git a/packages/reporting-storage/src/row-storage-object.ts b/packages/reporting-storage/src/row-storage-object.ts index d60cfbcce..88f472c89 100644 --- a/packages/reporting-storage/src/row-storage-object.ts +++ b/packages/reporting-storage/src/row-storage-object.ts @@ -161,7 +161,9 @@ export interface ReportingRowObjectProviderV1 { * too, never delete markers), the stored size, and the time the version was written. Return only keys * under `prefix`. A page may be short; `nextPageToken` is absent on the last page. A token is only * valid for the same `prefix` and location. Leave out an entry that cannot be addressed or aged (no - * usable version, size or write time) and count it in `skipped` rather than failing the page. + * usable version, size or write time) and count it in `skipped` rather than failing the page. The listing must + * be strongly consistent for keys it has already returned from `putIfAbsent` (read-after-write): the orphan + * collector's ownership fence reads the newest claim by listing. */ list?( input: ReportingRowObjectListInputV1, diff --git a/packages/reporting-storage/src/row-storage.ts b/packages/reporting-storage/src/row-storage.ts index e1e71b17a..9b8be8a9a 100644 --- a/packages/reporting-storage/src/row-storage.ts +++ b/packages/reporting-storage/src/row-storage.ts @@ -512,7 +512,8 @@ export class ReportingRowStorageV1 { * not an epoch name are ignored. Two live versions of one epoch's key (a versioned store after a manual * delete and re-create) are ambiguous and fail closed with `ROWS_INTEGRITY_FAILED`, as does a chain too * long to be one this collector wrote (it prunes to two objects). `tolerant` is for a takeover, which must - * be able to supersede a damaged chain: duplicates keep their first version and the page bound is wider. + * be able to supersede a damaged chain: every version of every epoch is returned (so it can all be removed) and + * the page bound is wider. * * @internal */ @@ -526,6 +527,7 @@ export class ReportingRowStorageV1 { const provider = this.providers[binding.provider]!; const ownedPrefix = `${binding.prefix}/${this.installation!.namespaceKey}/`; const entries = new Map(); + const versions: OrphanFenceEntry[] = []; const pageSize = 50; let token: string | undefined; for (let page = 0; ; page += 1) { @@ -560,6 +562,8 @@ export class ReportingRowStorageV1 { } const epoch = orphanFenceEpochOfKey(ownedPrefix, object.objectRef); if (epoch === undefined || object.objectRef !== orphanFenceKey(ownedPrefix, epoch)) continue; + if (options.tolerant) + versions.push({ epoch, objectRef: object.objectRef, nativeVersionRef: object.nativeVersionRef }); if (entries.has(epoch)) { if (options.tolerant) continue; throw new ReportingRowStoreError( @@ -575,7 +579,7 @@ export class ReportingRowStorageV1 { } token = listed.nextPageToken; } - return [...entries.values()].sort((a, b) => a.epoch - b.epoch); + return (options.tolerant ? versions : [...entries.values()]).sort((a, b) => a.epoch - b.epoch); } /** diff --git a/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js b/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js index 306f38259..a07e2d943 100644 --- a/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js +++ b/packages/reporting-storage/test/reporting-orphan-collector-fence-pg.test.js @@ -322,9 +322,11 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ ); assert.ok(onDisk(newer), 'a refused takeover changes nothing'); + const tokenBeforeTakeover = fenceRecord(3).token; const taken = await liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); assert.deepEqual(taken, [{ bindingIds: ['objects'], epoch: 4, previousEpoch: 3 }]); - assert.equal(fenceRecord(4).previous_token, fenceRecord(3).token); + assert.equal(fenceRecord(4).previous_token, tokenBeforeTakeover); + assert.deepEqual(fenceFiles(), [fenceName(4)], 'a takeover supersedes every older fence object'); const resumed = await collect(liveStore); assert.deepEqual(resumed.fenced, []); assert.equal(objectsBinding(resumed).status, 'complete'); @@ -831,6 +833,29 @@ describe('orphan collector ownership fence', { skip: !DATABASE_URL && 'PostgreSQ rmSync(path.join(root, 'orph'), { recursive: true, force: true }); } + test('a takeover removes every version of a duplicated chain, so normal collection works again', async () => { + await putObject(key('acct/seed.jsonl')); + await collect(liveStore); + await collect(liveStore); + const duplicated = storeWith(livePool, { + list: async (input, ctx, next) => { + const listed = await next(); + return isFencePrefix(input) + ? { + ...listed, + objects: [...listed.objects, ...listed.objects.map(entry => ({ ...entry, nativeVersionRef: 'v2' }))], + } + : listed; + }, + }); + await duplicated.readyRowStorage(); + assert.equal(objectsBinding(await collect(duplicated)).fence.cause, 'fence_unreadable'); + const taken = await duplicated.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true }); + assert.equal(taken[0].epoch, 3); + assert.deepEqual(fenceFiles(), [fenceName(3)], 'every superseded object is gone, the predecessor included'); + assert.deepEqual((await collect(liveStore)).fenced, [], 'a clean listing no longer refuses'); + }); + test('a takeover can be limited to named bindings, skips readOnly ones, and refuses a provider that cannot list', async () => { const taken = await liveStore.takeOverOrphanCollectorFence({ acknowledgePreviousOwnerGone: true, From c0b705559de841ffd69b7577ef43a28e095f6729 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 11:45:51 +0000 Subject: [PATCH 4/4] chore(packaging): raise the reporting-storage packed-byte ceiling for the orphan collector fence The fence adds about 22 KB packed (a new module, the claim and takeover, the row storage fence I/O and their declarations), over the 15 KB per-PR allowance. The ceiling goes from 490,000 to 520,000 so two further PRs still fit. Co-Authored-By: Claude Sonnet 5.5 --- scripts/package-size-budget.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/package-size-budget.json b/scripts/package-size-budget.json index 761b4bbf3..e55431071 100644 --- a/scripts/package-size-budget.json +++ b/scripts/package-size-budget.json @@ -20,7 +20,7 @@ }, "@adcp/reporting-storage": { "ceiling": { - "packedBytes": 490000, + "packedBytes": 520000, "unpackedBytes": 2250000, "files": 150 },