diff --git a/.changeset/remove-object-lock.md b/.changeset/remove-object-lock.md new file mode 100644 index 000000000..dcd38a836 --- /dev/null +++ b/.changeset/remove-object-lock.md @@ -0,0 +1,11 @@ +--- +'@adcp/reporting-storage': patch +--- + +Reporting storage (experimental): retirement, compaction and the write-intent sweep no longer take the per-object advisory lock, and a commit only takes it for an adopted object its own intent did not record (#3281). + +The lock (`adcp-reporting-object:{binding}:{objectRef}`) made every reference-checked delete a transaction that held a pooled connection and an advisory lock across the provider delete, and made every commit that adopted an existing object pay one lock round trip per object. Since every writable key template must include `{revision_key}`, only one row set can ever name a key, so the lock no longer ordered anything on those paths. Safety there comes from the reference check plus the fences that already existed: the account lock with the `FOR SHARE` on a header-only revision's predecessor, the `pruning` state each retirement and compaction commits before it deletes, the intent's `open` to `sweeping` transition against the commit that closes it, and version-pinned deletes. The full argument, case by case, is in `docs/guides/REPORTING-LEDGER.md` and on `deleteUnreferencedObjects`. + +Two places keep the lock, because nothing else orders them: the orphan collector's delete (its object is named by no record) and a commit that adopts an object its own write intent did not record (an earlier attempt's `PUT` landed before it was recorded). The collector now takes every binding's lock for the object in one transaction, as before. A retry that adopts an object its own intent recorded takes no lock and still proves it with the version-pinned `stat`. + +Why patch and not minor: no exported API, no schema and no wire behavior changes, and the lock key text is unchanged, so a rolling upgrade between a version that locks everywhere and this one is safe (every pair that still needs mutual exclusion keeps the lock on both sides). Nothing to migrate. The Python SDK carries the same item (adcp-client-python#1492); the exact final contract is under "Cross-SDK contract" in `docs/guides/REPORTING-LEDGER.md`. diff --git a/docs/guides/REPORTING-LEDGER.md b/docs/guides/REPORTING-LEDGER.md index 2010f53dd..8596bd052 100644 --- a/docs/guides/REPORTING-LEDGER.md +++ b/docs/guides/REPORTING-LEDGER.md @@ -744,16 +744,71 @@ await store.readyRowStorage(); several row sets. Every delete path (period retirement, compaction and the intent sweep) therefore skips an object while a live row set outside the operation still references the same binding, key and native version, or an - open write intent created it. (With unique keys, a create-only write only - finds an existing object when a writer retries the same row set.) The check - and delete run under the per-object lock described below. The reference check and the delete run in one - transaction under a per-object advisory lock, and a commit that adopts an - existing object takes the same lock, proves the object is still there and - holds the lock until its chunk rows commit. A delete that wins the lock makes the adopting - commit fail with `STATE_UNAVAILABLE` (the writer re-uploads once its intent - lapses); a commit that wins is seen by the delete, which skips. A skipped - object is deleted by the retirement, compaction or sweep of the last row set - that referenced it, so it is not leaked. + open write intent created it. With unique keys, a create-only write only + finds an existing object when a writer retries the same row set. +- **Object deletes and commits take no per-object lock, with one exception.** + Retirement, compaction and the intent sweep run the reference check as a + plain query, then delete the exact native version, holding no transaction or + pooled connection across the provider call. They are safe without a lock + because of five fences, none of which is the lock: + 1. Every writable key includes `{revision_key}` (a SHA-256 of the row set ID) + and `{content_sha256}`, so one row set alone can ever name a key through a + writable binding. A committed row set is replayed: a late writer that + passed the replay check can adopt its just-committed object, but its + commit inserts nothing (the row set exists), so it cannot reference an + object a retirement or compaction then deletes. A `readOnly` binding's old + template can map several row sets to one object, but nothing adopts through + it; there the reference check and fence 4 are the whole argument. Bindings + that share one bucket and prefix must route a row set to one binding across + attempts, because the reference check is per binding (the orphan collector + checks every binding). + 2. A header-only revision's commit holds the account lock and `FOR SHARE`-locks + a predecessor that must still be `live`. Retirement and compaction mark a + row set `pruning` under the same account lock before they delete anything, + and a `pruning` row set neither counts as a reference nor can be shared. So + a sharer commits first (its chunk rows then make the delete skip) or fails + with `STATE_UNAVAILABLE` and re-uploads under its own key. + 3. The intent sweep moves an expired `open` intent to `sweeping` under the + account lock; a writer closes that intent, still `open`, in the commit + transaction that inserts its chunk rows. Exactly one wins, so a writer past + its lease cannot commit objects the sweep deletes, and a retry cannot take + over an intent being swept. A takeover renews the expiry, so the sweep + cannot claim it. + 4. Concurrent retirements and compactions count only row sets that are not + `pruning`, each after committing its own `pruning` state, so at least one + of any two deletes a shared object and none can both skip it. + 5. Deletes name an exact native version, so one cannot remove an object + re-created at the same key. This needs write-unique native versions, a + provider requirement the conformance check `write-unique-version` tests. + + The advisory lock `adcp-reporting-object:{binding_id}:{object_ref}` stays in + two places that pair with each other, because nothing else orders them. A + commit that adopts an object its own intent did not record (an earlier + attempt's `PUT` landed before the intent recorded it, or a writer created it + after a takeover) takes the lock, in key order, before it proves the object with + a version-pinned `stat`, and holds it until its chunk rows commit. An object its own intent recorded takes + no lock: until the transaction commits, the open intent names it. The orphan + collector deletes only objects no record names, so each of its deletes takes + the lock of every binding (they can share a bucket and prefix), repeats the + reference check under it, and then deletes. A collector that wins makes the + adopting commit fail its version-pinned `stat` with `STATE_UNAVAILABLE` (the + writer re-uploads once its intent lapses); a commit that wins is seen by the + collector, which skips. The collector's clone fence is a separate protection: + an advisory lock is per database and cannot order a clone against the + original. A skipped object is deleted by the retirement, compaction or sweep + of the last row set that referenced it, so it is not leaked. +- **Cross-SDK contract.** A ledger database has one writer SDK (the design record's + ยง5), so the Python and JavaScript SDKs need not agree on a lock to coexist; this + is the intended contract for both, with the same key text. The Python side + (adcp-client-python#1492) mirrors it: (1) the delete paths above take no object lock; (2) the key text + is `adcp-reporting-object:` + binding ID + `:` + object ref, hashed with + `hashtextextended(text, 0)` for `pg_advisory_xact_lock`; (3) an adopting commit + locks only adopted objects absent from its own intent's `created_objects` + (matching on object ref and native version), in object-ref order, after the + account lock and after closing its intent, then proves each adopted object with + a version-pinned `stat`; (4) the orphan collector locks every binding's key for + the object, sorted by binding ID, in the transaction that repeats the + reference check and deletes. Lock order is account lock, then object locks. - An upload runs only while the obligation lease still covers `writeDeadlineMilliseconds` plus five seconds. A fenced write intent records every object the upload created, and the ledger commit closes the intent in @@ -780,7 +835,7 @@ await store.readyRowStorage(); absent, without reading the body. The commit transaction uses it to prove an adopted object is unchanged (native versions are write-unique, so the same version at the recorded size is the same bytes) instead of downloading it - while holding the account and object locks. A provider without `stat` falls + while holding the account lock. A provider without `stat` falls back to reading the whole object there. All bundled providers implement it. Also implement the optional `list({ prefix, pageToken, pageSize })`, which returns `{ objects: [{ objectRef, nativeVersionRef, sizeBytes, createdAt }], diff --git a/docs/guides/REPORTING-OPERATIONS.md b/docs/guides/REPORTING-OPERATIONS.md index fe3743e91..c675a7cd5 100644 --- a/docs/guides/REPORTING-OPERATIONS.md +++ b/docs/guides/REPORTING-OPERATIONS.md @@ -180,10 +180,13 @@ service runs for you. See `minAgeSeconds` ago (default 7 days, never less than 1 day, which must stay well above the write deadline plus intent expiry); no row set that is not yet `pruned` has a chunk at that key and no write intent lists it, at any version; - and, under the per-object advisory lock every other delete takes, no `live` or - `unavailable` row set references that exact version and no open intent - created it, checked under every binding because two bindings can share a - bucket and prefix (a retired binding kept `readOnly` next to its successor). + and, under the per-object advisory lock (the only delete that takes it: a + commit that adopts an object its own intent did not record takes the same + lock, so the two cannot interleave), no `live` or `unavailable` row set + references that exact version and no open intent created it, checked under + every binding because two bindings can share a bucket and prefix (a retired + binding kept `readOnly` next to its successor). Retirement, compaction and the + intent sweep take no lock and hold no connection across their provider deletes. `readOnly` bindings are skipped unless `includeReadOnly: true`. A binding whose provider has no `list` is reported as `unsupported` (and as a maintenance warning), never silently skipped. Everything under the owned prefix diff --git a/docs/migration-reporting-row-storage-alignment.md b/docs/migration-reporting-row-storage-alignment.md index be8ad91b3..0ffdf96b9 100644 --- a/docs/migration-reporting-row-storage-alignment.md +++ b/docs/migration-reporting-row-storage-alignment.md @@ -62,8 +62,10 @@ read-only binding is exempt from the requirement, is never chosen for new writes, and is still used for reads and retirement (`readyRowStorage` still probes it, so keep its location reachable). Compaction ignores read-only bindings, because nothing can adopt through them. The -reference-checked deletes and the per-object lock are unchanged, because -header-only revisions still share chunks within an obligation. +reference-checked deletes are unchanged, because header-only revisions still +share chunks within an obligation. (The per-object lock this note once said was +unchanged has since been removed from every path but the orphan collector; see +`docs/guides/REPORTING-LEDGER.md`.) ## Renamed fields diff --git a/packages/reporting-storage/src/postgres.ts b/packages/reporting-storage/src/postgres.ts index af41ef468..622691e90 100644 --- a/packages/reporting-storage/src/postgres.ts +++ b/packages/reporting-storage/src/postgres.ts @@ -2219,9 +2219,10 @@ ${managedDueArm} ) { preBeginAdvisoryLock: accountLock(accountId) } ); // Phase 2: external bytes, at their recorded versions. Row sets written under an older - // template without {revision_key} can name an object another obligation or account also - // references, so each delete is skipped while a live row set outside this period still - // references it (see deleteUnreferencedObjects). + // template without {revision_key} (readOnly bindings only) can name an object another obligation + // or account also references, so each delete is skipped while a live row set outside this period + // still references it. No lock orders that: the `pruning` state committed in phase 1 does (see + // deleteUnreferencedObjects). if (rowStorage) { const objects = await this.query<{ row_binding_id: string; object_ref: string; native_version_ref: string }>( `SELECT DISTINCT row_set.row_binding_id, chunk.object_ref, chunk.native_version_ref @@ -2603,29 +2604,14 @@ ${managedDueArm} ) async client => { // An older writer still running during a rolling upgrade may append the pre-alignment keys // (`objectKey`, `nativeVersion`); read both so its objects are still swept. - const updated = await client.query<{ - created_objects: { - object_ref?: string; - native_version_ref?: string; - objectKey?: string; - nativeVersion?: string; - }[]; - }>( + const updated = await client.query<{ created_objects: RecordedIntentObject[] }>( `UPDATE adcp_reporting_row_write_intents SET state = 'sweeping' WHERE row_binding_id = $1 AND row_set_id = $2 AND content_sha256 = $3 AND (state = 'sweeping' OR expires_at <= clock_timestamp()) RETURNING created_objects`, key ); - return updated.rows[0]?.created_objects - .map(object => ({ - objectRef: object.object_ref ?? object.objectKey, - nativeVersionRef: object.native_version_ref ?? object.nativeVersion, - })) - .filter( - (object): object is { objectRef: string; nativeVersionRef: string } => - typeof object.objectRef === 'string' && typeof object.nativeVersionRef === 'string' - ); + return updated.rows[0] ? recordedIntentObjects(updated.rows[0].created_objects) : undefined; }, { preBeginAdvisoryLock: accountLock(intent.account_id) } ); @@ -2690,7 +2676,8 @@ ${managedDueArm} ) * - no row set that is not already `pruned` has a chunk at its key, and no write intent lists it, * at any version and under any binding (a pre-check that errs towards leaking, because two * bindings can share a bucket and prefix); - * - under the per-object advisory lock, no `live` or `unavailable` row set under any binding + * - under the per-object advisory lock of every binding (taken only here and by a commit that adopts + * an object its own intent did not record), no `live` or `unavailable` row set under any binding * references that exact version and no open intent created it (the check every other delete makes, * widened to every binding), and the provider deletes that exact version. * @@ -2928,14 +2915,14 @@ ${managedDueArm} ) } sinceFenceCheck += 1; try { - // The per-object lock and exact-version reference check every other delete makes. - const removed = await this.deleteUnreferencedObjects( + // The only delete that takes the per-object lock: no record names this object, so a commit may be + // adopting it unrecorded. The exact-version reference check runs under that lock. + const removed = await this.deleteOrphanedObject( summary.id, - [{ objectRef: entry.objectRef, nativeVersionRef: entry.nativeVersionRef }], - { anyBinding: true }, + { objectRef: entry.objectRef, nativeVersionRef: entry.nativeVersionRef }, input.signal ); - if (removed > 0) outcome.deleted += 1; + if (removed) outcome.deleted += 1; else outcome.referenced += 1; } catch { input.signal?.throwIfAborted(); @@ -3303,92 +3290,142 @@ ${managedDueArm} ) } /** - * Delete external row objects, skipping any object another row set still references. + * Delete external row objects, skipping any object another row set still references. Takes no + * advisory lock and holds no transaction (or pooled connection) across the provider delete. + * + * An object is deleted only when no chunk of a live (or unavailable) row set outside this + * operation references the same binding, key and native version, and no other open write intent + * created it. Row sets already `pruning` or `pruned` do not count: their own operation deletes + * the object, and counting them would let two concurrent retirements each skip the object and + * leak it. `scope` names what this operation is deleting: an obligation (retirement) or a row set + * (compaction); the intent sweep passes neither. + * + * Why the check needs no lock against a committing writer (the proof, case by case): * - * Object keys are unique per row set (every writable template includes `{revision_key}`), but header-only - * revisions share chunks within an obligation and rows written under an older template can name one - * object from several row sets. An object is therefore deleted only when no - * chunk of a live (or unavailable) row set outside this operation references the same binding, - * key and native version, and no other open write intent created it. Row sets already `pruning` - * or `pruned` do not count: their own operation deletes the object, and counting them would let - * two concurrent retirements each skip the object and leak it. `scope` names what this - * operation is deleting: an obligation (retirement) or a row set (compaction). + * - Every writable template includes `{revision_key}` (a SHA-256 of the row set ID) and + * `{content_sha256}`, so only one row set can ever name a given key through a writable binding. + * Whatever deletes an object named by a committed chunk row or an open intent is deleting an object + * only that row set could commit. A committed row set is replayed: a late writer that passed the + * replay check can still adopt its just-committed object, but its commit inserts nothing (the row set + * exists), so it cannot reference an object a retirement or compaction then deletes. A `readOnly` + * binding's old template can map several row sets to one object, but nothing adopts through it; there + * the reference check and the `pruning` order below are the whole argument (and no lock ever covered + * more). Bindings that share one bucket and prefix must route a row set to one binding across + * attempts: the reference check is per binding outside the orphan collector. + * - Header-only revisions share chunks within an obligation. The sharing commit holds the account + * lock and `FOR SHARE`-locks a predecessor that must still be `live`; retirement and compaction + * move a row set to `pruning` under the same account lock before they delete anything. So a + * sharer either commits first (its chunk rows are then visible to the check, which skips the + * delete) or fails with `STATE_UNAVAILABLE`. + * - The intent sweep moves an expired `open` intent to `sweeping` under the account lock, and the + * writer closes that intent, still `open`, in the commit transaction that inserts its chunk + * rows. Exactly one wins; a writer past its lease cannot commit objects the sweep deletes. + * - Concurrent retirements and compactions are ordered by the state each commits before its check. + * - Deletes are pinned to an exact native version, so one cannot remove a re-created object. * - * The check and the provider delete run in one transaction under a per-object advisory lock. - * A commit that adopts an existing object takes the same lock before it inserts its chunk rows - * and re-verifies the object under it (`writePreparedRowSet`), and holds it until commit. So - * either the delete wins the lock and the adopting commit then fails cleanly, or the commit wins - * and the delete sees its chunk rows and skips. A skipped object stays for the row set that owns - * it: that row set's own retirement, compaction or sweep deletes it, because by then nothing - * else references it. Locks are taken one object at a time, never while waiting on the account - * lock, so commits (account lock, then object locks) cannot deadlock with these deletes. + * The one delete that still needs a lock is of an object no record names + * (`collectOrphanedRowObjects`): `deleteOrphanedObject` takes the per-object lock that a commit + * adopting an unrecorded object also takes (`writePreparedRowSet`). + * + * A skipped object stays for the row set that owns it: that row set's own retirement, + * compaction or sweep deletes it, because by then nothing else references it. */ private async deleteUnreferencedObjects( bindingId: string, objects: readonly { objectRef: string; nativeVersionRef: string }[], - scope: { obligationId?: string; rowSetId?: string; anyBinding?: boolean }, + scope: { obligationId?: string; rowSetId?: string }, signal?: AbortSignal ): Promise { let deleted = 0; + const pool: RowObjectQueryable = { query: (text, values) => this.query(text, values) }; for (const object of objects) { signal?.throwIfAborted(); - const removed = await this.transaction(async client => { - // A writer adopting an object takes the lock of its own binding only. When bindings can share a - // bucket and prefix, take every binding's lock for this object, in a fixed order, so the delete is - // serialized with an adoption through any of them. Writers hold one such lock, so no cycle forms. - const lockBindings = scope.anyBinding - ? [ - ...new Set([ - bindingId, - ...this.rowStorage.objectBindingIds(), - ...( - await client.query<{ row_binding_id: string }>( - `SELECT row_binding_id FROM adcp_reporting_row_bindings` - ) - ).rows.map(row => row.row_binding_id), - ]), - ].sort() - : [bindingId]; - for (const lockBinding of lockBindings) { - await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [ - objectLock(lockBinding, object.objectRef), - ]); - } - const referenced = await client.query( - `(SELECT 1 FROM adcp_reporting_row_chunks chunk - JOIN adcp_reporting_row_sets row_set ON row_set.row_set_id = chunk.row_set_id - WHERE ($8::boolean OR row_set.row_binding_id = $1) AND chunk.object_ref = $2 - AND chunk.native_version_ref = $3 - AND row_set.rows_state IN ('live', 'unavailable') - AND ($4::text IS NULL OR chunk.obligation_id <> $4) - AND ($5::text IS NULL OR chunk.row_set_id <> $5) - LIMIT 1) - UNION ALL - (SELECT 1 FROM adcp_reporting_row_write_intents intent - WHERE ($8::boolean OR intent.row_binding_id = $1) AND intent.state = 'open' - AND (intent.created_objects @> $6::jsonb OR intent.created_objects @> $7::jsonb) - LIMIT 1)`, - [ - bindingId, - object.objectRef, - object.nativeVersionRef, - scope.obligationId ?? null, - scope.rowSetId ?? null, - JSON.stringify([{ object_ref: object.objectRef, native_version_ref: object.nativeVersionRef }]), - // An older writer still running during a rolling upgrade records the pre-alignment keys. - JSON.stringify([{ objectKey: object.objectRef, nativeVersion: object.nativeVersionRef }]), - scope.anyBinding === true, - ] - ); - if (referenced.rows.length > 0) return false; - await this.rowStorage.deleteObjects(bindingId, [object], signal); - return true; - }); - if (removed) deleted += 1; + if (await this.isRowObjectReferenced(pool, bindingId, object, scope)) continue; + await this.rowStorage.deleteObjects(bindingId, [object], signal); + deleted += 1; } return deleted; } + /** + * Delete one object that no ledger record named when `collectOrphanedRowObjects` listed it, under the + * advisory lock `adcp-reporting-object:{binding}:{objectRef}` of every binding that can name it. This is + * the only delete that takes the lock: an object no record names can be adopted, unrecorded, by a + * writer's commit (a retry that finds an object an earlier attempt created but never recorded), and that + * commit takes the same lock before it proves the object and inserts its chunk rows. So either this + * delete wins and the commit's version-pinned `stat` fails it cleanly, or the commit wins and this + * check sees its chunk rows and skips. + * + * A writer takes the lock of its own binding only, so every binding (those this host configures and + * those the database has registered, which can share a bucket and prefix) is locked here, in a fixed + * order. Writers hold one such lock, so no cycle forms. A commit takes the account lock, then object + * locks; this delete never waits on the account lock while holding an object lock. + */ + private async deleteOrphanedObject( + bindingId: string, + object: { objectRef: string; nativeVersionRef: string }, + signal?: AbortSignal + ): Promise { + return this.transaction(async client => { + const lockBindings = [ + ...new Set([ + bindingId, + ...this.rowStorage.objectBindingIds(), + ...( + await client.query<{ row_binding_id: string }>(`SELECT row_binding_id FROM adcp_reporting_row_bindings`) + ).rows.map(row => row.row_binding_id), + ]), + ].sort(); + for (const lockBinding of lockBindings) { + await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [ + objectLock(lockBinding, object.objectRef), + ]); + } + if (await this.isRowObjectReferenced(client, bindingId, object, { anyBinding: true })) return false; + await this.rowStorage.deleteObjects(bindingId, [object], signal); + return true; + }); + } + + /** + * Whether a live or unavailable row set outside `scope`, or an open write intent, still names this + * exact object version. `anyBinding` widens the check to every binding (the orphan collector's). + */ + private async isRowObjectReferenced( + queryable: RowObjectQueryable, + bindingId: string, + object: { objectRef: string; nativeVersionRef: string }, + scope: { obligationId?: string; rowSetId?: string; anyBinding?: boolean } + ): Promise { + const referenced = await queryable.query( + `(SELECT 1 FROM adcp_reporting_row_chunks chunk + JOIN adcp_reporting_row_sets row_set ON row_set.row_set_id = chunk.row_set_id + WHERE ($8::boolean OR row_set.row_binding_id = $1) AND chunk.object_ref = $2 + AND chunk.native_version_ref = $3 + AND row_set.rows_state IN ('live', 'unavailable') + AND ($4::text IS NULL OR chunk.obligation_id <> $4) + AND ($5::text IS NULL OR chunk.row_set_id <> $5) + LIMIT 1) + UNION ALL + (SELECT 1 FROM adcp_reporting_row_write_intents intent + WHERE ($8::boolean OR intent.row_binding_id = $1) AND intent.state = 'open' + AND (intent.created_objects @> $6::jsonb OR intent.created_objects @> $7::jsonb) + LIMIT 1)`, + [ + bindingId, + object.objectRef, + object.nativeVersionRef, + scope.obligationId ?? null, + scope.rowSetId ?? null, + JSON.stringify([{ object_ref: object.objectRef, native_version_ref: object.nativeVersionRef }]), + // An older writer still running during a rolling upgrade records the pre-alignment keys. + JSON.stringify([{ objectKey: object.objectRef, nativeVersion: object.nativeVersionRef }]), + scope.anyBinding === true, + ] + ); + return referenced.rows.length > 0; + } + private async optionalTables(): Promise> { const result = await this.query<{ name: string }>( `SELECT table_name AS name FROM information_schema.tables @@ -3509,25 +3546,42 @@ ${managedDueArm} ) throw new ReportingRowStoreError('STATE_UNAVAILABLE', 'the shared row set is no longer live'); } } else { - const closed = await client.query( + // Closing the intent and inserting the chunk rows are one transaction, so a reader of the ledger sees + // an object named by this open intent or by committed chunk rows, never neither. + const closed = await client.query<{ created_objects: RecordedIntentObject[] }>( `DELETE FROM adcp_reporting_row_write_intents WHERE row_binding_id = $1 AND row_set_id = $2 AND content_sha256 = $3 - AND owner_token = $4 AND state = 'open'`, + AND owner_token = $4 AND state = 'open' + RETURNING created_objects`, [prepared.bindingId, prepared.input.rowSetId, prepared.input.binding.sha256, upload.ownerToken] ); if (closed.rowCount !== 1) { throw new ReportingRowStoreError('STATE_UNAVAILABLE', 'the row upload intent was taken over or swept'); } - } - if (!upload.sharedFrom) { - // A create-only write that found an object already there adopted it. Lock each adopted - // object (in key order) until this transaction ends, then prove it is still there: a - // delete holding the lock finished first and fails this commit, any later delete sees the - // chunk rows inserted below and skips (`deleteUnreferencedObjects`). + // A create-only write that found an object already there adopted it. Prove each adopted object is + // still there at its recorded version (a version-pinned `stat`, no download). + // + // An object this writer's own intent recorded as created (an earlier attempt at this row set that + // was taken over) needs no lock: until this transaction commits the open intent names it, so no + // delete skips the reference check, the intent sweep cannot claim an intent this commit just closed + // or one a takeover renewed, and the object's key is unique to this row set, so no other row set's + // retirement or compaction names it. + // + // An adopted object the intent does not record (an earlier attempt's PUT that landed before it was + // recorded, or one the writer created after a takeover) is named by no record. Only the orphan + // collector deletes such an object, and it deletes under the per-object lock: take it, in key + // order, until this transaction ends. A delete that held it finished first and the `stat` below + // fails this commit, and any later delete sees the chunk rows inserted below and skips. const adopted = upload.locators .filter(locator => !locator.created) .sort((left, right) => (left.objectRef < right.objectRef ? -1 : left.objectRef > right.objectRef ? 1 : 0)); + const recorded = new Set( + recordedIntentObjects(closed.rows[0]?.created_objects).map( + object => `${object.objectRef}\u0000${object.nativeVersionRef}` + ) + ); for (const locator of adopted) { + if (recorded.has(`${locator.objectRef}\u0000${locator.nativeVersionRef}`)) continue; await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [ objectLock(prepared.bindingId, locator.objectRef), ]); @@ -6391,7 +6445,10 @@ function accountLock(accountId: string): string { return `adcp-reporting-account:${accountId}`; } -/** Serializes deleting an external row object against a commit that adopts it. */ +/** + * Serializes the orphan collector's delete of an object no record names against a commit that adopts that + * object without having recorded it. No other delete or commit takes it. + */ function objectLock(bindingId: string, objectRef: string): string { return `adcp-reporting-object:${bindingId}:${objectRef}`; } @@ -6700,6 +6757,34 @@ function withRows[] }>(value: T, rows: return Array.isArray(value.rows) ? value : { ...value, rows: clone(rows) }; } +/** One entry of a write intent's `created_objects`; an older writer records `objectKey` and `nativeVersion`. */ +interface RecordedIntentObject { + object_ref?: string; + native_version_ref?: string; + objectKey?: string; + nativeVersion?: string; +} + +/** The objects a write intent recorded, reading both the current keys and an older writer's pre-alignment keys. */ +function recordedIntentObjects( + entries: readonly RecordedIntentObject[] | undefined +): { objectRef: string; nativeVersionRef: string }[] { + return (entries ?? []) + .map(entry => ({ + objectRef: entry.object_ref ?? entry.objectKey, + nativeVersionRef: entry.native_version_ref ?? entry.nativeVersion, + })) + .filter( + (object): object is { objectRef: string; nativeVersionRef: string } => + typeof object.objectRef === 'string' && typeof object.nativeVersionRef === 'string' + ); +} + +/** The part of a pool or transaction client the reference check needs. */ +interface RowObjectQueryable { + query: (text: string, values?: unknown[]) => Promise<{ rows: unknown[] }>; +} + /** Locations for an object-kind row set: freshly uploaded under an intent, or shared from identical rows. */ type RowSetUploadV1 = | { locators: ReportingRowChunkLocatorV1[]; ownerToken: string; sharedFrom?: undefined } diff --git a/packages/reporting-storage/src/row-storage.ts b/packages/reporting-storage/src/row-storage.ts index 84e8f6f5d..698c5ac0a 100644 --- a/packages/reporting-storage/src/row-storage.ts +++ b/packages/reporting-storage/src/row-storage.ts @@ -374,9 +374,9 @@ export class ReportingRowStorageV1 { * skipped: it never takes commits, so nothing can adopt through it, and every delete is * reference-checked. * - * Since the reference-checked, per-object-locked delete this is no longer the only guard: every - * delete path now skips an object another row set still references. It is kept as defense in - * depth for compaction, the one path that deletes content while the obligation stays live. + * Since the reference-checked delete this is no longer the only guard: every delete path now + * skips an object another row set still references. It is kept as defense in depth for + * compaction, the one path that deletes content while the obligation stays live. */ assertCompactionSafeKeyTemplates(): void { const unsafe = Object.entries(this.bindings) @@ -1190,14 +1190,16 @@ export class ReportingRowStorageV1 { /** * Prove each object is still present at its recorded version with the recorded size. - * A commit that adopted an object another row set created calls this under the object's - * lock, so a concurrent delete either finished first (this throws, the commit fails and - * the writer re-uploads) or sees the commit's chunk reference and skips the delete. + * A commit that adopted an existing object calls this inside its ledger transaction. An object + * the writer's own intent recorded is named by that open intent until the commit, so no delete + * touches it. An object no record names is also locked first (`writePreparedRowSet`), so the one + * delete that can reach it, the orphan collector's, either finished first (this throws, the + * commit fails and the writer re-uploads) or sees the commit's chunk reference and skips. * * Native versions are write-unique (S3 VersionId, Azure versionId, GCS generation, filesystem * `ino:birthtime:mtime:size`), so a version-pinned metadata check, "this exact version exists at the * recorded size", proves the object unchanged without downloading it while the commit holds its - * locks. Bytes were verified when the object was uploaded or adopted. A custom provider without + * transaction. Bytes were verified when the object was uploaded or adopted. A custom provider without * `stat` falls back to reading the whole object and comparing its physical digest. */ async assertObjectsPresent( diff --git a/packages/reporting-storage/test/reporting-row-shared-object-pg.test.js b/packages/reporting-storage/test/reporting-row-shared-object-pg.test.js index 9da0a3c81..c5d843b94 100644 --- a/packages/reporting-storage/test/reporting-row-shared-object-pg.test.js +++ b/packages/reporting-storage/test/reporting-row-shared-object-pg.test.js @@ -1,11 +1,17 @@ /** - * PostgreSQL integration tests for external row objects shared between row sets. + * PostgreSQL integration tests for external row objects shared between row sets, and for the absence of a + * per-object advisory lock on the delete and adopt paths (adcp-client#3281). + * * Object keys are unique per row set (every writable binding's key template includes `{revision_key}`), * but header-only revisions share chunks within an obligation (covered by the compaction tests below * and the object-storage suite) and chunk rows written under an older template can name one object * from several row sets. These tests re-point chunk rows at a shared object to model the latter. Every delete path must leave an object another row set still references, - * and must not leak it once the last reference is gone; a commit that adopts an object another - * writer created (a retry of the same row set) must prove it is still there under the object lock. + * and must not leak it once the last reference is gone. + * + * The delete paths (retirement, compaction, the intent sweep) take no lock and hold no connection across the + * provider delete: safety comes from the reference check, the row-state and intent-state fences and unique keys. + * Two paths keep the lock because nothing but the lock orders them: a commit that adopts an object its own + * intent did not record, and the orphan collector's delete. The tests below pin each side of that contract. * * REPORTING_LEDGER_PG_URL=postgres://localhost/test npm test --workspace=@adcp/reporting-storage -- test/reporting-row-shared-object-pg.test.js */ @@ -14,7 +20,7 @@ const { createHash } = require('node:crypto'); const { existsSync, mkdtempSync, readFileSync, rmSync, statSync, utimesSync, writeFileSync } = require('node:fs'); const { tmpdir } = require('node:os'); const path = require('node:path'); -const { after, before, describe, test } = require('node:test'); +const { after, afterEach, before, describe, test } = require('node:test'); const { createRowStorageFixture } = require('./helpers/reporting-row-storage-fixture.js'); @@ -36,6 +42,9 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR // Distinct per test, so one test never shares objects with another test's obligations. let current = rows; let afterAdoptedRead; + let beforeDelete; + // Pooled connections the store held checked out whenever the provider was asked to delete. + let heldAtDelete = []; let providerCalls = { get: 0, stat: 0 }; let binding; let provider; @@ -80,8 +89,8 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR * digested with its own ID, so revisions never collide; an adjustment is digested over its rows * alone. */ - async function officialObligation(suffix) { - const fixture = await createRowStorageFixture({ store, suffix }); + async function officialObligation(suffix, accountId) { + const fixture = await createRowStorageFixture({ store, suffix, ...(accountId ? { accountId } : {}) }); const official = fixture.revision(`rrev_${suffix}`, 1, rows.slice(0, 5), { finality: 'official', kind: 'official', @@ -116,11 +125,11 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR } /** Two obligations, each with an adjustment of identical rows, whose chunk rows name one object (as rows written under an older template could). */ - async function sharedPair() { + async function sharedPair({ separateAccounts = false } = {}) { sequence += 1; current = rows.map(row => ({ ...row, value: row.value + sequence * 100 })); - const first = await officialObligation(`shared_a${sequence}`); - const second = await officialObligation(`shared_b${sequence}`); + const first = await officialObligation(`shared_a${sequence}`, separateAccounts ? `acct_a${sequence}` : undefined); + const second = await officialObligation(`shared_b${sequence}`, separateAccounts ? `acct_b${sequence}` : undefined); await store.commitAdjustment(adjustmentFor(first), first.fixture.lease); await store.commitAdjustment(adjustmentFor(second), second.fixture.lease); const [chunkA] = await chunksOf(first.adjustmentId); @@ -146,6 +155,7 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR 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 base = ledger.createFilesystemReportingRowObjectProviderV1({ roots: { rows: root } }); baseProvider = base; // After an adopting writer reads the existing object, run a hook (once): the window between @@ -156,6 +166,14 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR providerCalls.stat += 1; return base.stat(input, context); }, + async delete(input, context) { + heldAtDelete.push(pool.totalCount - pool.idleCount); + const hook = beforeDelete; + // Cleared first: a delete the hook itself provokes must not re-enter it. `keep` re-arms it for the next delete. + beforeDelete = undefined; + if (hook && (await hook(input)) === 'keep') beforeDelete = hook; + return base.delete(input, context); + }, async get(input, context) { providerCalls.get += 1; const bytes = await base.get(input, context); @@ -173,11 +191,28 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR }; store = new ledger.PostgresReportingLedgerStore(pool, { acknowledgeIsolatedDatabase: true, - rowStorage: { bindings: { objects: binding }, providers: { filesystem: provider } }, + rowStorage: { + bindings: { + objects: binding, + // An older definition kept for the rows it wrote: no {revision_key}, never selected for new writes. + legacy: { + ...binding, + readOnly: true, + keyTemplate: '{prefix}/{namespace_key}/{account_key}/{content_sha256}/{chunk_index}', + }, + }, + providers: { filesystem: provider }, + }, }); await store.readyRowStorage(); }); + afterEach(() => { + // A failed test must not leave a hook armed for the next one. + afterAdoptedRead = undefined; + beforeDelete = undefined; + }); + after(async () => { if (pool) await pool.end(); if (bootstrapPool) { @@ -239,30 +274,402 @@ describe('reporting shared row objects', { skip: !DATABASE_URL && 'PostgreSQL UR assert.fail('no session ever waited on the object lock'); } - test('a delete waits for the object lock a committing adopter holds', async () => { + const objectLockKey = objectRef => `adcp-reporting-object:objects:${objectRef}`; + const timedOut = Symbol('timed out'); + const within = (promise, milliseconds = 5_000) => Promise.race([promise, sleep(milliseconds).then(() => timedOut)]); + + /** Hold the per-object advisory lock on a session outside the store's pool, until `release` is called. */ + async function holdObjectLock(objectRef) { + const holder = await bootstrapPool.connect(); + let released = false; + await holder.query('BEGIN'); + await holder.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [objectLockKey(objectRef)]); + return { + async release() { + if (released) return; + released = true; + await holder.query('ROLLBACK').catch(() => undefined); + holder.release(); + }, + }; + } + + /** Sessions waiting for exactly this advisory lock key, so another file's waiters cannot satisfy it. */ + const waitersOn = lockText => + count( + `SELECT count(*) AS n FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted + AND ((classid::bigint << 32) | objid::bigint) = hashtextextended($1, 0)`, + [lockText] + ); + + async function untilWaiting(lockText) { + for (let attempt = 0; attempt < 400; attempt += 1) { + if ((await waitersOn(lockText)) > 0) return; + await sleep(25); + } + assert.fail(`no session ever waited on ${lockText}`); + } + + test('retirement takes no object lock and holds no connection while the provider deletes', async () => { const { first, second, chunkA } = await sharedPair(); - // Both periods expire, so only the lock (not a surviving reference) can protect the object. await expire(first); await expire(second); - const lock = `adcp-reporting-object:objects:${chunkA.object_ref}`; - const client = await pool.connect(); - let settled = false; + const held = await holdObjectLock(chunkA.object_ref); + const retiring = retire(); try { - await client.query('BEGIN'); - await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [lock]); - const retiring = retire().then(result => { - settled = true; + heldAtDelete = []; + const result = await within(retiring); + assert.notEqual(result, timedOut, 'retirement does not wait on the object lock'); + assert.equal(result.retired, 2); + } finally { + await held.release(); + } + await retiring; + assert.equal(existsSync(fileOf(chunkA)), false); + assert.ok(heldAtDelete.length > 0, 'the provider was asked to delete'); + assert.ok( + heldAtDelete.every(held => held === 0), + 'no pooled connection is checked out across a delete' + ); + }); + + test('the intent sweep takes no object lock and holds no connection while the provider deletes', async () => { + const { target, objectRef, nativeVersionRef, file } = await preUploaded('sweeplock'); + await 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', $1, 'radj_sweep_no_lock', $2, 'open', 'owner', $3::jsonb, + clock_timestamp() - INTERVAL '1 minute')`, + [ + accountId(target), + 'a'.repeat(64), + JSON.stringify([{ object_ref: objectRef, native_version_ref: nativeVersionRef }]), + ] + ); + const held = await holdObjectLock(objectRef); + heldAtDelete = []; + const sweeping = store.sweepRowWriteIntents(); + try { + const result = await within(sweeping); + assert.notEqual(result, timedOut, 'the sweep does not wait on the object lock'); + assert.equal(result.objectsDeleted, 1); + } finally { + await held.release(); + } + await sweeping; + assert.equal(existsSync(file), false); + assert.deepEqual(heldAtDelete, [0]); + }); + + /** A leased obligation and a fresh adjustment for it, with no object uploaded yet. */ + async function freshTarget(suffix) { + sequence += 1; + current = rows.map(row => ({ ...row, value: row.value + sequence * 100 })); + return officialObligation(`${suffix}${sequence}`); + } + + /** + * An attempt that uploads and records its objects, then fails before its ledger commit lands: the intent + * stays open and names the created objects, as a writer that crashed or lost its connection leaves it. + */ + async function abandonedAttempt(target) { + const original = store.rowStorage.writeInTransaction; + store.rowStorage.writeInTransaction = async () => { + throw new Error('connection reset'); + }; + try { + await assert.rejects(() => store.commitAdjustment(adjustmentFor(target), target.fixture.lease)); + } finally { + store.rowStorage.writeInTransaction = original; + } + const intent = ( + await pool.query(`SELECT state, created_objects FROM adcp_reporting_row_write_intents WHERE row_set_id = $1`, [ + target.adjustmentId, + ]) + ).rows; + assert.equal(intent.length, 1, 'the failed attempt leaves its intent'); + assert.equal(intent[0].state, 'open'); + assert.equal(intent[0].created_objects.length, 1, 'and the intent names the object it created'); + return intent[0].created_objects[0]; + } + + const expireIntent = target => + pool.query( + `UPDATE adcp_reporting_row_write_intents SET expires_at = clock_timestamp() - INTERVAL '1 minute' + WHERE row_set_id = $1`, + [target.adjustmentId] + ); + + /** Run `hook` after the real upload of the next commit returns, before its ledger transaction opens. */ + function afterUpload(hook) { + const original = store.rowStorage.upload; + store.rowStorage.upload = async function (...args) { + const locators = await original.apply(this, args); + await hook(locators); + return locators; + }; + return () => { + store.rowStorage.upload = original; + }; + } + + test('a retry adopts the object its own earlier attempt recorded without taking the object lock', async () => { + const target = await freshTarget('selfretry'); + const recorded = await abandonedAttempt(target); + await expireIntent(target); + const held = await holdObjectLock(recorded.object_ref); + providerCalls = { get: 0, stat: 0 }; + const committing = store.commitAdjustment(adjustmentFor(target), target.fixture.lease); + try { + const result = await within(committing); + assert.notEqual(result, timedOut, 'adopting an object the writer recorded does not wait on the object lock'); + assert.equal(result.inserted, true); + } finally { + await held.release(); + } + await committing; + assert.equal(providerCalls.stat, 1, 'the adopted object is still proven with a version-pinned stat'); + assert.deepEqual( + (await chunksOf(target.adjustmentId)).map(chunk => chunk.object_ref), + [recorded.object_ref], + 'the chunk row names the adopted object' + ); + assert.equal( + await count(`SELECT count(*) AS n FROM adcp_reporting_row_write_intents WHERE row_set_id = $1`, [ + target.adjustmentId, + ]), + 0 + ); + assert.deepEqual(await adjustmentRows(target), current); + }); + + test('a retry whose intent was written by an older writer (pre-alignment keys) also adopts without the lock', async () => { + const target = await freshTarget('legacykeys'); + const recorded = await abandonedAttempt(target); + await pool.query(`UPDATE adcp_reporting_row_write_intents SET created_objects = $2::jsonb WHERE row_set_id = $1`, [ + target.adjustmentId, + JSON.stringify([{ objectKey: recorded.object_ref, nativeVersion: recorded.native_version_ref }]), + ]); + await expireIntent(target); + const held = await holdObjectLock(recorded.object_ref); + const committing = store.commitAdjustment(adjustmentFor(target), target.fixture.lease); + try { + const result = await within(committing); + assert.notEqual(result, timedOut, 'the pre-alignment record is read like the current one'); + assert.equal(result.inserted, true); + } finally { + await held.release(); + } + await committing; + assert.deepEqual(await adjustmentRows(target), current); + }); + + test('a retry takes the object lock when its intent names the key at a different native version', async () => { + const target = await freshTarget('otherversion'); + const recorded = await abandonedAttempt(target); + await pool.query(`UPDATE adcp_reporting_row_write_intents SET created_objects = $2::jsonb WHERE row_set_id = $1`, [ + target.adjustmentId, + JSON.stringify([{ object_ref: recorded.object_ref, native_version_ref: 'some-other-version' }]), + ]); + await expireIntent(target); + const held = await holdObjectLock(recorded.object_ref); + let committed = false; + const committing = store.commitAdjustment(adjustmentFor(target), target.fixture.lease).then(result => { + committed = true; + return result; + }); + try { + await untilWaiting(objectLockKey(recorded.object_ref)); + assert.equal(committed, false, 'a record of another version does not vouch for the adopted one'); + } finally { + await held.release(); + } + assert.equal((await committing).inserted, true); + }); + + test('a retry still takes the object lock for an adopted object its intent does not name', async () => { + const target = await freshTarget('unrecorded'); + const recorded = await abandonedAttempt(target); + // The earlier attempt's PUT landed before the intent recorded it (or its record failed after a takeover). + await pool.query( + `UPDATE adcp_reporting_row_write_intents SET created_objects = '[]'::jsonb WHERE row_set_id = $1`, + [target.adjustmentId] + ); + await expireIntent(target); + const held = await holdObjectLock(recorded.object_ref); + let committed = false; + try { + const committing = store.commitAdjustment(adjustmentFor(target), target.fixture.lease).then(result => { + committed = true; return result; }); - await lockWaiter(client); - assert.equal(settled, false, 'retirement is blocked on the object lock'); - assert.ok(existsSync(fileOf(chunkA)), 'the object survives while the lock is held'); - await client.query('COMMIT'); - assert.equal((await retiring).retired, 2); + await untilWaiting(objectLockKey(recorded.object_ref)); + assert.equal(committed, false, 'an unrecorded adoption waits for the object lock'); + await held.release(); + assert.equal((await committing).inserted, true); } finally { - client.release(); + await held.release(); + } + assert.deepEqual(await adjustmentRows(target), current); + }); + + test('a writer whose intent expired before its commit cannot commit the objects the sweep deleted', async () => { + const target = await freshTarget('pastlease'); + let sweep; + // The intent lapses while the writer is between its upload and its commit (a stalled or partitioned writer). + const restore = afterUpload(async () => { + await expireIntent(target); + sweep = await store.sweepRowWriteIntents(); + }); + try { + await assert.rejects( + () => store.commitAdjustment(adjustmentFor(target), target.fixture.lease), + isStateUnavailable + ); + } finally { + restore(); + } + assert.equal(sweep.swept, 1); + assert.equal(sweep.objectsDeleted, 1, 'the sweep deleted what the intent created'); + for (const table of ['adcp_reporting_row_chunks', 'adcp_reporting_row_sets', 'adcp_reporting_row_write_intents']) { + assert.equal( + await count(`SELECT count(*) AS n FROM ${table} WHERE row_set_id = $1`, [target.adjustmentId]), + 0, + table + ); + } + assert.equal((await store.listAdjustments(target.fixture.obligation.reporting_obligation_id)).length, 0); + // Nothing references a deleted object, and the writer's next attempt re-creates it. + await store.commitAdjustment(adjustmentFor(target), target.fixture.lease); + assert.deepEqual(await adjustmentRows(target), current); + }); + + test('a takeover renews the intent, so a sweep cannot claim the objects its retry is about to commit', async () => { + const target = await freshTarget('takeover'); + const recorded = await abandonedAttempt(target); + await expireIntent(target); + let sweep; + const restore = afterUpload(async () => { + sweep = await store.sweepRowWriteIntents(); + }); + try { + assert.equal((await store.commitAdjustment(adjustmentFor(target), target.fixture.lease)).inserted, true); + } finally { + restore(); + } + assert.equal(sweep.swept, 0, 'the retry took the expired intent over, so it is no longer expired'); + assert.equal(sweep.objectsDeleted, 0); + assert.deepEqual( + (await chunksOf(target.adjustmentId)).map(chunk => chunk.object_ref), + [recorded.object_ref] + ); + assert.deepEqual(await adjustmentRows(target), current); + }); + + test('a retry cannot take over an intent the sweep has claimed, and commits nothing', async () => { + const target = await freshTarget('sweepfirst'); + await abandonedAttempt(target); + await expireIntent(target); + // The sweep has claimed the intent (`sweeping`) and is about to delete: a retry must not take it over. + let retried; + beforeDelete = async () => { + retried = await store.commitAdjustment(adjustmentFor(target), target.fixture.lease).then( + () => 'committed', + error => error + ); + }; + const swept = await store.sweepRowWriteIntents(); + assert.equal(swept.objectsDeleted, 1); + assert.ok(isStateUnavailable(retried), 'a retry cannot take over an intent that is being swept'); + assert.equal( + await count(`SELECT count(*) AS n FROM adcp_reporting_row_sets WHERE row_set_id = $1`, [target.adjustmentId]), + 0 + ); + }); + + test('an orphan delete in flight blocks an adopter of the unrecorded object, which then fails cleanly', async () => { + const { target, objectRef, file } = await preUploaded('collector'); + const later = new Date(Date.now() + 8 * 24 * 3_600_000); + let committing; + // The collector holds the object lock and has passed its reference check; an adopter arrives now. + beforeDelete = async input => { + if (input.objectRef !== objectRef) return 'keep'; + committing = store.commitAdjustment(adjustmentFor(target), target.fixture.lease).then( + () => 'committed', + error => error + ); + await untilWaiting(objectLockKey(objectRef)); + }; + const result = await store.collectOrphanedRowObjects({ + now: later, + acknowledgeExclusiveNamespace: true, + bindings: ['objects'], + }); + assert.ok(result.deleted >= 1); + assert.equal(existsSync(file), false, 'the collector deleted the unrecorded object'); + const outcome = await committing; + assert.ok( + isStateUnavailable(outcome), + `the adopter fails cleanly instead of committing a missing object: ${outcome}` + ); + assert.equal( + await count(`SELECT count(*) AS n FROM adcp_reporting_row_chunks WHERE row_set_id = $1`, [target.adjustmentId]), + 0 + ); + }); + + test('retirements of legacy row sets in a readOnly binding sharing one object always free it', async () => { + // Old-template rows live only in a readOnly binding, which never takes a commit, so nothing adopts through + // it. Each retirement counts only live row sets outside itself, so however two interleave, the last frees it. + for (let round = 0; round < 4; round += 1) { + const { first, second, chunkA } = await sharedPair({ separateAccounts: true }); + await pool.query( + `UPDATE adcp_reporting_row_sets SET row_binding_id = 'legacy', row_location_version = row_location_version + 1 + WHERE row_set_id = ANY($1)`, + [[first.adjustmentId, second.adjustmentId]] + ); + await expire(first); + await expire(second); + // Hold both retirements at the shared object's reference check until both have committed `pruning` (phase + // 1 precedes the check): the interleaving in which a check that counted `pruning` row sets would make both skip. + const original = store.isRowObjectReferenced; + let arrived = 0; + let open; + const gate = new Promise(resolve => { + open = resolve; + }); + const timer = setTimeout(() => open(), 3_000); + store.isRowObjectReferenced = async function (queryable, bindingId, object, scope) { + if (object.objectRef === chunkA.object_ref) { + arrived += 1; + if (arrived === 2) open(); + await gate; + } + return original.call(this, queryable, bindingId, object, scope); + }; + let a; + let b; + try { + [a, b] = await Promise.all([ + store.retireExpiredPeriods({ statusRetentionDays: 30, account_id: accountId(first) }), + store.retireExpiredPeriods({ statusRetentionDays: 30, account_id: accountId(second) }), + ]); + } finally { + clearTimeout(timer); + store.isRowObjectReferenced = original; + } + assert.equal(arrived, 2, 'both retirements reached the shared object together'); + assert.deepEqual([a.retired, b.retired], [1, 1]); + assert.equal(a.failed.length + b.failed.length, 0); + assert.equal(existsSync(fileOf(chunkA)), false, `round ${round}: the last retirement deletes the object`); + assert.equal( + await count(`SELECT count(*) AS n FROM adcp_reporting_row_sets WHERE row_set_id = ANY($1)`, [ + [first.adjustmentId, second.adjustmentId], + ]), + 0 + ); } - assert.equal(existsSync(fileOf(chunkA)), false, 'once the lock is released the unreferenced object is deleted'); }); test('an adopting commit takes the object lock before it proves the object', async () => { @@ -706,4 +1113,97 @@ describe('reporting compaction with a shared object', { skip: !DATABASE_URL && ' assert.deepEqual(await states(), ['pruned', 'pruned']); assert.equal(existsSync(file), false, 'the last referencing revision deletes the object'); }); + + const compactAll = () => + store.compactSupersededSnapshots({ statusRetentionDays: 30, automatedRecoveryWindowSeconds: 300 }); + const rowSetStates = async ids => + ( + await pool.query(`SELECT row_set_id, rows_state FROM adcp_reporting_row_sets WHERE row_set_id = ANY($1)`, [ids]) + ).rows.reduce((byId, row) => ({ ...byId, [row.row_set_id]: row.rows_state }), {}); + const chunkRefs = async rowSetId => + ( + await pool.query(`SELECT object_ref FROM adcp_reporting_row_chunks WHERE row_set_id = $1 ORDER BY chunk_index`, [ + rowSetId, + ]) + ).rows.map(row => row.object_ref); + + test('compaction takes no object lock and holds no connection while the provider deletes', async () => { + const a = await chain('cmp_nolock'); + const [objectRef] = await chunkRefs(a.ids[0]); + const holder = await bootstrapPool.connect(); + const held = []; + const original = store.rowStorage.deleteObjects; + store.rowStorage.deleteObjects = function (...args) { + held.push(pool.totalCount - pool.idleCount); + return original.apply(this, args); + }; + try { + await holder.query('BEGIN'); + await holder.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [ + `adcp-reporting-object:objects:${objectRef}`, + ]); + const result = await Promise.race([compactAll(), sleep(5_000).then(() => 'blocked')]); + assert.notEqual(result, 'blocked', 'compaction does not wait on the object lock'); + assert.ok(result.compacted >= 1); + } finally { + await holder.query('ROLLBACK').catch(() => undefined); + holder.release(); + store.rowStorage.deleteObjects = original; + } + assert.equal(existsSync(path.join(root, objectRef)), false); + assert.ok(held.length > 0 && held.every(connections => connections === 0), `held: ${held}`); + }); + + test('a header-only revision whose predecessor is compacted before its commit fails cleanly and then re-uploads', async () => { + const a = await chain('cmp_hdr'); + const [r1, , r3] = a.ids; + const obligationId = a.fixture.obligation.reporting_obligation_id; + const [sharedObject] = await chunkRefs(r1); + // The writer holds a live lease while it prepares; compaction runs only once the lease has lapsed. + const setLease = (state, interval) => + pool.query( + `UPDATE adcp_reporting_obligations SET state = $2, lease_expires_at = clock_timestamp() + $3::interval + WHERE obligation_id = $1`, + [obligationId, state, interval] + ); + await setLease('pending', '10 minutes'); + const pulse = a.fixture.revision('rrev_cmp_hdr_4', 4, rowsFor(0), { supersedes_reporting_revision_id: r3 }); + + const original = store.rowStorage.findShareable; + let compacted = 0; + store.rowStorage.findShareable = async function (...args) { + const found = await original.apply(this, args); + if (found && compacted === 0) { + // The writer has decided to share r1's chunks. Its lease lapses, compaction deletes r1 and its + // object, and the lease is held again by the time the commit transaction runs. + await setLease('terminal', '-1 second'); + compacted = (await compactAll()).compacted; + await setLease('pending', '10 minutes'); + } + return found; + }; + try { + await assert.rejects( + () => store.commitRevision(pulse, a.fixture.lease), + error => + ledger.isReportingRowStoreError(error) && + error.code === 'STATE_UNAVAILABLE' && + /no longer live/.test(error.message) + ); + } finally { + store.rowStorage.findShareable = original; + } + assert.ok(compacted >= 1, 'compaction ran between the sharing decision and the commit'); + assert.equal((await rowSetStates([r1, 'rrev_cmp_hdr_4']))[r1], 'pruned'); + assert.equal(existsSync(path.join(root, sharedObject)), false, 'the predecessor object was deleted'); + assert.equal((await rowSetStates(['rrev_cmp_hdr_4']))['rrev_cmp_hdr_4'], undefined, 'no row set was committed'); + assert.deepEqual(await chunkRefs('rrev_cmp_hdr_4'), [], 'no chunk row names the deleted object'); + + // A retry no longer finds a live predecessor to share, so it uploads its own object and commits. + assert.equal((await store.commitRevision(pulse, a.fixture.lease)).inserted, true); + const [ownObject] = await chunkRefs('rrev_cmp_hdr_4'); + assert.ok(existsSync(path.join(root, ownObject)), 'the retry uploaded its own object'); + assert.equal(existsSync(path.join(root, sharedObject)), false, 'the deleted predecessor object stays deleted'); + assert.deepEqual((await store.getRevision('rrev_cmp_hdr_4', a.fixture.request.account.account_id)).rows, rows); + }); });