Skip to content

feat(events): background async batching writer for PostgresEventStore - #70

Merged
netty-linux merged 2 commits into
masterfrom
feat/postgres-async-batching
Aug 9, 2026
Merged

feat(events): background async batching writer for PostgresEventStore#70
netty-linux merged 2 commits into
masterfrom
feat/postgres-async-batching

Conversation

@netty-linux

Copy link
Copy Markdown
Owner

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 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 by tracing every call site, not speculative.

Changes

  • append() now enqueues onto a bounded asyncio.Queue and returns immediately; a single background task drains it in batches (executemany, batch size 100, 0.5s interval) 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 invariant.
  • This relaxes what a successful append() return means: "accepted for write," not "durably written" — documented explicitly in both EventStorePort.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_id all call flush() before reading, so a resume immediately after several append() calls can never silently miss events still 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. Idempotent; append() after close() raises rather than silently enqueuing into a store that will never drain it.

Test plan

  • The test 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.
  • Drain failures logged not raised; batch size never exceeds the configured max; flush-before-read on all three read methods; close() flushes before closing; append-after-close raises.
  • 11 new/rewritten tests (2 of the original tests tested the old inline-write behavior directly and needed rewriting for the new semantics).
  • Full suite: 659 passed, 7 skipped, 84.78% coverage.
  • uv run ruff check . / uv run ruff format --check . / uv run pyright — all clean.

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.
@netty-linux
netty-linux merged commit 1b4ee99 into master Aug 9, 2026
11 checks passed
@netty-linux
netty-linux deleted the feat/postgres-async-batching branch August 9, 2026 12:05
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant