feat(events): background async batching writer for PostgresEventStore - #70
Merged
Conversation
AgentLoop._emit() previously awaited PostgresEventStore.append() inline in the critical path — one pool.acquire() + one INSERT round-trip per event, every event, no create_task, no queue. Against a pooled connection (Supabase's documented deployment target) that's plausibly 20-100ms per event, several times per agent step. Confirmed real, not speculative: traced every call site. append() now enqueues onto a bounded asyncio.Queue and returns immediately; a single background task drains it in batches (executemany, _DRAIN_BATCH_SIZE=100, _DRAIN_INTERVAL_SECONDS=0.5) via one sequential writer — never parallelized across pool_max_size, since seq is BIGSERIAL assigned at INSERT time and every read path depends on ORDER BY seq ASC reflecting true append order; a concurrent multi-connection drain would silently corrupt that. This relaxes what a successful append() return means: "accepted for write," not "durably written" — documented explicitly in both EventStorePort.append's docstring (the contract every adapter shares) and this module's own. A drain-time Postgres failure is logged (postgres_event_store_drain_failed), never raised — there's no caller left to raise to by the time the drain runs. flush() is the new explicit synchronization point for a caller that needs the durable guarantee back. get_session_events/list_session_ids/get_latest_session_id all call flush() before reading, so a resume immediately after several append() calls can never silently miss events still sitting in the queue — Agent._load_session_history depends on this for correct replay. close() now flushes pending events, cancels the drain task, and closes the pool, in that order — no event loss on shutdown (the prior implementation only closed the pool). Idempotent via a _closed guard; append() after close() raises rather than silently enqueuing into a store that will never drain it. 11 new/rewritten tests, including the one that actually matters: strict seq-order preservation across 50 concurrently-issued append() calls (asyncio.gather, not sequential test code) — the property the single- writer design exists to guarantee. Also: drain failures logged not raised, batch size never exceeds _DRAIN_BATCH_SIZE, flush-before-read on all three read methods, close() flushes before closing, append-after- close raises.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Increment 3 of 4 in the approved plan (SearXNG → Crawl4AI/Wayback → Postgres batching → OTLP+StateSummary). Not stacked — this touches
events/, independent of the two web-tool PRs.AgentLoop._emit()previously awaitedPostgresEventStore.append()inline in the critical path — onepool.acquire()+ one INSERT round-trip per event, every event, nocreate_task, no queue. Against a pooled connection (Supabase's documented deployment target) that's plausibly 20-100ms per event, several times per agent step. Confirmed real by tracing every call site, not speculative.Changes
append()now enqueues onto a boundedasyncio.Queueand returns immediately; a single background task drains it in batches (executemany, batch size 100, 0.5s interval) via one sequential writer — never parallelized acrosspool_max_size, sinceseqisBIGSERIALassigned at INSERT time and every read path depends onORDER BY seq ASCreflecting true append order. A concurrent multi-connection drain would silently corrupt that invariant.append()return means: "accepted for write," not "durably written" — documented explicitly in bothEventStorePort.append's docstring (the shared contract) and this module's own. A drain-time Postgres failure is logged, never raised (there's no caller left to raise to).flush()is the new explicit synchronization point for a caller that needs the durable guarantee back.get_session_events/list_session_ids/get_latest_session_idall callflush()before reading, so a resume immediately after severalappend()calls can never silently miss events still in the queue —Agent._load_session_historydepends on this for correct replay.close()now flushes pending events, cancels the drain task, and closes the pool, in that order — no event loss on shutdown. Idempotent;append()afterclose()raises rather than silently enqueuing into a store that will never drain it.Test plan
seq-order preservation across 50 concurrently-issuedappend()calls (asyncio.gather, not sequential test code) — the property the single-writer design exists to guarantee.close()flushes before closing; append-after-close raises.uv run ruff check ./uv run ruff format --check ./uv run pyright— all clean.