fix: harden outbox claim leases

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};
 
@@ -614,16 +619,30 @@ 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)?;
+    }
 }
 ```
 
+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:

src/lib.rs

@@ -56,6 +56,7 @@ pub use outbox_worker::{
     // Publishers
     LogPublisher,
     LogPublisherError,
+    OutboxPublishFailureAction,
     OutboxPublisher,
     // Repository extension for claiming/completing messages
     OutboxRepositoryExt,

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<String>, 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<String>,
+        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(

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

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(msg.id())?;
+//! 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)?;
+//!     }
 //! }
 //! ```
 
@@ -41,7 +48,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};