Repository navigation
feat(pg): coalesce same-key PgBackend.hold() in-process, and offer LazyBackend-style lazy open for the other Pg stores #1297
Description
Summary
Two gaps on 8.0.0, both in the Postgres stores:
PgBackend.hold()takes onelock_poolconnection per concurrent request, even when those requests are duplicates of the same key in the same process (adcp/server/idempotency/backends.py:572-586). Each one checks out a connection, then blocks inpg_advisory_xact_lock. A burst of buyer retries for a single key can tie up every lock connection in a worker, which stalls idempotent writes for unrelated keys until the first handler finishes.- Only the idempotency backend has a lazy-open path.
LazyBackend(server/idempotency/lazy.py) resolves the backend once on first use.PgProposalStore(decisioning/pg/proposal_store.py:361-377) andPgTaskRegistry(decisioning/pg/task_registry.py:135-163) require a pool that is already open, from the caller, on the serving loop. Their docs say "We don't open, own, or close the pool."
Reproduction (adcp 8.0.0)
The script below uses a stub pool that emulates pg_advisory_xact_lock and counts how many lock_pool connections are checked out at once. PgBackend itself is the real 8.0.0 class.
import asyncio
from contextlib import asynccontextmanager
from adcp.server.idempotency import PgBackend
class StubPool:
def __init__(self):
self.checked_out = self.peak = 0
self.advisory = {}
@asynccontextmanager
async def connection(self):
self.checked_out += 1
self.peak = max(self.peak, self.checked_out)
conn = StubConn(self)
try:
yield conn
finally:
if conn.held is not None:
conn.held.release()
self.checked_out -= 1
class StubConn:
def __init__(self, pool):
self.pool, self.held = pool, None
@asynccontextmanager
async def transaction(self):
yield
async def execute(self, sql, params=()):
if "pg_advisory_xact_lock" in sql:
self.held = self.pool.advisory.setdefault(params[0], asyncio.Lock())
await self.held.acquire()
async def main():
lock_pool = StubPool()
backend = PgBackend(pool=StubPool(), lock_pool=lock_pool)
async def duplicate_request():
async with backend.hold("tenant:buyer", "same-key"):
await asyncio.sleep(0.05) # handler running
await asyncio.gather(*(duplicate_request() for _ in range(10)))
print("10 same-key requests in one process -> peak lock_pool connections:", lock_pool.peak)
asyncio.run(main())10 same-key requests in one process -> peak lock_pool connections: 10
MemoryBackend.hold already uses a per-key asyncio.Lock (backends.py:261-270). Putting the same lock in front of the advisory lock in PgBackend would bring the count down to one connection per distinct key per process.
Why an adopter hits this
Embedded Sales Agent overrides hold in _LazyBootstrapPgBackend (core/idempotency.py, ~70 LOC). The override keeps a reference-counted dict[(scope_key, key)] -> asyncio.Lock, so same-process duplicates wait in memory instead of each taking a scarce advisory-lock connection. The lock pool is sized by CORE_IDEMPOTENCY_LOCK_POOL_SIZE, default 10.
For lazy open, ESA subclasses PgProposalStore as _LazyOpenPgProposalStore (core/decisioning/proposal_store.py). The subclass overrides all eight public methods to call pool.open() once under a lock. The store singleton is built synchronously at platform construction and rebuilt per test database, so a one-time lifespan open() doesn't cover it. PgTaskRegistry is opened from a serve() startup hook (open_task_registry, core/decisioning/task_registry.py), which works but has to be wired by hand. Each new SDK store method has to be added to these overrides, or it hits a closed pool.
Ask
- In
PgBackend.hold, take a process-local per-keyasyncio.Lock(weak-valued, likeMemoryBackend) before checking out alock_poolconnection. - Make lazy construction general. Either accept a pool factory (
pool: AsyncConnectionPool | Callable[[], Awaitable[AsyncConnectionPool]]) onPgProposalStore,PgTaskRegistryandPgTaskWebhookOutbox, or shipLazyProposalStore/LazyTaskRegistrywrappers with the same resolve-once semantics asLazyBackend.
Activity
bokelley commentedon Oct 2, 2026
Triage
Classification: Feature request (two non-breaking optimizations for Postgres stores)
Bucket(s): middleware (idempotency), client (decisioning stores) — no exact-match labels found for these buckets; flagged in run summary
Status: ready-to-implement
Milestone: omitted — no target version named
What the experts said:
WeakValueDictionary+ a creation-guardasyncio.Lock(same pattern asMemoryBackend); GC safety holds because every waiting task keeps a strong reference. Ask 2 should use wrapper classes (LazyProposalStore/LazyTaskRegistry/LazyTaskWebhookOutbox) rather than a factory-param design — the factory approach breaks the construction-time pool identity check attask_registry.py:147(task_webhook_outbox._pool is pool).My take: Both asks have clear, correct implementations and are strictly additive. The reproduction script confirms the lock-pool exhaustion is real, and
LazyBackend(lazy.py) already proves the wrapper pattern works well in this codebase.Implementation scope:
Ask 1 —
PgBackend.hold()in-process coalescing (src/adcp/server/idempotency/backends.py:572):self._hold_guard = asyncio.Lock()andself._key_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock]toPgBackend.__init__. Under_hold_guard, fetch-or-create a per-keyasyncio.Lock; thenasync with key_lock:before checking out_lock_pool.connection(). MirrorsMemoryBackend.hold()(lines 261–270) exactly.lock_poolconnections == 1 for N concurrent same-key requests (adapting the repro script as a pytest fixture with a stub pool).Ask 2 —
LazyProposalStore,LazyTaskRegistry,LazyTaskWebhookOutbox(new modulesrc/adcp/decisioning/pg/lazy.py):LazyBackend(server/idempotency/lazy.py) with resolve-once asyncio.Lock semantics. Each wrapper delegates all public methods to the resolved store. Export fromadcp.decisioning.pg.__init__.LazyTaskRegistryfactory must yield a(PgTaskRegistry, PgTaskWebhookOutbox)pair so the pool-identity check fires on resolution, not construction.Required checks:
pyteston changed packages,mypy src/,ruff check .andblack --check ..Non-breaking rationale: Ask 1 changes no public API. Ask 2 adds new exported classes without modifying existing constructors.
Triaged by Claude Code. Session: https://claude.ai/code/session_01WXoZKZa7tEQCUhJVjqKKut
Generated by Claude Code
bokelley commentedon Oct 4, 2026
Completed by #1317 (same-key lock connection coalescing) and #1327 (lazy proposal/task/task-outbox store resolution on the serving loop, squash 5987c40).