feat: cancellable OutboxDrainRunner

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.

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<B>(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:<pid>` so claims never collide with the immediate
+    /// hook's `microsvc-immediate:<pid>`. 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<S, P>(
+        store: S,
+        publisher: P,
+        max_attempts: u32,
+    ) -> crate::OutboxDrainRunner<S, P>
+    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,

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<S, B>(
     store: S,
     bus: Arc<B>,
@@ -20,27 +25,19 @@ pub fn spawn_outbox_publish_loop<S, B>(
     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.