Skip to content

feat(outbox): collect exit verb on BatchOp - #97

Draft
bodymindarts wants to merge 4 commits into
mainfrom
feat/batch-op-collect
Draft

bodymindarts wants to merge 4 commits into
mainfrom
feat/batch-op-collect

Conversation

@bodymindarts

Copy link
Copy Markdown
Member

Summary

Adds collect to BatchOp as a sibling exit verb to defer() / commit(), completing the verb set introduced in #95.

Handlers entering via consume_in_batch could only exit with commit or defer — an event needing both direct op work now and a contribution to the Batch accumulator had to give up one or the other. BatchOp::collect_with (plus Vec/HashMap collect sugar, mirroring EventCtx) closes that gap: a defer-like exit that also contributes an item, applied by the handler's flush inside the same batch transaction when the batch lands.

let mut op = ctx.consume_in_batch().await?;
insert_row_in_op(&mut op, n).await?;   // direct op work, per event
Ok(op.collect(summary_update(n)))      // coalesced contribution, applied at flush

Design notes

  • The runner is untouched. Outcome::Collect already lands like Defer, and flush_batch already handles op+items batches — the whole change lives in ctx.rs.
  • No double counting. The exit only bumps tracker.collected; consume_in_batch already counted the event toward max_batch_size. The new test pins this: with max_batch_size == N, a double count would split the burst into two flushes.
  • Not on IsolatedOp — deliberately. A collected item lands with a future batch flush, not with the isolated op's commit, so a collect exit there would silently violate the isolation contract.
  • Compat: BatchOp<'inv>BatchOp<'inv, B = ()>. The defaulted param keeps existing BatchOp<'_> mentions compiling; only code naming BatchOp explicitly inside a non-unit-Batch handler (a combination made possible yesterday in feat(outbox)!: collect/flush statement batching, two-stream runner, sealed op guards #95) could notice. Effectively additive — no !.
  • Replay contract unchanged: on a failed batch the op work rolls back and replayed events re-collect their items.

Also includes a standalone commit fixing two pre-existing clippy lints in tests/post_persist_hook.rs (type_complexity, await_holding_lock) that were blocking nix flake check.

Testing

  • New integration test batch_op_collect_contributes_item_and_defers: each event writes a row directly into the shared batch op and collects an offset item; asserts both land (one batch, one flush, one checkpoint) and that events count once toward max_batch_size.
  • Full nix flake check (fmt, clippy -D warnings, deny, tests) passes; all 31 tests in the two touched suites green.

🤖 Generated with Claude Code

@bodymindarts
bodymindarts marked this pull request as draft August 3, 2026 13:34
@nicolasburtey
nicolasburtey force-pushed the feat/batch-op-collect branch from 7f0af16 to 38ac4ce Compare August 4, 2026 17:51
bodymindarts and others added 4 commits August 5, 2026 11:15
Newer clippy flags the RecordingHook chunks field (type_complexity) and a
MutexGuard held across the final await in
hook_sees_persisted_events_pre_commit (await_holding_lock). Introduce a
RecordedChunks alias and scope the guard so it drops before the await.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Handlers entering via consume_in_batch could only exit with commit or
defer — an event needing both direct op work and a contribution to the
Batch accumulator had to give up one or the other. BatchOp::collect_with
(with Vec/HashMap collect sugar, mirroring EventCtx) closes that gap: a
defer-like exit that also contributes an item, applied by the handler's
flush inside the same batch transaction when the batch lands.

BatchOp becomes generic over the accumulator (BatchOp<'inv, B = ()>);
the defaulted param keeps existing BatchOp<'_> mentions compiling. The
runner is untouched: Outcome::Collect already lands like Defer, and
flush_batch already handles op+items batches. The exit only bumps
tracker.collected — consume_in_batch already counted the event, so
bumping events_in_op again would double-count it against max_batch_size
(pinned by the new test's single-flush assertion).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
… test

ephemeral_never_interrupts_an_open_batch published its ephemeral after a
fixed 150ms sleep, assuming the job was already mid-drain. On a slow CI
runner the job hadn't started consuming yet, so the ephemeral was handled
before the batch opened and snapshotted zero rows (CI failure: left 0,
right 5) — a scheduling artifact, not a violation of the invariant under
test. Record persistent deliveries and publish the ephemeral only once
the second event is being handled: the first event has then provably
deferred into an open op, so the mid-batch arrival is guaranteed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Sibling regression to ephemeral_never_interrupts_an_open_batch for the
other batch-open flag: a slow collect-only handler's batch is pending
purely via the accumulator (collected > 0, no op ever opens until the
flush). Same evidence-gated timing — the ephemeral is published only
once the second delivery proves the first event already collected — and
the same invariant: the ephemeral runs at the batch boundary, after all
N items flushed in one batch.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
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