From 26b4c1b860588500003b24cb4407062e3d4de63c Mon Sep 17 00:00:00 2001 From: bodymindarts Date: Fri, 24 Jul 2026 16:36:14 +0200 Subject: [PATCH 1/4] fix(tests): resolve clippy lints in post_persist_hook test 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 --- tests/post_persist_hook.rs | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/tests/post_persist_hook.rs b/tests/post_persist_hook.rs index 5011ad9..7cf1bb2 100644 --- a/tests/post_persist_hook.rs +++ b/tests/post_persist_hook.rs @@ -74,11 +74,15 @@ fn default_config() -> MailboxConfig { .expect("Couldn't build MailboxConfig") } +/// Chunks recorded by [`RecordingHook`]: per invocation, the (sequence, +/// payload) pairs it saw. +type RecordedChunks = Arc>>>; + /// Records every chunk the hook sees (sequence + payload), plus the row /// counts visible through the op's connection (inside the tx) and through /// the pool (outside the tx) at invocation time. struct RecordingHook { - chunks: Arc>>>, + chunks: RecordedChunks, in_tx_counts: Arc>>, outside_tx_counts: Arc>>, pool: sqlx::PgPool, @@ -229,13 +233,15 @@ async fn hook_sees_persisted_events_pre_commit() -> anyhow::Result<()> { op.commit().await?; - let chunks = chunks.lock().unwrap(); - assert_eq!(chunks.len(), 1, "one chunk for a small publish"); - assert_eq!( - chunks[0], - [(1, SourceEvent::Source(1)), (2, SourceEvent::Source(2))], - "hook sees persisted events with assigned sequences" - ); + { + let chunks = chunks.lock().unwrap(); + assert_eq!(chunks.len(), 1, "one chunk for a small publish"); + assert_eq!( + chunks[0], + [(1, SourceEvent::Source(1)), (2, SourceEvent::Source(2))], + "hook sees persisted events with assigned sequences" + ); + } assert_eq!( in_tx_counts.lock().unwrap().as_slice(), [2], From febe4fb384bd4687e5419ad9c9dcfda044a24140 Mon Sep 17 00:00:00 2001 From: bodymindarts Date: Fri, 24 Jul 2026 16:36:22 +0200 Subject: [PATCH 2/4] feat(outbox): collect exit verb on BatchOp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/out/ctx.rs | 84 +++++++++++++++++++++++++----- src/out/job.rs | 7 +++ tests/outbox_event_handler.rs | 96 +++++++++++++++++++++++++++++++++++ 3 files changed, 175 insertions(+), 12 deletions(-) diff --git a/src/out/ctx.rs b/src/out/ctx.rs index 81e683f..58302cd 100644 --- a/src/out/ctx.rs +++ b/src/out/ctx.rs @@ -12,12 +12,13 @@ //! | [`EventCtx::consume_in_batch`] | do work joined to the pending batch op | shares one transaction (and one checkpoint) with neighboring events | //! | [`EventCtx::consume_isolated`] | land the pending batch first, then do my work in a fresh op | my event is its own atomic unit, fenced from history | //! -//! and — when an op was taken — one of two exit verbs: +//! and — when an op was taken — one of three exit verbs: //! //! | verb | meaning | //! |------|---------| //! | [`BatchOp::commit`] / [`IsolatedOp::commit`] | land the op (work + checkpoint, atomically) when the invocation returns | //! | [`BatchOp::defer`] | leave the op open so subsequent events can coalesce into it | +//! | [`BatchOp::collect_with`] (and the [`collect`](BatchOp::collect) sugar) | [`defer`](BatchOp::defer), plus contribute an item to the pending batch's accumulator | //! //! A pending batch — an open op, collected items, or both — only ever exists //! while there is ready persistent backlog: the runner never awaits the @@ -83,7 +84,9 @@ pub(crate) struct CtxParts<'inv> { /// /// Only obtainable from [`EventCtx::skip`], [`EventCtx::collect_with`] (or /// its [`collect`](EventCtx::collect) sugar), [`BatchOp::commit`], -/// [`BatchOp::defer`] or [`IsolatedOp::commit`] — the type system forces +/// [`BatchOp::defer`], [`BatchOp::collect_with`] (or its +/// [`collect`](BatchOp::collect) sugar) or [`IsolatedOp::commit`] — the +/// type system forces /// every [`handle_persistent`](super::OutboxEventHandler::handle_persistent) /// invocation to decide the transactional fate of its event. /// @@ -176,8 +179,8 @@ impl<'inv, B> EventCtx<'inv, B> { /// committed checkpoint. Only choose this when the handler's work is /// idempotent under such replay (pure in-op DB writes and in-op job /// spawns are — they roll back with the op). - pub async fn consume_in_batch(self) -> Result, HandlerError> { - let parts = self.parts; + pub async fn consume_in_batch(self) -> Result, HandlerError> { + let EventCtx { parts, batch, .. } = self; if parts.op_slot.is_none() { *parts.op_slot = Some( es_entity::DbOp::init_with_clock( @@ -188,7 +191,7 @@ impl<'inv, B> EventCtx<'inv, B> { ); } parts.tracker.events_in_op += 1; - Ok(BatchOp { parts }) + Ok(BatchOp { parts, batch }) } /// Land the pending batch first (its collected items, its work and its @@ -245,19 +248,26 @@ where /// An op joined to the pending batch. Implements /// [`AtomicOperation`](es_entity::AtomicOperation) — use it exactly like any -/// atomic operation, then exit with [`commit`](Self::commit) or -/// [`defer`](Self::defer). +/// atomic operation, then exit with [`commit`](Self::commit), +/// [`defer`](Self::defer) or [`collect_with`](Self::collect_with) (and its +/// [`collect`](Self::collect) sugar). +/// +/// Generic over the handler's +/// [`Batch`](super::OutboxEventHandler::Batch) accumulator `B` (defaulting +/// to `()` for handlers that never collect), so the collect exit can +/// contribute to the same accumulator as [`EventCtx::collect_with`]. /// /// There is no mutable access to the raw [`es_entity::DbOp`] (only a shared /// [`Deref`](std::ops::Deref) view): committing, rolling back, or swapping /// out the underlying op is unrepresentable, so work and checkpoint can only /// land together, through the runner. -#[must_use = "exit with .commit() or .defer() to produce the Handled token"] -pub struct BatchOp<'inv> { +#[must_use = "exit with .commit(), .defer() or .collect_with() to produce the Handled token"] +pub struct BatchOp<'inv, B = ()> { parts: CtxParts<'inv>, + batch: &'inv mut B, } -impl<'inv> BatchOp<'inv> { +impl<'inv, B> BatchOp<'inv, B> { fn op_mut(&mut self) -> &mut es_entity::DbOp<'static> { self.parts .op_slot @@ -284,9 +294,59 @@ impl<'inv> BatchOp<'inv> { _invocation: PhantomData, } } + + /// [`defer`](Self::defer), plus contribute to the pending batch's + /// accumulator — for events that need both direct op work now and a + /// coalesced contribution at flush time. The op stays open for + /// subsequent events, and the item is applied by the handler's + /// [`flush`](super::OutboxEventHandler::flush) inside this same batch + /// transaction when the batch lands. + /// + /// Unlike [`EventCtx::collect_with`] (a pure memory write) this exit + /// rides the op the handler already opened via + /// [`consume_in_batch`](EventCtx::consume_in_batch). The replay + /// contract is unchanged: on a failed batch the op work rolls back and + /// the replayed events re-collect their items. + /// + /// For `Vec` and `HashMap` accumulators the [`collect`](Self::collect) + /// sugar is usually more convenient. + pub fn collect_with(self, f: impl FnOnce(&mut B)) -> Handled<'inv> { + f(self.batch); + // No `events_in_op += 1` here: `consume_in_batch` already counted + // this event — bumping again would double-count it against + // `max_batch_size`. + self.parts.tracker.collected += 1; + Handled { + outcome: Outcome::Collect, + _invocation: PhantomData, + } + } +} + +impl<'inv, T> BatchOp<'inv, Vec> { + /// [`collect_with`](Self::collect_with) sugar for `Vec` accumulators: + /// append one item to the pending batch. + pub fn collect(self, item: T) -> Handled<'inv> { + self.collect_with(|batch| batch.push(item)) + } +} + +impl<'inv, K, V, S> BatchOp<'inv, std::collections::HashMap> +where + K: std::hash::Hash + Eq, + S: std::hash::BuildHasher, +{ + /// [`collect_with`](Self::collect_with) sugar for `HashMap` + /// accumulators: keyed last-write-wins insert — the coalescing fold + /// (see [`EventCtx::collect`](EventCtx::collect)). + pub fn collect(self, key: K, value: V) -> Handled<'inv> { + self.collect_with(|batch| { + batch.insert(key, value); + }) + } } -impl std::ops::Deref for BatchOp<'_> { +impl std::ops::Deref for BatchOp<'_, B> { type Target = es_entity::DbOp<'static>; fn deref(&self) -> &Self::Target { @@ -309,7 +369,7 @@ impl std::ops::Deref for BatchOp<'_> { /// mutable surface: with no `DerefMut`, a `&mut es_entity::DbOp` can never be /// obtained from the guard (which would allow `std::mem::swap`-ing in a decoy /// op and committing the real one without its checkpoint). -impl es_entity::AtomicOperation for BatchOp<'_> { +impl es_entity::AtomicOperation for BatchOp<'_, B> { fn maybe_now(&self) -> Option> { (**self).maybe_now() } diff --git a/src/out/job.rs b/src/out/job.rs index 45a49bd..fbabde3 100644 --- a/src/out/job.rs +++ b/src/out/job.rs @@ -52,6 +52,13 @@ pub enum EventSubscription { /// tolerate whole-batch replay after a mid-batch failure. /// - [`consume_in_batch`](EventCtx::consume_in_batch) then /// [`commit`](BatchOp::commit) — join the batch and close it with me. +/// - [`consume_in_batch`](EventCtx::consume_in_batch) then +/// [`collect_with`](BatchOp::collect_with) / the +/// [`collect`](BatchOp::collect) sugar — both channels for one event: +/// direct work in the shared batch op now, plus an item contributed to +/// the [`Batch`](Self::Batch) accumulator, applied via +/// [`flush`](Self::flush) when the batch lands. Defer-like exit: the op +/// stays open for subsequent events. /// - [`consume_isolated`](EventCtx::consume_isolated) then /// [`commit`](IsolatedOp::commit) — my event is its own atomic unit: the /// pending batch (items + work + checkpoint) lands before my work starts, diff --git a/tests/outbox_event_handler.rs b/tests/outbox_event_handler.rs index 0c0f738..ada5e35 100644 --- a/tests/outbox_event_handler.rs +++ b/tests/outbox_event_handler.rs @@ -423,6 +423,44 @@ impl OutboxEventHandler for MixedHandler { } } +/// Both channels per event: writes one row directly into the shared batch +/// op, then exits via [`obix::BatchOp`]'s `collect` — the defer-like exit +/// that also contributes an item (`n + 100`) to the accumulator. +struct OpWorkThenCollectHandler { + flush_sizes: Arc>>, +} + +impl OutboxEventHandler for OpWorkThenCollectHandler { + type Batch = Vec; + + async fn handle_persistent<'inv>( + &self, + ctx: EventCtx<'inv, Vec>, + event: &obix::out::PersistentOutboxEvent, + ) -> Result, Box> { + let Some(TestEvent::Ping(n)) = &event.payload else { + return Ok(ctx.skip()); + }; + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + let n = *n as i64; + let mut op = ctx.consume_in_batch().await?; + insert_effect_in_op(&mut op, n).await?; + Ok(op.collect(n + 100)) + } + + async fn flush( + &self, + op: &mut obix::FlushOp<'_>, + items: Vec, + ) -> Result<(), Box> { + self.flush_sizes.lock().await.push(items.len()); + for n in items { + insert_effect_in_op(op, n).await?; + } + Ok(()) + } +} + /// Collects everything except one event value, which it handles isolated — /// the isolation fence must land the collected items (and their checkpoint) /// before the isolated op exists, so the isolated failure replays alone. @@ -1411,6 +1449,64 @@ async fn mixed_collect_and_defer_land_in_one_batch() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +#[file_serial] +async fn batch_op_collect_contributes_item_and_defers() -> anyhow::Result<()> { + let pool = init_pool().await?; + reset_batch_effects_table(&pool).await?; + + let job_config = job::JobSvcConfig::builder() + .pool(pool.clone()) + .build() + .unwrap(); + let mut jobs = job::Jobs::init(job_config).await?; + + let flush_sizes = Arc::new(Mutex::new(Vec::new())); + // max_batch_size == N pins single-counting: if the collect exit counted + // its event against the batch size again (on top of consume_in_batch), + // the batch would force-flush halfway through the burst. + const N: u64 = 4; + let config = + OutboxEventJobConfig::new(job::JobType::new(JOB_TYPE)).with_max_batch_size(N as usize); + let outbox = init_outbox_with_handler_config( + &pool, + &mut jobs, + config, + OpWorkThenCollectHandler { + flush_sizes: flush_sizes.clone(), + }, + ) + .await?; + + // Publish before the job starts so the events arrive as ready backlog. + let mut op = outbox.begin_op().await?; + for n in 1..=N { + outbox + .publish_persisted_in_op(&mut op, TestEvent::Ping(n)) + .await?; + } + op.commit().await?; + + jobs.start_poll().await?; + + wait_for_effect_rows(&pool, 2 * N as usize).await?; + wait_for_checkpoint(&pool, N as i64).await?; + + // Each event landed twice: its direct op write at handle time and its + // collected item at flush time — one batch, one transaction, one + // checkpoint for all of it. + assert_eq!( + batch_effect_rows(&pool).await?, + vec![1, 2, 3, 4, 101, 102, 103, 104] + ); + // One flush for the whole burst — each event counted once toward + // max_batch_size (a double count would have split it into two flushes + // of 2). + assert_eq!(*flush_sizes.lock().await, vec![N as usize]); + + Ok(()) +} + #[tokio::test] #[file_serial] async fn isolated_entry_flushes_collected_items_first() -> anyhow::Result<()> { From ffda42ed9a63d1be259e80dc4cbdae6c05d64f61 Mon Sep 17 00:00:00 2001 From: bodymindarts Date: Fri, 24 Jul 2026 16:46:22 +0200 Subject: [PATCH 3/4] fix(tests): deterministic mid-batch trigger in ephemeral interruption test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- tests/outbox_event_handler.rs | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/tests/outbox_event_handler.rs b/tests/outbox_event_handler.rs index ada5e35..16f2659 100644 --- a/tests/outbox_event_handler.rs +++ b/tests/outbox_event_handler.rs @@ -254,9 +254,12 @@ impl OutboxEventHandler for IsolatingEffectHandler { /// Slow deferring worker that also records ephemerals and snapshots the /// committed effect rows when an ephemeral runs — used to prove an /// ephemeral arriving mid-batch never interrupts the batch: it is handled -/// at the batch boundary, after the full batch has landed. +/// at the batch boundary, after the full batch has landed. Records each +/// persistent delivery (before its slow sleep) so the test can publish the +/// ephemeral only once the batch is provably open. struct SlowDeferringHandler { pool: sqlx::PgPool, + persistent_seen: Arc>>, ephemeral_received: Arc>>, rows_at_ephemeral: Arc>, } @@ -273,6 +276,7 @@ impl OutboxEventHandler for SlowDeferringHandler { let Some(TestEvent::Ping(n)) = &event.payload else { return Ok(ctx.skip()); }; + self.persistent_seen.lock().await.push(*n); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let mut op = ctx.consume_in_batch().await?; sqlx::query("INSERT INTO test_batch_effects (n) VALUES ($1)") @@ -1159,6 +1163,7 @@ async fn ephemeral_never_interrupts_an_open_batch() -> anyhow::Result<()> { .unwrap(); let mut jobs = job::Jobs::init(job_config).await?; + let persistent_seen = Arc::new(Mutex::new(Vec::new())); let ephemeral_received = Arc::new(Mutex::new(Vec::new())); let rows_at_ephemeral = Arc::new(Mutex::new(0usize)); let outbox = init_outbox_with_handler( @@ -1166,6 +1171,7 @@ async fn ephemeral_never_interrupts_an_open_batch() -> anyhow::Result<()> { &mut jobs, SlowDeferringHandler { pool: pool.clone(), + persistent_seen: persistent_seen.clone(), ephemeral_received: ephemeral_received.clone(), rows_at_ephemeral: rows_at_ephemeral.clone(), }, @@ -1173,7 +1179,11 @@ async fn ephemeral_never_interrupts_an_open_batch() -> anyhow::Result<()> { .await?; // Publish a backlog of slow (100ms each) deferring events, then publish - // an ephemeral while the batch is guaranteed to still be open mid-drain. + // an ephemeral while the batch is guaranteed to be open mid-drain: the + // second delivery proves the first event already deferred into an open + // op (a fixed sleep here flakes on slow runners — the job may not have + // started consuming yet, and an ephemeral handled before the batch + // opens sees zero rows without violating the invariant under test). const N: u64 = 5; let mut op = outbox.begin_op().await?; for n in 1..=N { @@ -1185,7 +1195,7 @@ async fn ephemeral_never_interrupts_an_open_batch() -> anyhow::Result<()> { jobs.start_poll().await?; - tokio::time::sleep(std::time::Duration::from_millis(150)).await; + wait_for_n_deliveries(&persistent_seen, 2, std::time::Duration::from_secs(10)).await?; outbox .publish_ephemeral( obix::out::EphemeralEventType::new("mid_batch"), From 9f62ca9df1a6068fc0b9d1675e1bdba412cffdb3 Mon Sep 17 00:00:00 2001 From: bodymindarts Date: Fri, 24 Jul 2026 16:54:04 +0200 Subject: [PATCH 4/4] test: ephemeral cannot interrupt a collect-only batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- tests/outbox_event_handler.rs | 118 ++++++++++++++++++++++++++++++++++ 1 file changed, 118 insertions(+) diff --git a/tests/outbox_event_handler.rs b/tests/outbox_event_handler.rs index 16f2659..5a765dd 100644 --- a/tests/outbox_event_handler.rs +++ b/tests/outbox_event_handler.rs @@ -177,6 +177,58 @@ async fn insert_effect_in_op( Ok(()) } +/// The collect-path sibling of [`SlowDeferringHandler`]: a slow worker whose +/// batch is open purely through the accumulator (`collected > 0`, no op) — +/// proving an ephemeral cannot interrupt a collect-only batch either. +struct SlowCollectingHandler { + pool: sqlx::PgPool, + persistent_seen: Arc>>, + ephemeral_received: Arc>>, + rows_at_ephemeral: Arc>, +} + +impl OutboxEventHandler for SlowCollectingHandler { + type Batch = Vec; + + async fn handle_persistent<'inv>( + &self, + ctx: EventCtx<'inv, Vec>, + event: &obix::out::PersistentOutboxEvent, + ) -> Result, Box> { + let Some(TestEvent::Ping(n)) = &event.payload else { + return Ok(ctx.skip()); + }; + self.persistent_seen.lock().await.push(*n); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + Ok(ctx.collect(*n as i64)) + } + + async fn flush( + &self, + op: &mut obix::FlushOp<'_>, + items: Vec, + ) -> Result<(), Box> { + for n in items { + insert_effect_in_op(op, n).await?; + } + Ok(()) + } + + async fn handle_ephemeral( + &self, + event: &obix::out::EphemeralOutboxEvent, + ) -> Result<(), Box> { + let TestEvent::Ping(n) = &event.payload; + self.ephemeral_received.lock().await.push(*n); + let rows = batch_effect_rows(&self.pool) + .await + .expect("read effect rows") + .len(); + *self.rows_at_ephemeral.lock().await = rows; + Ok(()) + } +} + /// Batch-safe worker: inserts one row per event inside the shared batch op /// and defers. Optionally fails the first time a given event value is seen. /// Sleeps briefly per event so the backfill keeps the ready backlog ahead of @@ -1221,6 +1273,72 @@ async fn ephemeral_never_interrupts_an_open_batch() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +#[file_serial] +async fn ephemeral_never_interrupts_a_collect_only_batch() -> anyhow::Result<()> { + let pool = init_pool().await?; + reset_batch_effects_table(&pool).await?; + + let job_config = job::JobSvcConfig::builder() + .pool(pool.clone()) + .build() + .unwrap(); + let mut jobs = job::Jobs::init(job_config).await?; + + let persistent_seen = Arc::new(Mutex::new(Vec::new())); + let ephemeral_received = Arc::new(Mutex::new(Vec::new())); + let rows_at_ephemeral = Arc::new(Mutex::new(0usize)); + let outbox = init_outbox_with_handler( + &pool, + &mut jobs, + SlowCollectingHandler { + pool: pool.clone(), + persistent_seen: persistent_seen.clone(), + ephemeral_received: ephemeral_received.clone(), + rows_at_ephemeral: rows_at_ephemeral.clone(), + }, + ) + .await?; + + const N: u64 = 5; + let mut op = outbox.begin_op().await?; + for n in 1..=N { + outbox + .publish_persisted_in_op(&mut op, TestEvent::Ping(n)) + .await?; + } + op.commit().await?; + + jobs.start_poll().await?; + + // Same evidence-gated timing as the deferring variant: the second + // delivery proves the first event already collected, so the batch is + // open (via `collected > 0` — no op exists on this path) when the + // ephemeral arrives. + wait_for_n_deliveries(&persistent_seen, 2, std::time::Duration::from_secs(10)).await?; + outbox + .publish_ephemeral( + obix::out::EphemeralEventType::new("mid_collect"), + TestEvent::Ping(9), + ) + .await?; + + wait_for_effect_rows(&pool, N as usize).await?; + wait_for_n_deliveries(&ephemeral_received, 1, std::time::Duration::from_secs(5)).await?; + + // A collect-only batch holds no op — it is open purely through the + // accumulator — and the ephemeral still cannot interrupt it: all N + // items flushed in one batch before the ephemeral handler ran. + let rows_at_ephemeral = *rows_at_ephemeral.lock().await; + assert_eq!( + rows_at_ephemeral, N as usize, + "expected the ephemeral to run after the collect-only batch flushed" + ); + assert_eq!(batch_effect_rows(&pool).await?, vec![1, 2, 3, 4, 5]); + + Ok(()) +} + #[tokio::test] #[file_serial] async fn single_deferred_event_commits_promptly_at_low_traffic() -> anyhow::Result<()> {