Opt-in batched terminal flush#131
Merged
Merged
Conversation
Opt-in terminal_flush_batch_size (default 1 = per-row, bit-for-bit identical). A per-worker coalescer flushes completed plain-delete rows as one DELETE ... RETURNING, collapsing the terminal round-trip from 1/message to 1/batch; retry and DLQ stay per-row. Records the accumulated-redelivery tradeoff (opt-in, ungraceful-crash only) and a gated benchmark before/after.
…ated) Add the public configuration knob terminal_flush_batch_size with default value of 1 and validation that it must be >= 1. Thread the knob through all five construction paths (config, factory, registrator, router, fastapi/router) and update test fixtures that construct subscribers. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Route batchable terminal deletes (plain DELETE, no DLQ payload) through a per-worker buffer flushed as one delete_batch_with_lease when it reaches terminal_flush_batch_size or the inflight queue goes idle. task_done and the per-row metric are deferred to flush-time so _inflight.join() stays an honest drain barrier; _flush_buffer task_done's every row even on DB error so join() cannot hang. At terminal_flush_batch_size==1 and on the test-broker path (buffer=None) every row stays on the inline flush path, bit-for-bit identical. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… path The existing batched-terminal-flush tests all set terminal_flush_batch_size above the seeded row count, so the buffer's size trigger (len(buffer) >= batch_size) never fired -- every flush went through the idle path instead. Add a test that keeps the inflight queue saturated to force the size trigger, and a test proving a DLQ-writing terminal row stays on the inline CTE path (not silently dropped by a batched delete) even when batching is enabled alongside successful rows. Also correct architecture/drain.md: a timed-out drain does not unconditionally abandon buffered rows to lease-expiry -- _worker_inner's finally still attempts a best-effort buffer flush on loop exit, and only rows that flush fails to reach fall back to lease-expiry redelivery.
Task 4 of the batched-terminal-flush change. Drain safety: - Add test_graceful_drain_flushes_partial_batch: seeds fewer rows than terminal_flush_batch_size so the buffer never fills, then asserts a graceful stop() leaves the outbox empty (the drain barrier flushes the partial buffer, no redelivery). Benchmark: - Add RunConfig.terminal_flush_batch_size (default 1), thread it into run_consumer's subscriber, and add one batched sweep point (consumer/w1/b100/tfbs100). Extend report._key with a /tfbsN segment (only when >1) so the batched point does not collide with its tfbs=1 sibling. - Regenerate baseline.json. Measured delete_calls at the batched point across 5 runs: deterministic 50 (== messages/tfbs), tup_del == 5000 every run. Because fetch_batch_size == terminal_flush_batch_size delivers full buffers and empty fetch polls never fire a flush, delete_calls is structural, so it stays exact-gated for all points (documented at EXACT_KEYS). bench-check passes. Close pre-existing coverage gaps in the batched-flush code (Task 1-3) so the 100% gate holds: - Unit tests for _flush_buffer (empty-buffer no-op, DB-error task_done+reraise, mixed acked/lease_lost from the RETURNING set) and _worker_inner's loop-exit flush, plus the real client's delete_batch_with_lease None-conn guard. - Replace the batched integration tests' inline row-count poll loops with _wait_until on the acked-event count (the inline sleep was skipped, and thus coverage-flaky, whenever the flush emptied the table before the first poll). - Rework the lease-steal in test_batched_flush_lease_lost to run in the main task via an event handshake; its final async-with is pragma'd (executes, but coverage cannot trace a SQLAlchemy-async await concurrent with the live broker's greenlet work; the invariant is covered by the _flush_buffer unit test). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The best-effort `_flush_buffer` in `_worker_inner`'s `finally` is only reachable on a drain timeout, where it `await`s a DELETE during task cancellation. A wedged connection there could extend `stop()` past its budget. Wrap the flush in `asyncio.timeout(_FLUSH_ON_EXIT_TIMEOUT_SECONDS)` (2.0s); on timeout the rows fall back to lease-expiry redelivery, the same fate as any in-flight row on a drain timeout. Also: - relabel the graceful-drain integration test to name its real mechanism (idle-flush + `_inflight.join()` barrier, not the exit-flush finally); - add a unit test proving the exit flush is timeout-bounded when the DELETE hangs; - sharpen the exact-gate revisit trigger (report.py + spec) to include `messages % tfbs != 0`; - correct the spec throughput figures to the committed baseline (~450 -> ~9670 msg/s); - note the timeout bound in architecture/drain.md. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
_flush_buffer only cleaned up task_done accounting on Exception, but the bounded exit-flush timeout cancels via CancelledError (a BaseException), skipping that cleanup and leaking _inflight's unfinished-task count across the reused queue -- wedging every subsequent graceful stop()'s join(). Widen the catch to BaseException so cancellation balances task_done too.
…mula, exit-flush comment Final-review accuracy nits: name the non-yielding bench handler as a third condition for the exact delete_calls gate; correct the lease-bound ceiling to fetch_batch_size + max_workers x (tfbs + 1); note the exit-flush is also reachable on a mid-operation inline-flush error with a non-empty buffer.
…11 coverage The bounded exit-flush tests await _worker_inner directly, so the internal asyncio.timeout cancellation propagates through the test frame and corrupts coverage.py's line tracer on Python 3.11 only (phantom sub-100% miss on adjacent assertion lines). Wrapping the call in asyncio.create_task keeps the cancellation inside the task frame. Verified: tests/test_unit.py 100% under 3.11.
lesnik512
added a commit
that referenced
this pull request
Jul 15, 2026
#132) Add the knob to the subscriber-options table and a dedicated 'Batching terminal deletes' section: the 100x round-trip win, the idle-flush no-latency behavior, and the at-least-once redelivery tradeoff on ungraceful crash. Off by default; recommended per-subscriber for high-throughput idempotent handlers. Fills the user-facing gap (the feature shipped in #131 with only internal architecture docs).
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
1 participant
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.
Adds an opt-in batched terminal DELETE to the outbox subscriber. Off by default (
terminal_flush_batch_size=1= today's per-row path, bit-for-bit identical); set it>1to opt in.Design and the measurements behind it (the change's truth home):
planning/changes/2026-07-15.01-batched-terminal-flush.md. Builds on the benchmark harness (#130).What it does
The subscriber's terminal write is one
DELETEround-trip per message — the benchmark baseline measures exactlydelete_calls == messages, and atmax_workers=1those DELETEs serialize (~840 µs/message on loopback). This adds a per-worker coalescer: each worker accumulates completed batchable rows and flushes them as onecollapsing the terminal round-trip from 1/message to 1/batch. The
RETURNINGset is the per-row lease check (a reclaimed row isn't returned → treated aslease_lost), preserving the lease-token invariant with no new mechanism.Batchable = plain terminal DELETE with no DLQ payload (successful acks, and terminal-failures without a
dlq_table). The retryUPDATEand the DLQDELETE…INSERTCTE stay per-row/inline — they carry per-row state and are the exceptional paths, not the bottleneck.Correctness
task_doneaccounting is the drain barrier.task_donemoves to flush-time for buffered rows so_inflight.join()still waits for the actual delete, not merely for dispatch._flush_bufferbalancestask_doneon every exit — success, DB error, and cancellation (except BaseException), so a bounded exit-flush timeout can't leak the unfinished-count into a later start/stop cycle.terminal_flush_batch_size=1and the test-broker path (buffer=None) both take the unchanged inline path. Metric emission and its P17 ordering are preserved on both paths.stop()flushes the buffer via the idle-flush +join()barrier (no redelivery). On a drain timeout, a bounded (asyncio.timeout, 2s) best-effort flush runs so a wedged connection can't hangstop(); unflushed rows fall back to lease-expiry.The tradeoff (opt-in, documented)
Batching accumulates completed rows before the DELETE lands, so on an ungraceful crash up to
batch_sizecompleted-but-undeleted rows redeliver. The outbox is already at-least-once (handlers must be idempotent), so this is not a new failure class — and it's off by default, so no existing deployment changes on upgrade.Benchmark (gated before/after)
New sweep point
consumer/w1/b100/tfbs100:delete_callsdrops 5000 → 50 (messages / tfbs) whiletup_delstays 5000 (same rows deleted, fewer round-trips), ~21× wall-clock.just bench-checkexact-gates both; the exact gate on the batched count is safe because the bounded_inflightqueue makes it structural whenmessages % tfbs == 0(revisit-trigger documents the flake boundary).Verification
--cov-fail-under=100.delete_batch_with_lease); adversarial drain/lease-loss/mixed-batch integration + unit tests.architecture/subscriber.mdanddrain.mdpromoted in-branch.🤖 Generated with Claude Code