From 0bf44afe2fe672ed2d5ff3b5fe3c27becc12dabf Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Wed, 20 May 2026 20:59:49 -0500 Subject: [PATCH 1/2] fix: harden outbox claim leases Implements [[tasks/durable-outbox-claim-leases]] --- README.md | 11 + src/lib.rs | 1 + src/outbox/message.rs | 49 ++- src/outbox/mod.rs | 5 + src/outbox_worker/mod.rs | 4 +- src/outbox_worker/repository_ext.rs | 444 +++++++++++++++++++++++----- src/outbox_worker/thread.rs | 69 ++++- src/repository/error.rs | 18 ++ 8 files changed, 521 insertions(+), 80 deletions(-) diff --git a/README.md b/README.md index 37edda5f..adbd67e7 100644 --- a/README.md +++ b/README.md @@ -580,6 +580,11 @@ let repo = HashMapRepository::new() Each outbox message is its own aggregate, committed alongside your domain entity: +Outbox messages are explicit publication records. Aggregate event records are +write-side replay history; they become domain events, integration events, +commands, or transport messages only when application code creates an +`OutboxMessage` for that purpose. + ```rust use sourced_rust::{OutboxCommitExt, OutboxMessage}; @@ -624,6 +629,12 @@ for message in &mut claimed { } ``` +Claims use leases. Pending messages and expired in-flight messages can be +claimed by workers, while unexpired in-flight messages are skipped so competing +workers do not publish the same message concurrently. Repository-backed workers +should record publish failures with a retry ceiling; exhausted messages move to +`Failed` instead of being released forever. + ## Service Bus The service bus supports two messaging patterns: diff --git a/src/lib.rs b/src/lib.rs index 6eaf0166..18dc74e6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,6 +56,7 @@ pub use outbox_worker::{ // Publishers LogPublisher, LogPublisherError, + OutboxPublishFailureAction, OutboxPublisher, // Repository extension for claiming/completing messages OutboxRepositoryExt, diff --git a/src/outbox/message.rs b/src/outbox/message.rs index 03baa825..bb1696c8 100644 --- a/src/outbox/message.rs +++ b/src/outbox/message.rs @@ -226,6 +226,22 @@ impl OutboxMessage { self.status == OutboxMessageStatus::Failed } + pub fn has_expired_lease_at(&self, now: SystemTime) -> bool { + self.is_in_flight() && self.leased_until.map(|until| until <= now).unwrap_or(true) + } + + pub fn is_claimable_at(&self, now: SystemTime) -> bool { + self.is_pending() || self.has_expired_lease_at(now) + } + + pub fn is_claimed_by(&self, worker_id: &str) -> bool { + self.worker_id.as_deref() == Some(worker_id) + } + + fn is_claimable(&self) -> bool { + self.is_claimable_at(SystemTime::now()) + } + // Commands #[digest("MessageCreated")] pub fn initialize( @@ -246,7 +262,7 @@ impl OutboxMessage { self.created_at = SystemTime::now(); } - #[digest("MessageClaimed", when = self.is_pending())] + #[digest("MessageClaimed", when = self.is_claimable())] pub fn claim(&mut self, worker_id: String, until_secs: u64) { let until_time = SystemTime::UNIX_EPOCH + Duration::from_secs(until_secs); self.status = OutboxMessageStatus::InFlight; @@ -257,7 +273,18 @@ impl OutboxMessage { /// Claim with a Duration (convenience method that computes until_secs) pub fn claim_for(&mut self, worker_id: impl Into, lease: Duration) -> SourcedResult { - let until_secs = Self::lease_deadline_secs(SystemTime::now(), lease)?; + self.claim_at(worker_id, lease, SystemTime::now()) + } + + /// Claim with an explicit clock value. This is useful for deterministic + /// tests and repository implementations that capture time once per batch. + pub fn claim_at( + &mut self, + worker_id: impl Into, + lease: Duration, + now: SystemTime, + ) -> SourcedResult { + let until_secs = Self::lease_deadline_secs(now, lease)?; self.claim(worker_id.into(), until_secs) } @@ -406,6 +433,24 @@ mod tests { assert!(message.is_published()); } + #[test] + fn expired_in_flight_message_can_be_claimed_again() { + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + message + .claim_at("worker-1", Duration::from_secs(1), SystemTime::UNIX_EPOCH) + .unwrap(); + + assert!(message.has_expired_lease_at(SystemTime::now())); + assert!(message.is_claimable_at(SystemTime::now())); + + message + .claim_for("worker-2", Duration::from_secs(60)) + .unwrap(); + + assert_eq!(message.worker_id.as_deref(), Some("worker-2")); + assert_eq!(message.attempts, 2); + } + #[test] fn claim_deadline_overflow_returns_error() { let err = OutboxMessage::lease_deadline_secs( diff --git a/src/outbox/mod.rs b/src/outbox/mod.rs index 481edfe1..5789022d 100644 --- a/src/outbox/mod.rs +++ b/src/outbox/mod.rs @@ -17,6 +17,11 @@ //! 1. **Commit phase** (this module) - Atomically commit aggregate event records + outbox message //! 2. **Worker phase** (see `outbox_worker` module) - Drain outbox and publish messages to external systems //! +//! Outbox messages are explicit publication records. Aggregate event records are +//! replayable write-side history; they do not automatically become domain or +//! integration events until application code creates an `OutboxMessage` for that +//! publication. +//! //! ## Example //! //! ```ignore diff --git a/src/outbox_worker/mod.rs b/src/outbox_worker/mod.rs index 2598bc82..80ac4ea4 100644 --- a/src/outbox_worker/mod.rs +++ b/src/outbox_worker/mod.rs @@ -25,7 +25,7 @@ //! let mut worker = OutboxWorker::new(LogPublisher::default()); //! for msg in messages { //! worker.process_message(&mut msg); -//! repo.complete_outbox_message(msg.id())?; +//! repo.complete_outbox_message_for_worker(msg.id(), "worker-1")?; //! } //! ``` @@ -41,7 +41,7 @@ pub use publisher::LocalEmitterPublisher; pub use publisher::{LogPublisher, LogPublisherError, OutboxPublisher}; // Repository helpers -pub use repository_ext::OutboxRepositoryExt; +pub use repository_ext::{OutboxPublishFailureAction, OutboxRepositoryExt}; // Worker pub use worker::{DrainResult, OutboxWorker, ProcessOneResult}; diff --git a/src/outbox_worker/repository_ext.rs b/src/outbox_worker/repository_ext.rs index 72412586..8c1d64d2 100644 --- a/src/outbox_worker/repository_ext.rs +++ b/src/outbox_worker/repository_ext.rs @@ -1,11 +1,17 @@ -use std::time::Duration; +use std::time::{Duration, SystemTime}; use crate::aggregate::hydrate; -use crate::entity::Entity; +use crate::entity::{Entity, EventRecord}; use crate::hashmap_repo::HashMapRepository; use crate::outbox::{OutboxMessage, OutboxMessageStatus}; use crate::repository::RepositoryError; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OutboxPublishFailureAction { + Released, + Failed, +} + /// Extension trait for repositories that expose outbox message operations. pub trait OutboxRepositoryExt: Send + Sync { /// Return all outbox messages with the given status. @@ -30,11 +36,123 @@ pub trait OutboxRepositoryExt: Send + Sync { /// Mark an outbox message as completed (published). fn complete_outbox_message(&self, message_id: &str) -> Result<(), RepositoryError>; + /// Mark an outbox message as completed if it is still claimed by this worker. + fn complete_outbox_message_for_worker( + &self, + message_id: &str, + worker_id: &str, + ) -> Result<(), RepositoryError>; + /// Release an outbox message back to pending (for retry). fn release_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError>; + /// Release an outbox message if it is still claimed by this worker. + fn release_outbox_message_for_worker( + &self, + message_id: &str, + worker_id: &str, + error: &str, + ) -> Result<(), RepositoryError>; + /// Mark an outbox message as permanently failed. fn fail_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError>; + + /// Record a publish failure for a claimed message, releasing it for retry + /// or permanently failing it when the attempt ceiling is reached. + fn record_outbox_publish_failure( + &self, + message_id: &str, + worker_id: &str, + error: &str, + max_attempts: u32, + ) -> Result; +} + +fn normalize_outbox_id(message_id: &str) -> String { + if message_id.starts_with(OutboxMessage::ID_PREFIX) { + message_id.to_string() + } else { + format!("{}{}", OutboxMessage::ID_PREFIX, message_id) + } +} + +fn hydrate_outbox_message( + normalized_id: &str, + events: &[EventRecord], +) -> Result { + let mut entity = Entity::with_id(normalized_id.to_string()); + entity.load_from_history(events.to_vec()); + hydrate::(entity) +} + +fn persist_outbox_message(events: &mut Vec, message: &mut OutboxMessage) { + *events = message.entity.events().to_vec(); + message.entity.mark_committed(); +} + +fn outbox_state(message: &OutboxMessage) -> String { + format!( + "{:?}, worker={:?}, leased_until={:?}, attempts={}", + message.status, message.worker_id, message.leased_until, message.attempts + ) +} + +fn invalid_outbox_state(message: &OutboxMessage, expected: &'static str) -> RepositoryError { + RepositoryError::InvalidState { + id: message.id().to_string(), + expected, + actual: outbox_state(message), + } +} + +fn ensure_active_claim( + message: &OutboxMessage, + worker_id: Option<&str>, + now: SystemTime, +) -> Result<(), RepositoryError> { + if !message.is_in_flight() { + return Err(invalid_outbox_state(message, "in-flight outbox message")); + } + + if let Some(worker_id) = worker_id { + if !message.is_claimed_by(worker_id) { + return Err(invalid_outbox_state( + message, + "outbox claim held by requesting worker", + )); + } + } + + if message.has_expired_lease_at(now) { + return Err(invalid_outbox_state(message, "unexpired outbox claim")); + } + + Ok(()) +} + +impl HashMapRepository { + fn update_outbox_message( + &self, + message_id: &str, + update: impl FnOnce(&mut OutboxMessage) -> Result, + ) -> Result { + let normalized_id = normalize_outbox_id(message_id); + let mut storage = self + .event_store() + .write() + .map_err(|_| RepositoryError::LockPoisoned("write"))?; + + let events = storage + .get_mut(&normalized_id) + .ok_or_else(|| RepositoryError::NotFound { + id: normalized_id.clone(), + })?; + + let mut message = hydrate_outbox_message(&normalized_id, events)?; + let result = update(&mut message)?; + persist_outbox_message(events, &mut message); + Ok(result) + } } impl OutboxRepositoryExt for HashMapRepository { @@ -53,9 +171,7 @@ impl OutboxRepositoryExt for HashMapRepository { continue; } - let mut entity = Entity::with_id(id.to_string()); - entity.load_from_history(events.clone()); - let message = hydrate::(entity)?; + let message = hydrate_outbox_message(id, events)?; if message.status == status { messages.push(message); @@ -76,20 +192,22 @@ impl OutboxRepositoryExt for HashMapRepository { .write() .map_err(|_| RepositoryError::LockPoisoned("write"))?; + if max == 0 { + return Ok(Vec::new()); + } + + let now = SystemTime::now(); let mut claimed = Vec::new(); for (id, events) in storage.iter_mut() { if !id.starts_with(OutboxMessage::ID_PREFIX) { continue; } - let mut entity = Entity::with_id(id.to_string()); - entity.load_from_history(events.clone()); - let mut message = hydrate::(entity)?; + let mut message = hydrate_outbox_message(id, events)?; - if message.is_pending() { - message.claim_for(worker_id, lease)?; - *events = message.entity.events().to_vec(); - message.entity.mark_committed(); + if message.is_claimable_at(now) { + message.claim_at(worker_id, lease, now)?; + persist_outbox_message(events, &mut message); claimed.push(message); } @@ -102,81 +220,265 @@ impl OutboxRepositoryExt for HashMapRepository { } fn complete_outbox_message(&self, message_id: &str) -> Result<(), RepositoryError> { - let normalized_id = if message_id.starts_with(OutboxMessage::ID_PREFIX) { - message_id.to_string() - } else { - format!("{}{}", OutboxMessage::ID_PREFIX, message_id) - }; + self.update_outbox_message(message_id, |message| { + ensure_active_claim(message, None, SystemTime::now())?; + message.complete()?; + Ok(()) + }) + } - let mut storage = self - .event_store() - .write() - .map_err(|_| RepositoryError::LockPoisoned("write"))?; + fn complete_outbox_message_for_worker( + &self, + message_id: &str, + worker_id: &str, + ) -> Result<(), RepositoryError> { + self.update_outbox_message(message_id, |message| { + ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + message.complete()?; + Ok(()) + }) + } - if let Some(events) = storage.get_mut(&normalized_id) { - let mut entity = Entity::with_id(normalized_id); - entity.load_from_history(events.clone()); - let mut message = hydrate::(entity)?; + fn release_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError> { + self.update_outbox_message(message_id, |message| { + ensure_active_claim(message, None, SystemTime::now())?; + message.release(error.to_string())?; + Ok(()) + }) + } + + fn release_outbox_message_for_worker( + &self, + message_id: &str, + worker_id: &str, + error: &str, + ) -> Result<(), RepositoryError> { + self.update_outbox_message(message_id, |message| { + ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + message.release(error.to_string())?; + Ok(()) + }) + } - if message.is_in_flight() { - message.complete()?; - *events = message.entity.events().to_vec(); - message.entity.mark_committed(); + fn fail_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError> { + self.update_outbox_message(message_id, |message| { + if message.is_published() || message.is_failed() { + return Err(invalid_outbox_state( + message, + "outbox message that can be failed", + )); } - } + message.fail(error.to_string())?; + Ok(()) + }) + } - Ok(()) + fn record_outbox_publish_failure( + &self, + message_id: &str, + worker_id: &str, + error: &str, + max_attempts: u32, + ) -> Result { + self.update_outbox_message(message_id, |message| { + ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + if message.attempts >= max_attempts { + message.fail(error.to_string())?; + Ok(OutboxPublishFailureAction::Failed) + } else { + message.release(error.to_string())?; + Ok(OutboxPublishFailureAction::Released) + } + }) } +} - fn release_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError> { - let normalized_id = if message_id.starts_with(OutboxMessage::ID_PREFIX) { - message_id.to_string() - } else { - format!("{}{}", OutboxMessage::ID_PREFIX, message_id) - }; +#[cfg(test)] +mod tests { + use super::*; + use crate::aggregate::GetAggregate; + use crate::repository::Commit; + use std::sync::{Arc, Barrier}; + use std::thread; + + fn store_message(repo: &HashMapRepository, message: &mut OutboxMessage) -> String { + let id = message.id().to_string(); + repo.commit(&mut message.entity).unwrap(); + id + } - let mut storage = self - .event_store() - .write() - .map_err(|_| RepositoryError::LockPoisoned("write"))?; + fn load_message(repo: &HashMapRepository, id: &str) -> OutboxMessage { + repo.get_aggregate::(id).unwrap().unwrap() + } - if let Some(events) = storage.get_mut(&normalized_id) { - let mut entity = Entity::with_id(normalized_id); - entity.load_from_history(events.clone()); - let mut message = hydrate::(entity)?; + #[test] + fn claim_includes_expired_in_flight_messages() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + message + .claim_at("worker-1", Duration::from_secs(1), SystemTime::UNIX_EPOCH) + .unwrap(); + let id = store_message(&repo, &mut message); + + let claimed = repo + .claim_outbox_messages("worker-2", 1, Duration::from_secs(60)) + .unwrap(); + + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].worker_id.as_deref(), Some("worker-2")); + assert_eq!(claimed[0].attempts, 2); + + let stored = load_message(&repo, &id); + assert_eq!(stored.worker_id.as_deref(), Some("worker-2")); + assert_eq!(stored.attempts, 2); + assert!(stored.is_in_flight()); + } - if message.is_in_flight() { - message.release(error.to_string())?; - *events = message.entity.events().to_vec(); - message.entity.mark_committed(); - } - } + #[test] + fn claim_skips_unexpired_in_flight_messages() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + message + .claim_for("worker-1", Duration::from_secs(60)) + .unwrap(); + let id = store_message(&repo, &mut message); + + let claimed = repo + .claim_outbox_messages("worker-2", 1, Duration::from_secs(60)) + .unwrap(); + + assert!(claimed.is_empty()); + let stored = load_message(&repo, &id); + assert_eq!(stored.worker_id.as_deref(), Some("worker-1")); + assert_eq!(stored.attempts, 1); + } - Ok(()) + #[test] + fn competing_workers_only_claim_message_once() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + let id = store_message(&repo, &mut message); + + let barrier = Arc::new(Barrier::new(3)); + let repo_a = repo.clone(); + let repo_b = repo.clone(); + let barrier_a = Arc::clone(&barrier); + let barrier_b = Arc::clone(&barrier); + + let worker_a = thread::spawn(move || { + barrier_a.wait(); + repo_a + .claim_outbox_messages("worker-a", 1, Duration::from_secs(60)) + .unwrap() + .len() + }); + let worker_b = thread::spawn(move || { + barrier_b.wait(); + repo_b + .claim_outbox_messages("worker-b", 1, Duration::from_secs(60)) + .unwrap() + .len() + }); + + barrier.wait(); + let total_claimed = worker_a.join().unwrap() + worker_b.join().unwrap(); + + assert_eq!(total_claimed, 1); + let stored = load_message(&repo, &id); + assert!(stored.is_in_flight()); + assert_eq!(stored.attempts, 1); } - fn fail_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError> { - let normalized_id = if message_id.starts_with(OutboxMessage::ID_PREFIX) { - message_id.to_string() - } else { - format!("{}{}", OutboxMessage::ID_PREFIX, message_id) - }; + #[test] + fn publish_failure_releases_until_retry_ceiling_then_fails() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + let id = store_message(&repo, &mut message); + + repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) + .unwrap(); + let action = repo + .record_outbox_publish_failure(&id, "worker-1", "first failure", 2) + .unwrap(); + assert_eq!(action, OutboxPublishFailureAction::Released); + + let stored = load_message(&repo, &id); + assert!(stored.is_pending()); + assert_eq!(stored.attempts, 1); + assert_eq!(stored.last_error.as_deref(), Some("first failure")); + + repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) + .unwrap(); + let action = repo + .record_outbox_publish_failure(&id, "worker-1", "second failure", 2) + .unwrap(); + assert_eq!(action, OutboxPublishFailureAction::Failed); + + let stored = load_message(&repo, &id); + assert!(stored.is_failed()); + assert_eq!(stored.attempts, 2); + assert_eq!(stored.last_error.as_deref(), Some("second failure")); + } - let mut storage = self - .event_store() - .write() - .map_err(|_| RepositoryError::LockPoisoned("write"))?; + #[test] + fn missing_message_updates_return_not_found() { + let repo = HashMapRepository::new(); + let expected = RepositoryError::NotFound { + id: "outbox:missing".into(), + }; - if let Some(events) = storage.get_mut(&normalized_id) { - let mut entity = Entity::with_id(normalized_id); - entity.load_from_history(events.clone()); - let mut message = hydrate::(entity)?; + assert_eq!( + repo.complete_outbox_message("missing").unwrap_err(), + expected + ); + assert_eq!( + repo.release_outbox_message("missing", "error").unwrap_err(), + expected + ); + assert_eq!( + repo.fail_outbox_message("missing", "error").unwrap_err(), + expected + ); + } - message.fail(error.to_string())?; - *events = message.entity.events().to_vec(); - message.entity.mark_committed(); - } + #[test] + fn stale_or_mismatched_claims_cannot_be_completed() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + let id = store_message(&repo, &mut message); + + repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) + .unwrap(); + let err = repo + .complete_outbox_message_for_worker(&id, "worker-2") + .unwrap_err(); + assert!(matches!(err, RepositoryError::InvalidState { .. })); + + let mut expired = OutboxMessage::create("msg-2", "Event", b"{}".to_vec()).unwrap(); + expired + .claim_at("worker-1", Duration::from_secs(1), SystemTime::UNIX_EPOCH) + .unwrap(); + let expired_id = store_message(&repo, &mut expired); + let err = repo + .complete_outbox_message_for_worker(&expired_id, "worker-1") + .unwrap_err(); + assert!(matches!(err, RepositoryError::InvalidState { .. })); + } - Ok(()) + #[test] + fn already_published_message_is_not_completed_again() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + let id = store_message(&repo, &mut message); + + repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) + .unwrap(); + repo.complete_outbox_message_for_worker(&id, "worker-1") + .unwrap(); + + let err = repo + .complete_outbox_message_for_worker(&id, "worker-1") + .unwrap_err(); + assert!(matches!(err, RepositoryError::InvalidState { .. })); } } diff --git a/src/outbox_worker/thread.rs b/src/outbox_worker/thread.rs index 3d8cc665..0a5d9a4c 100644 --- a/src/outbox_worker/thread.rs +++ b/src/outbox_worker/thread.rs @@ -11,6 +11,8 @@ use std::{error::Error, fmt}; use crate::bus::{Event, Publisher, Sender as BusSender}; use crate::OutboxRepositoryExt; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; + /// Statistics from the outbox worker. #[derive(Debug, Default, Clone)] pub struct WorkerStats { @@ -117,13 +119,20 @@ impl OutboxWorkerThread { match publisher.publish(event) { Ok(()) => { // Mark as complete - if repo.complete_outbox_message(msg.id()).is_ok() { + if repo + .complete_outbox_message_for_worker(msg.id(), &worker_id) + .is_ok() + { stats.messages_published += 1; } } Err(_) => { - // Release for retry - let _ = repo.release_outbox_message(msg.id(), "publish failed"); + let _ = repo.record_outbox_publish_failure( + msg.id(), + &worker_id, + "publish failed", + DEFAULT_MAX_ATTEMPTS, + ); stats.messages_failed += 1; } } @@ -201,12 +210,20 @@ impl OutboxWorkerThread { match result { Ok(()) => { - if repo.complete_outbox_message(msg.id()).is_ok() { + if repo + .complete_outbox_message_for_worker(msg.id(), &worker_id) + .is_ok() + { stats.messages_published += 1; } } Err(_) => { - let _ = repo.release_outbox_message(msg.id(), "publish failed"); + let _ = repo.record_outbox_publish_failure( + msg.id(), + &worker_id, + "publish failed", + DEFAULT_MAX_ATTEMPTS, + ); stats.messages_failed += 1; } } @@ -257,6 +274,18 @@ impl Drop for OutboxWorkerThread { #[cfg(test)] mod tests { use super::*; + use crate::aggregate::GetAggregate; + use crate::bus::PublishError; + use crate::repository::Commit; + use crate::{HashMapRepository, OutboxMessage}; + + struct FailingPublisher; + + impl Publisher for FailingPublisher { + fn publish(&self, _event: Event) -> Result<(), PublishError> { + Err(PublishError::Rejected("forced failure".into())) + } + } #[test] fn stop_returns_stats_when_worker_exits_cleanly() { @@ -298,4 +327,34 @@ mod tests { assert_eq!(err, OutboxWorkerJoinError); } + + #[test] + fn worker_thread_fails_message_after_retry_ceiling() { + let repo = HashMapRepository::new(); + let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + let id = message.id().to_string(); + repo.commit(&mut message.entity).unwrap(); + + let worker = OutboxWorkerThread::spawn_with_id( + repo.clone(), + FailingPublisher, + Duration::from_millis(1), + "worker-1", + ); + + for _ in 0..100 { + let stored = repo.get_aggregate::(&id).unwrap().unwrap(); + if stored.is_failed() { + break; + } + thread::sleep(Duration::from_millis(5)); + } + + let stats = worker.stop(); + let stored = repo.get_aggregate::(&id).unwrap().unwrap(); + + assert!(stored.is_failed()); + assert_eq!(stored.attempts, DEFAULT_MAX_ATTEMPTS); + assert!(stats.messages_failed >= DEFAULT_MAX_ATTEMPTS as usize); + } } diff --git a/src/repository/error.rs b/src/repository/error.rs index 02614e04..06adb761 100644 --- a/src/repository/error.rs +++ b/src/repository/error.rs @@ -16,6 +16,14 @@ pub enum RepositoryError { DuplicateStreamInBatch { id: String, }, + NotFound { + id: String, + }, + InvalidState { + id: String, + expected: &'static str, + actual: String, + }, Replay(String), Model(String), } @@ -39,6 +47,16 @@ impl fmt::Display for RepositoryError { RepositoryError::DuplicateStreamInBatch { id } => { write!(f, "duplicate stream id in commit batch: {}", id) } + RepositoryError::NotFound { id } => write!(f, "entity not found: {}", id), + RepositoryError::InvalidState { + id, + expected, + actual, + } => write!( + f, + "invalid state for entity {} (expected {}, got {})", + id, expected, actual + ), RepositoryError::Replay(message) => write!(f, "replay error: {}", message), RepositoryError::Model(message) => write!(f, "model error: {}", message), } From 0fa4373e2041f1f51f70e69cdf76fbeaf4b013f9 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Wed, 20 May 2026 23:11:35 -0500 Subject: [PATCH 2/2] fix: address outbox worker review feedback --- README.md | 22 +++++--- src/outbox_worker/mod.rs | 17 ++++-- src/outbox_worker/repository_ext.rs | 19 ++++++- src/outbox_worker/thread.rs | 88 ++++++++++++++++++++--------- 4 files changed, 105 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index adbd67e7..ec74550b 100644 --- a/README.md +++ b/README.md @@ -619,13 +619,21 @@ use sourced_rust::{LogPublisher, OutboxRepositoryExt, OutboxWorker}; use std::time::Duration; let repo = HashMapRepository::new(); -let mut worker = OutboxWorker::new(LogPublisher::new()); - -let mut claimed = repo.claim_outbox_messages("worker-1", 100, Duration::from_secs(30))?; -let _ = worker.process_batch(&mut claimed); - -for message in &mut claimed { - repo.commit(&mut message.entity)?; +let worker_id = "worker-1"; +let mut worker = OutboxWorker::new(LogPublisher::new()) + .with_worker_id(worker_id) + .with_max_attempts(3); + +let claimed = repo.claim_outbox_messages(worker_id, 100, Duration::from_secs(30))?; + +for mut message in claimed { + let result = worker.process_message(&mut message)?; + if result.completed { + repo.complete_outbox_message_for_worker(message.id(), worker_id)?; + } else if result.released || result.failed { + let error = message.last_error.as_deref().unwrap_or("publish failed"); + repo.record_outbox_publish_failure(message.id(), worker_id, error, 3)?; + } } ``` diff --git a/src/outbox_worker/mod.rs b/src/outbox_worker/mod.rs index 80ac4ea4..8a1265a1 100644 --- a/src/outbox_worker/mod.rs +++ b/src/outbox_worker/mod.rs @@ -17,15 +17,22 @@ //! //! ```ignore //! use sourced_rust::{OutboxWorker, OutboxRepositoryExt, LogPublisher}; +//! use std::time::Duration; //! //! // Claim pending messages -//! let messages = repo.claim_outbox_messages("worker-1", 10, Duration::from_secs(60))?; +//! let worker_id = "worker-1"; +//! let messages = repo.claim_outbox_messages(worker_id, 10, Duration::from_secs(60))?; //! //! // Process with a worker -//! let mut worker = OutboxWorker::new(LogPublisher::default()); -//! for msg in messages { -//! worker.process_message(&mut msg); -//! repo.complete_outbox_message_for_worker(msg.id(), "worker-1")?; +//! let mut worker = OutboxWorker::new(LogPublisher::default()).with_worker_id(worker_id); +//! for mut msg in messages { +//! let result = worker.process_message(&mut msg)?; +//! if result.completed { +//! repo.complete_outbox_message_for_worker(msg.id(), worker_id)?; +//! } else if result.released || result.failed { +//! let error = msg.last_error.as_deref().unwrap_or("publish failed"); +//! repo.record_outbox_publish_failure(msg.id(), worker_id, error, 3)?; +//! } //! } //! ``` diff --git a/src/outbox_worker/repository_ext.rs b/src/outbox_worker/repository_ext.rs index 8c1d64d2..d39d358f 100644 --- a/src/outbox_worker/repository_ext.rs +++ b/src/outbox_worker/repository_ext.rs @@ -33,7 +33,14 @@ pub trait OutboxRepositoryExt: Send + Sync { lease: Duration, ) -> Result, RepositoryError>; - /// Mark an outbox message as completed (published). + /// Administrative completion path that does not verify a worker ID. + /// + /// Worker loops should use [`Self::complete_outbox_message_for_worker`] + /// so stale or stolen leases cannot complete messages claimed by another + /// worker. + #[deprecated( + note = "worker code should use complete_outbox_message_for_worker to validate the active claim" + )] fn complete_outbox_message(&self, message_id: &str) -> Result<(), RepositoryError>; /// Mark an outbox message as completed if it is still claimed by this worker. @@ -43,7 +50,14 @@ pub trait OutboxRepositoryExt: Send + Sync { worker_id: &str, ) -> Result<(), RepositoryError>; - /// Release an outbox message back to pending (for retry). + /// Administrative release path that does not verify a worker ID. + /// + /// Worker loops should use [`Self::release_outbox_message_for_worker`] or + /// [`Self::record_outbox_publish_failure`] so stale or stolen leases cannot + /// release messages claimed by another worker. + #[deprecated( + note = "worker code should use release_outbox_message_for_worker or record_outbox_publish_failure to validate the active claim" + )] fn release_outbox_message(&self, message_id: &str, error: &str) -> Result<(), RepositoryError>; /// Release an outbox message if it is still claimed by this worker. @@ -421,6 +435,7 @@ mod tests { } #[test] + #[allow(deprecated)] fn missing_message_updates_return_not_found() { let repo = HashMapRepository::new(); let expected = RepositoryError::NotFound { diff --git a/src/outbox_worker/thread.rs b/src/outbox_worker/thread.rs index 0a5d9a4c..0f0c3e29 100644 --- a/src/outbox_worker/thread.rs +++ b/src/outbox_worker/thread.rs @@ -33,6 +33,43 @@ impl fmt::Display for OutboxWorkerJoinError { impl Error for OutboxWorkerJoinError {} +fn record_publish_success( + repo: &R, + message_id: &str, + worker_id: &str, + stats: &mut WorkerStats, +) { + match repo.complete_outbox_message_for_worker(message_id, worker_id) { + Ok(()) => { + stats.messages_published += 1; + } + Err(err) => { + eprintln!("outbox worker `{worker_id}` could not complete `{message_id}`: {err}"); + stats.messages_failed += 1; + } + } +} + +fn record_publish_failure( + repo: &R, + message_id: &str, + worker_id: &str, + error: &str, + stats: &mut WorkerStats, +) { + match repo.record_outbox_publish_failure(message_id, worker_id, error, DEFAULT_MAX_ATTEMPTS) { + Ok(_) => { + stats.messages_failed += 1; + } + Err(err) => { + eprintln!( + "outbox worker `{worker_id}` could not record publish failure for `{message_id}`: {err}" + ); + stats.messages_failed += 1; + } + } +} + /// A background thread that drains outbox messages and publishes to a bus. /// /// ## Example @@ -118,28 +155,23 @@ impl OutboxWorkerThread { match publisher.publish(event) { Ok(()) => { - // Mark as complete - if repo - .complete_outbox_message_for_worker(msg.id(), &worker_id) - .is_ok() - { - stats.messages_published += 1; - } + record_publish_success(&repo, msg.id(), &worker_id, &mut stats); } - Err(_) => { - let _ = repo.record_outbox_publish_failure( + Err(err) => { + let error = err.to_string(); + record_publish_failure( + &repo, msg.id(), &worker_id, - "publish failed", - DEFAULT_MAX_ATTEMPTS, + &error, + &mut stats, ); - stats.messages_failed += 1; } } } } - Err(_) => { - // Repository error, continue polling + Err(err) => { + eprintln!("outbox worker `{worker_id}` could not claim messages: {err}"); } } @@ -210,26 +242,24 @@ impl OutboxWorkerThread { match result { Ok(()) => { - if repo - .complete_outbox_message_for_worker(msg.id(), &worker_id) - .is_ok() - { - stats.messages_published += 1; - } + record_publish_success(&repo, msg.id(), &worker_id, &mut stats); } - Err(_) => { - let _ = repo.record_outbox_publish_failure( + Err(err) => { + let error = err.to_string(); + record_publish_failure( + &repo, msg.id(), &worker_id, - "publish failed", - DEFAULT_MAX_ATTEMPTS, + &error, + &mut stats, ); - stats.messages_failed += 1; } } } } - Err(_) => {} + Err(err) => { + eprintln!("outbox worker `{worker_id}` could not claim messages: {err}"); + } } thread::sleep(poll_interval); @@ -350,11 +380,15 @@ mod tests { thread::sleep(Duration::from_millis(5)); } - let stats = worker.stop(); + let stats = worker.stop().unwrap(); let stored = repo.get_aggregate::(&id).unwrap().unwrap(); assert!(stored.is_failed()); assert_eq!(stored.attempts, DEFAULT_MAX_ATTEMPTS); + assert_eq!( + stored.last_error.as_deref(), + Some("Event rejected: forced failure") + ); assert!(stats.messages_failed >= DEFAULT_MAX_ATTEMPTS as usize); } }