From 395701fc3b51b1ca5adaee6242871932dff349bc Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 17 Aug 2026 05:12:22 -0500 Subject: [PATCH] feat: add cancellable OutboxDrainRunner Immediate after-commit publish stays the fast path. This loop is the safety net: dispatch_batch until the outbox is empty, sleep when idle, back off on store errors, stop on abort. Worker id is drain: so claims never collide with microsvc-immediate:. Service::outbox_drain builds the runner; spawn_outbox_publish_loop now wraps it. Dropping the handle leaves the task running; stop() aborts. Implements [[tasks/outbox-background-drain-runner]] --- src/lib.rs | 11 + src/microsvc/runtime.rs | 75 ++++++- src/microsvc/workers.rs | 43 ++-- src/outbox_worker/drain.rs | 405 +++++++++++++++++++++++++++++++++++++ src/outbox_worker/mod.rs | 28 ++- 5 files changed, 532 insertions(+), 30 deletions(-) create mode 100644 src/outbox_worker/drain.rs diff --git a/src/lib.rs b/src/lib.rs index 65d356ee..d14501fa 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -354,6 +354,17 @@ pub use outbox_worker::{ BusOutboxPublishHook, BusPublisher, ClaimOutboxMessages, OutboxClaimRef, OutboxDispatchOutcome, OutboxDispatcher, OutboxPublishFailureAction, OutboxSource, OutboxStore, ReceivedOutboxMessage, }; +#[cfg(any( + feature = "http", + feature = "grpc", + feature = "postgres", + feature = "sqlite", + feature = "nats", + feature = "rabbitmq", + feature = "kafka", + test, +))] +pub use outbox_worker::{drain_worker_id, OutboxDrainHandle, OutboxDrainRunner}; pub use queued_repo::{ // WithOpts + unlock traits for the queued repository variant. diff --git a/src/microsvc/runtime.rs b/src/microsvc/runtime.rs index 9b2df1d9..edfb8618 100644 --- a/src/microsvc/runtime.rs +++ b/src/microsvc/runtime.rs @@ -37,11 +37,10 @@ impl Service { /// registered command names (competing) and subscribes to the event names /// (fan-out). /// - /// `with_bus` and [`run`](Self::run) do not start an [`OutboxDispatcher`] - /// polling loop. Deploy one separately (in this process or another) when - /// durable recovery after a commit-to-publish process crash is required. - /// - /// [`OutboxDispatcher`]: crate::outbox_worker::OutboxDispatcher + /// `with_bus` and [`run`](Self::run) do not start the drain loop. Spawn + /// [`crate::OutboxDrainRunner`] (or `microsvc::spawn_outbox_publish_loop`) + /// next to `run` when durable recovery after a commit-to-publish crash is + /// required. pub fn with_bus(mut self, bus: B) -> Self where B: Bus + BusConsumer + 'static, @@ -86,6 +85,33 @@ impl Service { .map_err(TransportError::from)?; runner(Arc::new(self), options).await } + + /// Build a drain runner over `store` that publishes through `publisher`. + /// + /// Worker id is `drain:` so claims never collide with the immediate + /// hook's `microsvc-immediate:`. Spawn it next to [`run`](Self::run); + /// this does not start a task. + #[cfg(any( + feature = "http", + feature = "grpc", + feature = "postgres", + feature = "sqlite", + feature = "nats", + feature = "rabbitmq", + feature = "kafka", + test, + ))] + pub fn outbox_drain( + store: S, + publisher: P, + max_attempts: u32, + ) -> crate::OutboxDrainRunner + where + S: crate::OutboxStore, + P: crate::bus::MessagePublisher, + { + crate::OutboxDrainRunner::for_store(store, publisher, max_attempts) + } } /// A running transport consumer (a `listen` or `subscribe` loop), borrowing the @@ -142,7 +168,7 @@ mod tests { use crate::outbox_worker::OutboxStore; use crate::{ sourced, AggregateBuilder, AggregateRepository, Entity, InMemoryRepository, OutboxMessage, - OutboxMessageStatus, Queueable, QueuedRepository, Snapshot, + OutboxMessageStatus, Queueable, QueuedRepository, Snapshot, TransactionalCommit, }; #[derive(Default)] @@ -324,6 +350,43 @@ mod tests { ); } + #[tokio::test] + async fn outbox_drain_sugar_publishes_rows_the_hook_never_saw() { + let repo = InMemoryRepository::new(); + let store = repo.outbox_store(); + let mut batch = crate::CommitBatch::empty(); + batch + .outbox_messages + .push(OutboxMessage::create("evt-left", "dummy.touched", b"{}".to_vec()).unwrap()); + repo.commit_batch(batch).await.unwrap(); + + let handle = Service::outbox_drain( + store, + crate::BusPublisher::new(std::sync::Arc::new(InMemoryBus::new())), + 5, + ) + .with_poll_interval(std::time::Duration::from_millis(5)) + .spawn(); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if repo + .outbox_store() + .messages_by_status(OutboxMessageStatus::Published, 8) + .await + .unwrap() + .len() + == 1 + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .expect("Service::outbox_drain should publish leftover rows"); + handle.stop().await.unwrap(); + } + #[derive(Default, Snapshot)] struct SnapCounter { entity: Entity, diff --git a/src/microsvc/workers.rs b/src/microsvc/workers.rs index 532337f7..3a4b5a98 100644 --- a/src/microsvc/workers.rs +++ b/src/microsvc/workers.rs @@ -7,9 +7,14 @@ use std::time::Duration; use crate::bus::{Bus, RunOptions}; use crate::microsvc::Service; -use crate::outbox_worker::{BusPublisher, OutboxDispatcher, OutboxStore}; +use crate::outbox_worker::{ + drain_worker_id, BusPublisher, OutboxDispatcher, OutboxDrainRunner, OutboxStore, +}; -/// Spawn the standard outbox publish loop for a bus-backed store. +/// Spawn the standard outbox drain loop for a bus-backed store. +/// +/// Fire-and-forget: the task lives until the process exits. Prefer +/// [`OutboxDrainRunner`] when the caller needs to stop the loop. pub fn spawn_outbox_publish_loop( store: S, bus: Arc, @@ -20,27 +25,19 @@ pub fn spawn_outbox_publish_loop( S: OutboxStore + 'static, B: Bus + Send + Sync + 'static, { - let service_name = service_name.into(); - tokio::spawn(async move { - let dispatcher = OutboxDispatcher::new( - store, - BusPublisher::new(bus), - format!("outbox:{}", std::process::id()), - lease, - max_attempts, - ) - .with_service(service_name); - loop { - match dispatcher.dispatch_batch(32).await { - Ok(o) if o.published > 0 || o.claimed > 0 => {} - Ok(_) => tokio::time::sleep(Duration::from_millis(25)).await, - Err(e) => { - eprintln!("outbox: {e}"); - tokio::time::sleep(Duration::from_millis(100)).await; - } - } - } - }); + let dispatcher = OutboxDispatcher::new( + store, + BusPublisher::new(bus), + drain_worker_id(), + lease, + max_attempts, + ) + .with_service(service_name); + let _handle = OutboxDrainRunner::new(dispatcher) + .with_batch_size(32) + .with_poll_interval(Duration::from_millis(25)) + .with_error_backoff(Duration::from_millis(100)) + .spawn(); } /// Spawn a service consumer loop that re-runs the bus handler continuously. diff --git a/src/outbox_worker/drain.rs b/src/outbox_worker/drain.rs new file mode 100644 index 00000000..e8c19705 --- /dev/null +++ b/src/outbox_worker/drain.rs @@ -0,0 +1,405 @@ +//! Cancellable background drain over [`OutboxDispatcher::dispatch_batch`]. +//! +//! Immediate after-commit publish stays the fast path. This loop is the +//! safety net: it only claims rows that are still pending (released after a +//! publish failure, or never claimed because the process crashed after +//! commit). It does not resurrect the old in-memory `OutboxWorker`. + +use std::future::Future; +use std::time::Duration; + +use tokio::task::JoinHandle; + +use crate::bus::{MessagePublisher, TransportError}; + +use super::{OutboxDispatchOutcome, OutboxDispatcher, OutboxStore}; + +/// Default time between empty drain passes. +pub const DEFAULT_DRAIN_POLL_INTERVAL: Duration = Duration::from_secs(1); + +/// Default claim batch for one drain pass. +pub const DEFAULT_DRAIN_BATCH_SIZE: usize = 32; + +/// Default lease for drain claims. Longer than the immediate hook's 5s lease +/// so a pass can publish a real backlog without the row becoming claimable +/// by a second drainer. +pub const DEFAULT_DRAIN_LEASE: Duration = Duration::from_secs(30); + +/// Default first backoff after a store error. Doubles up to +/// [`DEFAULT_DRAIN_MAX_ERROR_BACKOFF`]. +pub const DEFAULT_DRAIN_ERROR_BACKOFF: Duration = Duration::from_secs(1); + +/// Ceiling for store-error backoff. +pub const DEFAULT_DRAIN_MAX_ERROR_BACKOFF: Duration = Duration::from_secs(30); + +/// Worker-id prefix that distinguishes drain claims from +/// `microsvc-immediate:`. +pub fn drain_worker_id() -> String { + format!("drain:{}", std::process::id()) +} + +/// Repeatedly [`OutboxDispatcher::dispatch_batch`] until cancelled. +pub struct OutboxDrainRunner { + dispatcher: OutboxDispatcher, + batch_size: usize, + poll_interval: Duration, + error_backoff: Duration, + max_error_backoff: Duration, +} + +impl OutboxDrainRunner +where + S: OutboxStore, + P: MessagePublisher, +{ + /// Wrap an existing dispatcher. Reuse the dispatcher's lease / attempts / + /// concurrency rather than duplicating them here. + pub fn new(dispatcher: OutboxDispatcher) -> Self { + Self { + dispatcher, + batch_size: DEFAULT_DRAIN_BATCH_SIZE, + poll_interval: DEFAULT_DRAIN_POLL_INTERVAL, + error_backoff: DEFAULT_DRAIN_ERROR_BACKOFF, + max_error_backoff: DEFAULT_DRAIN_MAX_ERROR_BACKOFF, + } + } + + /// Build a dispatcher + runner with the drain worker id and default lease. + pub fn for_store(store: S, publisher: P, max_attempts: u32) -> Self { + Self::new(OutboxDispatcher::new( + store, + publisher, + drain_worker_id(), + DEFAULT_DRAIN_LEASE, + max_attempts, + )) + } + + pub fn with_batch_size(mut self, batch_size: usize) -> Self { + self.batch_size = batch_size.max(1); + self + } + + pub fn with_poll_interval(mut self, poll_interval: Duration) -> Self { + self.poll_interval = poll_interval; + self + } + + pub fn with_error_backoff(mut self, error_backoff: Duration) -> Self { + self.error_backoff = error_backoff; + self + } + + pub fn with_max_error_backoff(mut self, max_error_backoff: Duration) -> Self { + self.max_error_backoff = max_error_backoff; + self + } + + /// The dispatcher this runner drives. + pub fn dispatcher(&self) -> &OutboxDispatcher { + &self.dispatcher + } + + /// Drain until `shutdown` resolves. Store errors back off; they do not + /// terminate the loop. Empty passes sleep `poll_interval`. A full batch + /// is followed immediately by another pass. + pub async fn run(self, shutdown: impl Future) -> Result<(), TransportError> { + tokio::pin!(shutdown); + let mut backoff = self.error_backoff; + loop { + let pass = async { + match self.dispatcher.dispatch_batch(self.batch_size).await { + Ok(outcome) => Pass::Drained(outcome), + Err(error) => Pass::StoreError(error), + } + }; + tokio::select! { + _ = &mut shutdown => return Ok(()), + result = pass => match result { + Pass::Drained(outcome) => { + backoff = self.error_backoff; + if outcome.claimed < self.batch_size { + tokio::select! { + _ = &mut shutdown => return Ok(()), + _ = tokio::time::sleep(self.poll_interval) => {} + } + } + } + Pass::StoreError(error) => { + eprintln!("outbox drain: {error}"); + tokio::select! { + _ = &mut shutdown => return Ok(()), + _ = tokio::time::sleep(backoff) => {} + } + backoff = backoff.saturating_mul(2).min(self.max_error_backoff); + } + } + } + } + } + + /// Spawn on the current tokio runtime. Dropping the handle does **not** + /// stop the task (call [`OutboxDrainHandle::stop`]). + pub fn spawn(self) -> OutboxDrainHandle + where + S: 'static, + P: 'static, + { + let join = tokio::spawn(async move { self.run(std::future::pending()).await }); + OutboxDrainHandle { join } + } +} + +enum Pass { + Drained(OutboxDispatchOutcome), + StoreError(TransportError), +} + +/// Handle for a spawned [`OutboxDrainRunner`]. Abort-only on [`stop`]; drop +/// leaves the task running so fire-and-forget hosts stay drained. +pub struct OutboxDrainHandle { + join: JoinHandle>, +} + +impl OutboxDrainHandle { + /// Abort the loop and wait for it to unwind. + pub async fn stop(self) -> Result<(), TransportError> { + self.join.abort(); + match self.join.await { + Ok(result) => result, + Err(error) if error.is_cancelled() => Ok(()), + Err(error) => Err(TransportError::retryable(format!( + "outbox drain task: {error}" + ))), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bus::Message; + use crate::outbox_worker::{ClaimOutboxMessages, OutboxClaimRef, OutboxStore}; + use crate::repository::RepositoryError; + use crate::{ + CommitBatch, InMemoryOutboxStore, InMemoryRepository, OutboxMessage, OutboxMessageStatus, + TransactionalCommit, + }; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + struct RecordingPublisher { + published: Mutex>, + } + + impl RecordingPublisher { + fn new() -> Arc { + Arc::new(Self { + published: Mutex::new(Vec::new()), + }) + } + fn ids(&self) -> Vec { + self.published.lock().unwrap().clone() + } + } + + impl MessagePublisher for Arc { + async fn publish(&self, message: Message) -> Result<(), TransportError> { + self.published + .lock() + .unwrap() + .push(message.id().unwrap_or_default().to_string()); + Ok(()) + } + } + + fn outbox(id: &str) -> OutboxMessage { + OutboxMessage::create(id, "OrderCreated", b"{}".to_vec()).unwrap() + } + + fn store_message(repo: &InMemoryRepository, message: OutboxMessage) -> String { + let id = message.id().to_string(); + let mut batch = CommitBatch::empty(); + batch.outbox_messages.push(message); + futures_executor_block(repo.commit_batch(batch)).unwrap(); + id + } + + fn futures_executor_block(future: F) -> F::Output { + // commit_batch is runtime-free; keep that path off the tokio worker. + crate::outbox_worker::testing::block_on(future) + } + + fn load(repo: &InMemoryRepository, id: &str) -> OutboxMessage { + repo.outbox_storage() + .read() + .unwrap() + .get(id) + .unwrap() + .clone() + } + + #[tokio::test] + async fn immediate_publish_leaves_nothing_for_the_drainer() { + let repo = InMemoryRepository::new(); + let id = store_message(&repo, outbox("evt-1")); + let publisher = RecordingPublisher::new(); + let immediate = OutboxDispatcher::new( + repo.outbox_store(), + publisher.clone(), + "immediate:test", + Duration::from_secs(60), + 3, + ); + let outcome = immediate + .dispatch_ids(std::slice::from_ref(&id)) + .await + .unwrap(); + assert_eq!(outcome.published, 1); + assert_eq!(load(&repo, &id).status, OutboxMessageStatus::Published); + + let drain = OutboxDrainRunner::new(OutboxDispatcher::new( + repo.outbox_store(), + publisher.clone(), + drain_worker_id(), + Duration::from_secs(60), + 3, + )) + .with_batch_size(8) + .with_poll_interval(Duration::from_millis(5)); + let handle = drain.spawn(); + tokio::time::sleep(Duration::from_millis(30)).await; + handle.stop().await.unwrap(); + + assert_eq!(publisher.ids(), vec!["evt-1".to_string()]); + assert!(repo.outbox_store().pending(8).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn drain_publishes_rows_the_immediate_path_never_attempted() { + let repo = InMemoryRepository::new(); + store_message(&repo, outbox("evt-crash")); + let publisher = RecordingPublisher::new(); + let drain = OutboxDrainRunner::for_store(repo.outbox_store(), publisher.clone(), 3) + .with_batch_size(8) + .with_poll_interval(Duration::from_millis(5)); + let handle = drain.spawn(); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if publisher.ids() == ["evt-crash"] { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("drain should publish the unclaimed row"); + handle.stop().await.unwrap(); + assert_eq!( + load(&repo, "evt-crash").status, + OutboxMessageStatus::Published + ); + } + + #[tokio::test] + async fn store_error_is_retried_and_the_task_survives() { + let repo = InMemoryRepository::new(); + store_message(&repo, outbox("evt-retry")); + let publisher = RecordingPublisher::new(); + let store = FlakyClaimStore { + inner: repo.outbox_store(), + fail_remaining: AtomicU32::new(2), + claims: AtomicU32::new(0), + }; + let drain = OutboxDrainRunner::for_store(store, publisher.clone(), 3) + .with_batch_size(8) + .with_poll_interval(Duration::from_millis(5)) + .with_error_backoff(Duration::from_millis(5)) + .with_max_error_backoff(Duration::from_millis(20)); + let handle = drain.spawn(); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if publisher.ids() == ["evt-retry"] { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("drain should recover after store errors"); + handle.stop().await.unwrap(); + } + + #[tokio::test] + async fn stop_returns_promptly() { + let repo = InMemoryRepository::new(); + let publisher = RecordingPublisher::new(); + let drain = OutboxDrainRunner::for_store(repo.outbox_store(), publisher, 3) + .with_poll_interval(Duration::from_secs(30)); + let handle = drain.spawn(); + let started = std::time::Instant::now(); + handle.stop().await.unwrap(); + assert!( + started.elapsed() < Duration::from_millis(500), + "stop should abort the idle sleep" + ); + } + + struct FlakyClaimStore { + inner: InMemoryOutboxStore, + fail_remaining: AtomicU32, + claims: AtomicU32, + } + + impl OutboxStore for FlakyClaimStore { + fn messages_by_status( + &self, + status: OutboxMessageStatus, + limit: usize, + ) -> impl Future, RepositoryError>> + Send + '_ { + self.inner.messages_by_status(status, limit) + } + + fn claim<'a>( + &'a self, + request: ClaimOutboxMessages, + ) -> impl Future, RepositoryError>> + Send + 'a { + async move { + if self.fail_remaining.load(Ordering::SeqCst) > 0 { + self.fail_remaining.fetch_sub(1, Ordering::SeqCst); + return Err(RepositoryError::Storage { + operation: "injected claim failure".into(), + retryable: true, + source: None, + }); + } + self.claims.fetch_add(1, Ordering::SeqCst); + self.inner.claim(request).await + } + } + + fn complete<'a>( + &'a self, + claim: &'a OutboxClaimRef, + ) -> impl Future> + Send + 'a { + self.inner.complete(claim) + } + + fn release<'a>( + &'a self, + claim: &'a OutboxClaimRef, + error: &'a str, + ) -> impl Future> + Send + 'a { + self.inner.release(claim, error) + } + + fn fail<'a>( + &'a self, + claim: &'a OutboxClaimRef, + error: &'a str, + ) -> impl Future> + Send + 'a { + self.inner.fail(claim, error) + } + } +} diff --git a/src/outbox_worker/mod.rs b/src/outbox_worker/mod.rs index b00c4c28..dae637da 100644 --- a/src/outbox_worker/mod.rs +++ b/src/outbox_worker/mod.rs @@ -26,12 +26,23 @@ //! ``` mod bus_publisher; +#[cfg(any( + feature = "http", + feature = "grpc", + feature = "postgres", + feature = "sqlite", + feature = "nats", + feature = "rabbitmq", + feature = "kafka", + test, +))] +mod drain; mod outbox_dispatch; mod outbox_source; mod publish_hook; mod store; #[cfg(test)] -mod testing; +pub(crate) mod testing; // Repository helpers #[cfg(any(feature = "postgres", feature = "sqlite"))] @@ -43,6 +54,21 @@ pub use store::{ // Outbox -> bus bridge (moved out of the bus module; depends up on bus traits). pub use bus_publisher::BusPublisher; +#[cfg(any( + feature = "http", + feature = "grpc", + feature = "postgres", + feature = "sqlite", + feature = "nats", + feature = "rabbitmq", + feature = "kafka", + test, +))] +pub use drain::{ + drain_worker_id, OutboxDrainHandle, OutboxDrainRunner, DEFAULT_DRAIN_BATCH_SIZE, + DEFAULT_DRAIN_ERROR_BACKOFF, DEFAULT_DRAIN_LEASE, DEFAULT_DRAIN_MAX_ERROR_BACKOFF, + DEFAULT_DRAIN_POLL_INTERVAL, +}; pub use outbox_dispatch::{OutboxDispatchOutcome, OutboxDispatcher, SOURCED_METADATA_PREFIX}; pub use outbox_source::{ OutboxSource, ReceivedOutboxMessage, DEFAULT_OUTBOX_SOURCE_BATCH, DEFAULT_OUTBOX_SOURCE_LEASE,