From 4cab4e13b509991f43c5dacfb114eb1914a159ce Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 3 Nov 2026 03:10:33 +0000 Subject: [PATCH] perf(reporting-storage): drop the per-object advisory lock where nothing pairs on it Mirrors the cross-SDK contract from adcp-client#3606. Retirement, the intent sweep and a commit that adopts an object its own intent recorded take no object lock and hold no transaction or pooled connection across the provider delete. The lock stays in the two places that pair with each other: the orphan collector delete, which locks every binding key for the object, and a commit that adopts an object its intent did not record. A commit closes its intent with RETURNING created_objects to know which adopted objects it recorded. The lock key text is unchanged, so a Python deleter and a JavaScript adopter agree in one database. Closes #1492. Co-Authored-By: Claude Sonnet 5.5 --- docs/reporting-release-notes.md | 12 + docs/reporting-row-storage.md | 32 +- .../ledger/orphan_collector.py | 17 +- .../ledger/row_retention.py | 11 +- .../ledger/row_storage.py | 261 ++++++++++------ .../ledger/row_write.py | 18 +- ...t_reporting_row_object_lock_contract_pg.py | 293 ++++++++++++++++++ 7 files changed, 535 insertions(+), 109 deletions(-) create mode 100644 tests/conformance/reporting/test_reporting_row_object_lock_contract_pg.py diff --git a/docs/reporting-release-notes.md b/docs/reporting-release-notes.md index 94f82b3ef4..088be43984 100644 --- a/docs/reporting-release-notes.md +++ b/docs/reporting-release-notes.md @@ -53,6 +53,18 @@ Reads handle both forms. `migrate_inline_rows` moves legacy rows with verificati rows do not reproduce its binding, and `retire_legacy_inline_rows` then renames the legacy table so older binaries fail closed. See [row storage](reporting-row-storage.md). +## The per-object lock is kept only where it pairs + +Retirement, the intent sweep and a commit that adopts an object its own intent recorded no longer take the per-object +advisory lock, and no longer hold a transaction or pooled connection across the provider delete. The lock stays in the +two places that pair with each other: the orphan collector's delete, which locks every binding's key for the object, and +a commit that adopts an object its intent did not record. The lock text is unchanged, so a Python deleter and a +JavaScript adopter still agree, and rolling upgrades between a version that locks everywhere and this one are safe (every +pair that needs exclusion keeps the lock on both sides). A commit now closes its intent with `RETURNING created_objects` +to know which adopted objects it recorded. No schema or wire change. See +[the object lock contract](reporting-row-storage.md#the-object-lock-contract) +([#1492](https://github.com/adcontextprotocol/adcp-client-python/issues/1492), adcp-client#3606). + ## Opt-in orphaned-object collector, fenced against clones `PgReportingLedgerStore(orphan_collection=ReportingOrphanCollectionOptionsV1(...))` and diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index 76c30bb088..5095fd3023 100644 --- a/docs/reporting-row-storage.md +++ b/docs/reporting-row-storage.md @@ -327,6 +327,36 @@ stays as a small marker that makes reads answer as expired; the records are alre whose horizon excludes periods past retention. Versioned buckets, soft delete and retention locks are allowed as backstops, but a `pruned` period is not "erased" while they keep the old bytes. +## The object lock contract + +The per-object advisory lock is kept only where nothing else orders the two sides, in the contract the JavaScript SDK +shares ([adcp-client#3606](https://github.com/adcontextprotocol/adcp-client/pull/3606)). The lock text is +`adcp-reporting-object:` + binding ID + `:` + object ref, passed to `pg_advisory_xact_lock(hashtextextended(text, 0))`, +so a Python deleter and a JavaScript adopter agree in one database. + +1. **No lock:** retirement, the intent sweep and a commit that adopts an object its own intent recorded. Per object, one + statement checks that no `live` or `unavailable` row set outside the operation names the same binding, `object_ref` + and `native_version_ref`, and no `open` intent in the binding lists it in `created_objects` (both the current and the + pre-alignment `objectKey` / `nativeVersion` spellings are read); then the exact version is deleted. No transaction or + pooled connection is held across the provider delete. +2. **A commit** takes the account lock and closes its intent with `DELETE ... WHERE owner_token = ... AND state = + 'open' RETURNING created_objects` (no row fails `STATE_UNAVAILABLE`). For each adopted object whose + `(object_ref, native_version_ref)` is **not** in the returned list it takes its binding's object lock, in object-ref + order. It then proves every adopted object with a version-pinned `stat` (a full read when the provider has none) + before inserting the chunk rows. +3. **The orphan collector** locks every binding's key for the object (configured plus registered, sorted by binding + ID), repeats the reference check across all bindings under those locks, then deletes. It is the only locked delete. +4. Lock order: the account lock, then object locks. Native versions must be write-unique (the conformance suite's + `write-unique-version`). + +Why this is safe without the lock: the account lock orders commits, retirement phase 1, intent takeover and the sweep +claim; `pruning` is committed under it before any delete, and `pruning` row sets neither count as references nor can be +shared; an intent moves `open -> sweeping` against the commit's `DELETE ... WHERE state = 'open'`, in the same transaction +as the chunk insert, so a reader sees the open intent or the chunk rows, never neither; deletes name an exact native +version; and every writable key includes `{revision_key}`, so one row set alone can name it. **Precondition (documented, +not enforced):** two writable bindings must not share one bucket and prefix, because the reference check is per binding +outside the collector. + ## Orphaned objects: the opt-in listing collector The intent sweep deletes only what an intent lists, so it cannot see three cases: an upload whose `PUT` succeeded before @@ -363,7 +393,7 @@ these hold: old or new JSON keys) naming the key. Two bindings can share a bucket and prefix (a retired `read_only` binding kept next to its successor), so the check is not scoped to the collecting binding. 4. Under the per-object advisory lock of **every** binding, taken in a fixed order so the delete is serialized with an - adoption through any of them, no `live` or `unavailable` row set references that exact version and no open intent + adoption through any of them (see [the object lock contract](#the-object-lock-contract)), no `live` or `unavailable` row set references that exact version and no open intent created it. The provider then deletes that exact version. `read_only` bindings are skipped unless `include_read_only`. `dry_run` deletes nothing and reports up to 1000 diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/orphan_collector.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/orphan_collector.py index 3f16d92f2c..060fc645a4 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/orphan_collector.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/orphan_collector.py @@ -15,7 +15,8 @@ * 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 of **every** binding, no ``live`` or ``unavailable`` row set +* 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 then deletes that exact version. @@ -541,15 +542,13 @@ async def gate(binding_id: str, mode: Literal["claim", "check"]) -> _Gate: since_fence_check = 0 since_fence_check += 1 try: - # The per-object lock and exact-version reference check every other - # delete makes, widened to every binding. - removed = await storage.delete_unreferenced( - store._pool, - summary.id, - [(entry.object_ref, entry.native_version_ref)], - any_binding=True, + # The only delete that takes the per-object lock: no record names this + # object, so a commit may be adopting it unrecorded. The reference check + # across every binding runs under the lock of every binding. + removed = await storage.delete_orphaned_object( + store._pool, summary.id, entry.object_ref, entry.native_version_ref ) - if removed > 0: + if removed: outcome.deleted += 1 else: outcome.referenced += 1 diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_retention.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_retention.py index 1834aa4b29..bb1690547f 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_retention.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_retention.py @@ -378,8 +378,9 @@ async def delete_period_objects( An object that a live chunk of another obligation still references is kept: the default key template cannot collide across obligations, but a custom template can, and a - delete must never take bytes another revision still needs. See - ``ReportingRowStorageV1.delete_unreferenced`` for the locking contract. + delete must never take bytes another revision still needs. No object lock is taken: see the + contract in ``ReportingRowStorageV1`` (``# -- object lifetime``) and + ``ReportingRowStorageV1.delete_unreferenced``. """ storage = store._row_storage assert storage is not None @@ -397,8 +398,10 @@ async def delete_period_objects( for binding_id, object_ref, native_version_ref in objects: by_binding.setdefault(binding_id, []).append((object_ref, native_version_ref)) for binding_id, pairs in by_binding.items(): - # Objects a chunk of any other row set still references are kept, checked - # atomically with the delete so a concurrent commit cannot adopt a doomed object. + # Objects a chunk of any other row set still references are kept. The check needs no lock + # against a committing writer: the period's row sets are already ``pruning`` (committed + # under the account lock in phase 1), which orders this delete against every other + # deleter. await storage.delete_unreferenced( store._pool, binding_id, pairs, obligation_id=obligation_id ) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_storage.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_storage.py index af9f32f31d..35c0806411 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_storage.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_storage.py @@ -26,7 +26,7 @@ import inspect import json import zlib -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Collection, Mapping, Sequence from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, TypeVar @@ -371,6 +371,25 @@ def _assert_native_version_ref(value: object) -> None: ) +def recorded_intent_objects(created_objects: object) -> set[tuple[str, str]]: + """The ``(object_ref, native_version_ref)`` pairs a write intent recorded. + + Reads both the current keys and an older writer's pre-alignment ``objectKey`` and + ``nativeVersion``, which a writer still running during a rolling upgrade records. + """ + pairs: set[tuple[str, str]] = set() + if not isinstance(created_objects, list): + return pairs + for item in created_objects: + if not isinstance(item, dict): + continue + object_ref = item.get("object_ref", item.get("objectKey")) + version = item.get("native_version_ref", item.get("nativeVersion")) + if isinstance(object_ref, str) and isinstance(version, str): + pairs.add((object_ref, version)) + return pairs + + def _inflate(physical: bytes, compression: Compression, byte_count: int) -> bytes | None: """Decompress to exactly ``byte_count`` bytes, refusing trailing data and extra members.""" if compression == "none": @@ -1408,13 +1427,25 @@ async def delete_collector_fence( # -- object lifetime -------------------------------------------------- # - # Chunks are adopted create-only, so a custom key template that omits an identity - # placeholder lets two row sets name the same (binding, key, version). Every path that - # deletes an object and every commit that adopts one serialize on a transaction-scoped - # advisory lock per object: the deleter checks references and deletes under it, and the - # committer re-verifies the adopted object under it, so whichever runs second sees the - # other's effect. The lock text is shared with the JavaScript SDK and hashed in SQL so - # both SDKs derive the same lock id in one database. + # The cross-SDK contract (adcp-client#3606, adcp-client-python#1492). The per-object advisory + # lock ``adcp-reporting-object:{binding}:{object_ref}`` is kept in exactly two places that pair + # with each other, because nothing else orders them: the orphan collector's delete + # (``delete_orphaned_object``) and a commit that adopts an object its own intent did not record + # (``lock_and_verify_adopted``). Retirement, the intent sweep and a commit that adopts an object + # its own intent recorded take no object lock and hold no connection across a provider delete. + # Lock order everywhere: the account lock, then object locks. The lock text is hashed in SQL so + # both SDKs derive one lock id from it in one database. + # + # Fences that exist without the lock: (A) the account lock held by a commit, retirement phase 1, + # an intent takeover and the sweep claim; (B) ``rows_state`` moves to ``pruning`` under the + # account lock before any delete, and ``pruning`` row sets neither count as references nor can + # be shared; (C) the intent moves ``open -> sweeping`` against the commit's ``DELETE ... WHERE + # state = 'open'``, in the same transaction as the chunk insert, so a reader sees the open + # intent or the chunk rows, never neither; (D) deletes name an exact native version, so native + # versions must be write-unique; (E) every writable key includes ``{revision_key}`` and + # ``{content_sha256}``, so only one row set can name a key through a writable binding. + # Precondition, documented and not enforced: two writable bindings must not share one bucket and + # prefix, because the reference check is per binding outside the collector. @staticmethod def _object_lock_text(binding_id: str, object_ref: str) -> str: @@ -1425,19 +1456,33 @@ async def lock_and_verify_adopted( connection: Any, binding_id: str, locators: Sequence[ReportingRowChunkLocatorV1], + recorded: Collection[tuple[str, str]] = (), ) -> None: - """Hold the object lock for every adopted object and prove each still exists. - - Call inside the ledger transaction, after the account lock and before the chunk rows - are inserted. Objects this upload created are protected by its open write intent - instead. (A writer's own just-created object is not locked: the window between its - create and the intent record is the same one the JavaScript SDK leaves open.) + """Prove every adopted object is still there, locking the ones no record names. + + Call inside the ledger transaction, after the account lock and the intent close, before + the chunk rows are inserted. ``recorded`` is the ``(object_ref, native_version_ref)`` pairs + the closed intent listed in ``created_objects`` (``recorded_intent_objects``). + + An object the 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 sweep cannot claim an intent this commit + just closed or one a takeover renewed, and the key is unique to this row set. An adopted + object the intent does not record (an earlier attempt's PUT that landed before it was + recorded, or one 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, so take it here, in key + order, until this transaction ends: a delete that held it finished first and the + version-pinned ``stat`` fails this commit, and any later delete sees the chunk rows + inserted next and skips. (A writer's own just-created object is not locked: the window + between its create and the intent record is the same one the JavaScript SDK leaves open.) """ adopted = sorted((x for x in locators if not x.created), key=lambda x: x.object_ref) if not adopted: return binding, provider = self._object_binding(binding_id) for locator in adopted: + if (locator.object_ref, locator.native_version_ref) in recorded: + continue await connection.execute( "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", (self._object_lock_text(binding_id, locator.object_ref),), @@ -1496,91 +1541,125 @@ async def delete_unreferenced( *, obligation_id: str | None = None, row_set_id: str | None = None, - any_binding: bool = False, ) -> int: """Delete each object nothing outside the operation's scope still needs. - An object is kept while a chunk of a live or unavailable row set of the same binding - references it (outside ``obligation_id`` / ``row_set_id``, the thing being deleted), - or another open write intent created it. Row sets already pruning or pruned do not - count: their own operation deletes the object. One transaction per object: the lock, - the reference check and the provider delete happen together, so a concurrent commit - adopting the object either already references it (it is kept) or finds it gone and - fails closed. Returns the number of objects deleted. - - ``any_binding`` widens both checks to every binding and takes every binding's object lock, - in a fixed order: two bindings can share a bucket and prefix, and a writer adopting an - object takes the lock of its own binding only, so the delete must be serialized with an - adoption through any of them. Writers hold one such lock, so no cycle forms. The orphan - collector deletes this way. + Takes no advisory lock and holds no transaction or pooled connection across the provider + delete: per object, one statement checks the references, then the exact version is + deleted. An object is kept while a chunk of a live or unavailable row set of the same + binding references it (outside ``obligation_id`` / ``row_set_id``, the thing being + deleted), or another 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. Each deleter commits + ``pruning`` before its check, so for any two at least one deletes. See the contract above + (``# -- object lifetime``) for why no lock is needed against a committing writer. A skipped + object stays for the row set that owns it: its own retirement or sweep deletes it. + Returns the number of objects deleted. """ deleted = 0 for object_ref, native_version_ref in objects: - async with pool.connection() as connection, connection.transaction(): - lock_bindings = [binding_id] - if any_binding: - known = await ( - await connection.execute( - "SELECT row_binding_id FROM adcp_reporting_row_bindings" - ) - ).fetchall() - lock_bindings = sorted( - {binding_id, *self.object_binding_ids(), *(row[0] for row in known)} - ) - for lock_binding in lock_bindings: - await connection.execute( - "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", - (self._object_lock_text(lock_binding, object_ref),), - ) - referenced = await ( - await connection.execute( - "(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 (%(any_binding)s::boolean OR row_set.row_binding_id = %(binding)s)" - " AND chunk.object_ref = %(key)s" - " AND chunk.native_version_ref = %(version)s" - " AND row_set.rows_state IN ('live', 'unavailable')" - " AND (%(obligation)s::text IS NULL" - " OR chunk.obligation_id <> %(obligation)s)" - " AND (%(row_set)s::text IS NULL OR chunk.row_set_id <> %(row_set)s)" - " LIMIT 1)" - " UNION ALL" - " (SELECT 1 FROM adcp_reporting_row_write_intents intent" - " WHERE (%(any_binding)s::boolean OR intent.row_binding_id = %(binding)s)" - " AND intent.state = 'open'" - " AND (intent.created_objects @> %(created)s::jsonb" - " OR intent.created_objects @> %(created_legacy)s::jsonb)" - " LIMIT 1)", - { - "binding": binding_id, - "any_binding": any_binding, - "key": object_ref, - "version": native_version_ref, - "obligation": obligation_id, - "row_set": row_set_id, - "created": json.dumps( - [ - { - "object_ref": object_ref, - "native_version_ref": native_version_ref, - } - ] - ), - # A writer from before the AdCP field alignment, still running during a - # rolling upgrade, records the earlier keys. - "created_legacy": json.dumps( - [{"objectKey": object_ref, "nativeVersion": native_version_ref}] - ), - }, - ) - ).fetchone() - if referenced is not None: - continue - await self.delete_objects(binding_id, [(object_ref, native_version_ref)]) - deleted += 1 + async with pool.connection() as connection: + referenced = await self._is_referenced( + connection, + binding_id, + object_ref, + native_version_ref, + obligation_id=obligation_id, + row_set_id=row_set_id, + ) + if referenced: + continue + await self.delete_objects(binding_id, [(object_ref, native_version_ref)]) + deleted += 1 return deleted + async def delete_orphaned_object( + self, pool: Any, binding_id: str, object_ref: str, native_version_ref: str + ) -> bool: + """Delete one object no ledger record named when the collector listed it. + + Under the advisory lock ``adcp-reporting-object:{binding}:{object_ref}`` of every binding + that can name it: 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, sorted by binding ID. The reference check then runs across all bindings. + Lock order: the account lock, then object locks; this never waits on the account lock while + holding an object lock. Returns whether the object was deleted. + """ + async with pool.connection() as connection, connection.transaction(): + known = await ( + await connection.execute("SELECT row_binding_id FROM adcp_reporting_row_bindings") + ).fetchall() + for lock_binding in sorted( + {binding_id, *self.object_binding_ids(), *(r[0] for r in known)} + ): + await connection.execute( + "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", + (self._object_lock_text(lock_binding, object_ref),), + ) + if await self._is_referenced( + connection, binding_id, object_ref, native_version_ref, any_binding=True + ): + return False + await self.delete_objects(binding_id, [(object_ref, native_version_ref)]) + return True + + async def _is_referenced( + self, + connection: Any, + binding_id: str, + object_ref: str, + native_version_ref: str, + *, + obligation_id: str | None = None, + row_set_id: str | None = None, + any_binding: bool = False, + ) -> bool: + """Whether a live or unavailable row set outside the scope, or an open write intent, still + names this exact object version. ``any_binding`` widens the check to every binding.""" + found = await ( + await connection.execute( + "(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 (%(any_binding)s::boolean OR row_set.row_binding_id = %(binding)s)" + " AND chunk.object_ref = %(key)s" + " AND chunk.native_version_ref = %(version)s" + " AND row_set.rows_state IN ('live', 'unavailable')" + " AND (%(obligation)s::text IS NULL OR chunk.obligation_id <> %(obligation)s)" + " AND (%(row_set)s::text IS NULL OR chunk.row_set_id <> %(row_set)s)" + " LIMIT 1)" + " UNION ALL" + " (SELECT 1 FROM adcp_reporting_row_write_intents intent" + " WHERE (%(any_binding)s::boolean OR intent.row_binding_id = %(binding)s)" + " AND intent.state = 'open'" + " AND (intent.created_objects @> %(created)s::jsonb" + " OR intent.created_objects @> %(created_legacy)s::jsonb)" + " LIMIT 1)", + { + "binding": binding_id, + "any_binding": any_binding, + "key": object_ref, + "version": native_version_ref, + "obligation": obligation_id, + "row_set": row_set_id, + "created": json.dumps( + [{"object_ref": object_ref, "native_version_ref": native_version_ref}] + ), + # A writer from before the AdCP field alignment, still running during a + # rolling upgrade, records the earlier keys. + "created_legacy": json.dumps( + [{"objectKey": object_ref, "nativeVersion": native_version_ref}] + ), + }, + ) + ).fetchone() + return found is not None + # -- read ------------------------------------------------------------- async def find( diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_write.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_write.py index 7f36249938..01a986da0b 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_write.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/row_write.py @@ -44,6 +44,7 @@ ReportingPreparedRowSetV1, ReportingRowChunkLocatorV1, ReportingRowStoreError, + recorded_intent_objects, ) if TYPE_CHECKING: @@ -338,16 +339,25 @@ async def persist_revision_row_set( "STATE_UNAVAILABLE", "the shared row set is no longer live" ) else: + # 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. closed = await connection.execute( "DELETE FROM adcp_reporting_row_write_intents" " WHERE row_binding_id = %s AND row_set_id = %s AND content_sha256 = %s" - " AND owner_token = %s AND state = 'open'", + " AND owner_token = %s AND state = 'open' RETURNING created_objects", (prepared.binding_id, prepared.row_set_id, prepared.content_sha256, staged.owner_token), ) - if closed.rowcount != 1: + closed_row = await closed.fetchone() + if closed_row is None: raise ReportingRowStoreError( "STATE_UNAVAILABLE", "the row upload intent was taken over or swept" ) - if staged.shared_from is None: - await storage.lock_and_verify_adopted(connection, prepared.binding_id, staged.locators) + # Adopted objects the intent recorded take no lock; any other adopted object takes its + # binding's object lock, then every adopted object is proven by a version-pinned stat. + await storage.lock_and_verify_adopted( + connection, + prepared.binding_id, + staged.locators, + recorded_intent_objects(closed_row[0]), + ) await storage.write_in_transaction(connection, prepared, staged.locators, staged.shared_from) diff --git a/tests/conformance/reporting/test_reporting_row_object_lock_contract_pg.py b/tests/conformance/reporting/test_reporting_row_object_lock_contract_pg.py new file mode 100644 index 0000000000..e1d031fc2a --- /dev/null +++ b/tests/conformance/reporting/test_reporting_row_object_lock_contract_pg.py @@ -0,0 +1,293 @@ +"""The per-object advisory lock is kept only where nothing else orders the two sides. + +Ported from the JavaScript SDK (adcp-client #3606, ``reporting-row-shared-object-pg``). Retirement, +the intent sweep and a commit that adopts an object its own intent recorded take no object lock; the +orphan collector's delete and a commit that adopts an object no record names pair on it. +""" + +from __future__ import annotations + +import asyncio +import hashlib +from pathlib import Path +from typing import Any + +import pytest +from adcp_reporting_storage.ledger.row_storage import ( + ReportingRowChunkLocatorV1, + ReportingRowStoreError, + recorded_intent_objects, +) + +from ._orphan_support import Fixture, orphan_fixture +from .test_reporting_row_storage_pg import fetch_all + +BODY = b'{"a":1}\n' + + +async def lock_holder(fx: Fixture, binding: str, object_ref: str) -> Any: + """A connection holding the per-object lock of ``binding`` until it commits.""" + storage = fx.store._row_storage + assert storage is not None + holder = await fx.pool.getconn() + await holder.execute( + "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", + (storage._object_lock_text(binding, object_ref),), + ) + return holder + + +async def release(fx: Fixture, holder: Any) -> None: + await holder.commit() + await fx.pool.putconn(holder) + + +def locator(object_ref: str, version: str) -> ReportingRowChunkLocatorV1: + return ReportingRowChunkLocatorV1( + object_ref, version, "gzip", hashlib.sha256(BODY).hexdigest(), len(BODY), created=False + ) + + +async def verify( + fx: Fixture, binding: str, loc: ReportingRowChunkLocatorV1, recorded: Any = () +) -> None: + storage = fx.store._row_storage + assert storage is not None + async with fx.pool.connection() as connection, connection.transaction(): + await storage.lock_and_verify_adopted(connection, binding, [loc], recorded) + + +async def blocks( + fx: Fixture, binding: str, loc: ReportingRowChunkLocatorV1, recorded: Any = () +) -> bool: + """Whether the adoption waits on the object lock (and then completes once it is released).""" + task = asyncio.create_task(verify(fx, binding, loc, recorded)) + done, _ = await asyncio.wait({task}, timeout=0.6) + waited = not done + if waited: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + return waited + + +async def idle_in_transaction(fx: Fixture) -> int: + rows = await fetch_all( + fx.pool, + "SELECT count(*) FROM pg_stat_activity WHERE state LIKE 'idle in transaction%%'" + " AND datname = current_database() AND pid <> pg_backend_pid()", + ) + return int(rows[0][0]) + + +async def test_a_delete_of_a_referenced_object_takes_no_lock_and_holds_no_connection_across_the_provider_delete( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + ref, version = await fx.commit_revision("lock_free_retire") + storage = fx.store._row_storage + assert storage is not None + holder = await lock_holder(fx, "objects", ref) # the lock is held elsewhere throughout + seen: list[int] = [] + + async def during_delete(object_ref: str) -> None: + seen.append(await idle_in_transaction(fx)) + + fx.provider.on_delete = during_delete + try: + # Retirement scopes the delete to its own obligation, so its own references do not count. + removed = await asyncio.wait_for( + storage.delete_unreferenced( + fx.pool, + "objects", + [(ref, version)], + obligation_id=fx.obligation.reporting_obligation_id, + ), + 10, + ) + assert removed == 1 and not fx.on_disk(ref) + # Only the lock holder is idle in a transaction: nothing of the deleter's is open. + assert seen == [1] + finally: + await release(fx, holder) + + +async def test_a_referenced_object_is_kept_and_a_pruning_row_set_does_not_count( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + ref, version = await fx.commit_revision("kept") + storage = fx.store._row_storage + assert storage is not None + assert await storage.delete_unreferenced(fx.pool, "objects", [(ref, version)]) == 0 + assert fx.on_disk(ref) + async with fx.pool.connection() as connection: + await connection.execute("UPDATE adcp_reporting_row_sets SET rows_state = 'pruning'") + # Each deleter commits ``pruning`` before its check, so for any two one of them deletes. + assert await storage.delete_unreferenced(fx.pool, "objects", [(ref, version)]) == 1 + assert not fx.on_disk(ref) + + +async def test_the_intent_sweep_takes_no_object_lock_while_the_lock_is_held_elsewhere( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + ref = fx.key("acct/swept.jsonl.gz") + version = await fx.put_object(ref) + await fx.add_intent("rrev_expired", ref, version) + async with fx.pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_row_write_intents" + " SET expires_at = clock_timestamp() - INTERVAL '1 hour'" + ) + holder = await lock_holder(fx, "objects", ref) + try: + swept = await asyncio.wait_for(fx.store.sweep_row_write_intents(), 15) + assert swept.swept == 1 and swept.objects_deleted == 1 and not fx.on_disk(ref) + finally: + await release(fx, holder) + + +async def test_an_adoption_of_an_object_the_intent_recorded_takes_no_lock(tmp_path: Path) -> None: + async with orphan_fixture(tmp_path) as fx: + ref = fx.key("acct/own.jsonl.gz") + version = await fx.put_object(ref, BODY) + holder = await lock_holder(fx, "objects", ref) + try: + recorded = {(ref, version)} + await asyncio.wait_for(verify(fx, "objects", locator(ref, version), recorded), 5) + finally: + await release(fx, holder) + + +async def test_the_pre_alignment_intent_keys_are_read_so_an_older_writers_object_takes_no_lock() -> ( + None +): + pairs = recorded_intent_objects( + [ + {"object_ref": "a", "native_version_ref": "1"}, + {"objectKey": "b", "nativeVersion": "2"}, + {"object_ref": "c", "nativeVersion": "3"}, # mixed spellings are tolerated + {"object_ref": "d"}, + {"unrelated": True}, + "not an object", + {"object_ref": 5, "native_version_ref": "9"}, + ] + ) + assert pairs == {("a", "1"), ("b", "2"), ("c", "3")} + assert recorded_intent_objects(None) == set() and recorded_intent_objects({}) == set() + + +async def test_an_adoption_takes_the_lock_for_a_different_version_or_an_object_the_intent_does_not_name( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + ref = fx.key("acct/adopted.jsonl.gz") + version = await fx.put_object(ref, BODY) + other = fx.key("acct/other.jsonl.gz") + holder = await lock_holder(fx, "objects", ref) + try: + # The intent names the key at another native version: that is not this object. + assert await blocks(fx, "objects", locator(ref, version), {(ref, "9:9:9:9")}) + # The intent names another object, or none. + assert await blocks(fx, "objects", locator(ref, version), {(other, version)}) + assert await blocks(fx, "objects", locator(ref, version)) + finally: + await release(fx, holder) + # Once the lock is free the adoption proves the object and succeeds. + await asyncio.wait_for(verify(fx, "objects", locator(ref, version)), 5) + + +async def test_an_orphan_delete_in_flight_blocks_an_adopter_of_the_unrecorded_object_which_then_fails_cleanly( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + ref = fx.key("acct/raced.jsonl.gz") + version = await fx.put_object(ref, BODY) + storage = fx.store._row_storage + assert storage is not None + deleting = asyncio.Event() + proceed = asyncio.Event() + + async def pause(object_ref: str) -> None: + deleting.set() + await proceed.wait() + + fx.provider.on_delete = pause + orphan = asyncio.create_task( + storage.delete_orphaned_object(fx.pool, "objects", ref, version) + ) + await asyncio.wait_for(deleting.wait(), 10) + adopter = asyncio.create_task(verify(fx, "objects", locator(ref, version))) + done, _ = await asyncio.wait({adopter}, timeout=0.6) + assert not done, "the adoption waits for the delete that holds the object lock" + proceed.set() + assert await asyncio.wait_for(orphan, 10) is True + with pytest.raises(ReportingRowStoreError) as failure: + await asyncio.wait_for(adopter, 10) + assert failure.value.code == "STATE_UNAVAILABLE", "the version-pinned stat fails it cleanly" + + +async def test_an_orphan_delete_takes_every_bindings_lock_for_the_object(tmp_path: Path) -> None: + async with orphan_fixture(tmp_path) as fx: + ref = fx.key("acct/shared.jsonl.gz") + version = await fx.put_object(ref, BODY) + storage = fx.store._row_storage + assert storage is not None + # An adopter through the twin binding (same bucket and prefix) holds only the twin's lock. + holder = await lock_holder(fx, "twin", ref) + task = asyncio.create_task(storage.delete_orphaned_object(fx.pool, "objects", ref, version)) + done, _ = await asyncio.wait({task}, timeout=0.6) + assert not done and fx.on_disk(ref) + await release(fx, holder) + assert await asyncio.wait_for(task, 10) is True and not fx.on_disk(ref) + + +async def test_an_orphan_delete_keeps_an_object_a_record_in_another_binding_names( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + ref = fx.key("acct/named.jsonl.gz") + version = await fx.put_object(ref, BODY) + await fx.add_intent("rrev_twin", ref, version, binding="twin") + storage = fx.store._row_storage + assert storage is not None + assert await storage.delete_orphaned_object(fx.pool, "objects", ref, version) is False + assert fx.on_disk(ref) + + +async def test_a_writer_whose_intent_expired_before_its_commit_cannot_commit_the_swept_objects( + tmp_path: Path, +) -> None: + async with orphan_fixture(tmp_path) as fx: + from .test_reporting_row_storage_pg import many_rows, seeded, typed_revision + + fx.obligation = await seeded(fx.store) + async with fx.pool.connection() as connection: + await connection.execute( + "UPDATE reporting_configurations SET lease_worker_id = 'worker-1'," + " lease_expires_at = clock_timestamp() + INTERVAL '10 minutes'" + ) + rows = many_rows(4) + revision = typed_revision(fx.obligation, rows, suffix="swept_writer") + assert await fx.store.stage_revision_rows(revision, rows) is True + async with fx.pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_row_write_intents" + " SET expires_at = clock_timestamp() - INTERVAL '1 hour'" + ) + swept = await fx.store.sweep_row_write_intents() + assert swept.swept == 1 and swept.objects_deleted == 1 + # Exactly one of the sweep claim and the commit's intent close wins; the writer lost. + with pytest.raises(ReportingRowStoreError) as failure: + await fx.store.commit_revision(revision, rows) + assert failure.value.code == "STATE_UNAVAILABLE" + assert await fetch_all(fx.pool, "SELECT count(*) FROM adcp_reporting_row_sets") == [(0,)] + + +def test_the_lock_text_is_unchanged_and_shared_with_the_javascript_sdk() -> None: + from adcp_reporting_storage.ledger.row_storage import ReportingRowStorageV1 + + assert ( + ReportingRowStorageV1._object_lock_text("reports", "adcp-rows/ns/a/b") + == "adcp-reporting-object:reports:adcp-rows/ns/a/b" + )