diff --git a/crates/traverse-runtime/src/events/broker.rs b/crates/traverse-runtime/src/events/broker.rs index 4df2a29..72719d6 100644 --- a/crates/traverse-runtime/src/events/broker.rs +++ b/crates/traverse-runtime/src/events/broker.rs @@ -69,6 +69,8 @@ struct BrokerState { buffers: HashMap>, seen_event_ids: HashMap>, subscriptions: HashMap, + /// See [`EventBroker::seed_restart_floor`]. + restart_floor: u64, } /// Synchronous, in-memory implementation of [`EventBroker`]. @@ -182,6 +184,104 @@ impl InProcessBroker { cursor: cursor_to_string(from_cursor), }) } + + /// Shared implementation for `publish` and `publish_with_cursor`. + /// `assigned_cursor`, when given, is adopted as this event's cursor + /// instead of self-incrementing the per-type counter (spec 066 FR-007: + /// keeps live-delivery cursors consistent with the durable journal's + /// cursor space). The per-type counter still tracks the highest cursor + /// ever used so `validate_from_cursor`'s empty-buffer fallback remains + /// correct regardless of cursor source. + fn publish_internal( + &self, + event: &TraverseEvent, + assigned_cursor: Option, + ) -> Result<(), EventError> { + let entry = self + .catalog + .get(&event.event_type) + .ok_or_else(|| EventError::UnregisteredEventType(event.event_type.clone()))?; + + match entry.lifecycle_status { + LifecycleStatus::Active => {} + LifecycleStatus::Deprecated => { + return Err(EventError::LifecycleViolation(format!( + "event type '{}' is Deprecated and cannot be published", + event.event_type + ))); + } + LifecycleStatus::Draft => { + return Err(EventError::LifecycleViolation(format!( + "event type '{}' is Draft and cannot be published", + event.event_type + ))); + } + } + + let now = self.clock.now(); + + let mut state = self + .state + .lock() + .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?; + + prune_expired( + &mut state, + &event.event_type, + self.config.retention_window, + now, + ); + + let seen = state + .seen_event_ids + .entry(event.event_type.clone()) + .or_default(); + if seen.contains(&event.id) { + // Duplicate emissions are silently discarded. + return Ok(()); + } + seen.insert(event.id.clone()); + + let next = state + .next_cursor + .entry(event.event_type.clone()) + .or_insert(0); + let cursor = if let Some(assigned) = assigned_cursor { + *next = (*next).max(assigned); + assigned + } else { + *next = next.saturating_add(1); + *next + }; + + let buffered = BufferedEvent { + cursor, + published_at: now, + event: event.clone(), + }; + + state + .buffers + .entry(event.event_type.clone()) + .or_default() + .push_back(buffered.clone()); + + for sub in state.subscriptions.values_mut() { + if sub.event_type != event.event_type { + continue; + } + if sub + .subject_id + .as_deref() + .is_some_and(|subject_id| event.subject_id.as_deref() != Some(subject_id)) + { + continue; + } + enqueue_with_drop_oldest(&mut sub.queue, self.config.max_queue_len, buffered.clone()); + } + + Ok(()) + } } fn parse_cursor(raw: &str) -> Result { @@ -263,7 +363,12 @@ fn validate_from_cursor( return Ok(()); } - let last_cursor = state.next_cursor.get(event_type).copied().unwrap_or(0); + let last_cursor = state + .next_cursor + .get(event_type) + .copied() + .unwrap_or(0) + .max(state.restart_floor); if let Some(buffer) = state.buffers.get(event_type) && let Some(front) = buffer.front() { @@ -299,6 +404,12 @@ impl EventBroker for InProcessBroker { self.subscribe_with_subject(event_type, from_cursor, subject_id) } + fn seed_restart_floor(&self, floor: u64) { + if let Ok(mut state) = self.state.lock() { + state.restart_floor = state.restart_floor.max(floor); + } + } + /// Publish `event` to all registered subscribers. /// /// # Errors @@ -306,85 +417,23 @@ impl EventBroker for InProcessBroker { /// - [`EventError::UnregisteredEventType`] if the event type is not in the catalog. /// - [`EventError::LifecycleViolation`] if the catalog entry is `Draft` or `Deprecated`. fn publish(&self, event: TraverseEvent) -> Result<(), EventError> { - let entry = self - .catalog - .get(&event.event_type) - .ok_or_else(|| EventError::UnregisteredEventType(event.event_type.clone()))?; - - match entry.lifecycle_status { - LifecycleStatus::Active => {} - LifecycleStatus::Deprecated => { - return Err(EventError::LifecycleViolation(format!( - "event type '{}' is Deprecated and cannot be published", - event.event_type - ))); - } - LifecycleStatus::Draft => { - return Err(EventError::LifecycleViolation(format!( - "event type '{}' is Draft and cannot be published", - event.event_type - ))); - } - } - - let now = self.clock.now(); - - let mut state = self - .state - .lock() - .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?; - - prune_expired( - &mut state, - &event.event_type, - self.config.retention_window, - now, - ); - - let seen = state - .seen_event_ids - .entry(event.event_type.clone()) - .or_default(); - if seen.contains(&event.id) { - // Duplicate emissions are silently discarded. - return Ok(()); - } - seen.insert(event.id.clone()); - - let next = state - .next_cursor - .entry(event.event_type.clone()) - .or_insert(0); - *next = next.saturating_add(1); - let cursor = *next; - - let buffered = BufferedEvent { - cursor, - published_at: now, - event: event.clone(), - }; - - state - .buffers - .entry(event.event_type.clone()) - .or_default() - .push_back(buffered.clone()); - - for sub in state.subscriptions.values_mut() { - if sub.event_type != event.event_type { - continue; - } - if sub - .subject_id - .as_deref() - .is_some_and(|subject_id| event.subject_id.as_deref() != Some(subject_id)) - { - continue; - } - enqueue_with_drop_oldest(&mut sub.queue, self.config.max_queue_len, buffered.clone()); - } + self.publish_internal(&event, None) + } - Ok(()) + /// Publish `event`, adopting `cursor` as this broker's own cursor for it + /// instead of self-assigning the next per-type sequence value. Used by + /// [`DurableBroker`](super::durable::DurableBroker) so live-delivery + /// cursors stay numerically consistent with the durable journal's cursor + /// space (spec 066 FR-007): a cursor obtained during live polling + /// remains valid when a later `poll` falls back to durable replay. + /// + /// # Errors + /// + /// Returns [`EventError::InvalidCursor`] when `cursor` is not a base-10 + /// unsigned integer, plus the same errors as [`Self::publish`]. + fn publish_with_cursor(&self, event: TraverseEvent, cursor: &str) -> Result<(), EventError> { + let assigned = parse_cursor(cursor)?; + self.publish_internal(&event, Some(assigned)) } /// Create a subscription for `event_type` starting from `from_cursor`. @@ -571,6 +620,105 @@ mod tests { assert!(matches!(err, EventError::InvalidRetentionWindow(_))); } + #[test] + fn publish_with_cursor_adopts_the_given_cursor_instead_of_self_assigning() { + let event_type = "dev.traverse.injected-cursor"; + let catalog = make_catalog(event_type, LifecycleStatus::Active); + let broker = InProcessBroker::new(catalog).expect("broker must be created"); + + broker + .publish_with_cursor(sample_event(event_type, "evt-1"), "42") + .expect("publish_with_cursor must succeed"); + + let subscription = broker + .subscribe(event_type, "0") + .expect("subscribe must succeed"); + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("poll must succeed"); + assert_eq!(poll.events.len(), 1); + assert_eq!(poll.events[0].cursor, "42"); + assert_eq!(poll.cursor, "42"); + } + + #[test] + fn publish_with_cursor_rejects_a_malformed_cursor() { + let event_type = "dev.traverse.injected-cursor-invalid"; + let catalog = make_catalog(event_type, LifecycleStatus::Active); + let broker = InProcessBroker::new(catalog).expect("broker must be created"); + + let err = broker + .publish_with_cursor(sample_event(event_type, "evt-1"), "not-a-cursor") + .expect_err("malformed cursor must be rejected"); + assert!(matches!(err, EventError::InvalidCursor(_))); + } + + #[test] + fn default_publish_with_cursor_ignores_the_cursor_and_self_assigns() { + struct SelfAssigningOnlyBroker(InProcessBroker); + + impl EventBroker for SelfAssigningOnlyBroker { + fn publish(&self, event: TraverseEvent) -> Result<(), EventError> { + self.0.publish(event) + } + fn subscribe( + &self, + event_type: &str, + from_cursor: &str, + ) -> Result { + self.0.subscribe(event_type, from_cursor) + } + fn subscribe_for_subject( + &self, + event_type: &str, + from_cursor: &str, + subject_id: Option<&str>, + ) -> Result { + self.0 + .subscribe_for_subject(event_type, from_cursor, subject_id) + } + fn poll( + &self, + subscription_id: &str, + max_events: usize, + ) -> Result { + self.0.poll(subscription_id, max_events) + } + fn cancel(&self, subscription_id: &str) -> Result<(), EventError> { + self.0.cancel(subscription_id) + } + } + + let event_type = "dev.traverse.default-publish-with-cursor"; + let catalog = make_catalog(event_type, LifecycleStatus::Active); + let broker = + SelfAssigningOnlyBroker(InProcessBroker::new(catalog).expect("broker must be created")); + + // The default `publish_with_cursor` ignores the supplied cursor and + // behaves exactly like `publish` (self-assigning cursor "1"). + broker + .publish_with_cursor(sample_event(event_type, "evt-1"), "999") + .expect("default publish_with_cursor must succeed"); + + let subscription = broker + .subscribe(event_type, "0") + .expect("subscribe must succeed"); + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("poll must succeed"); + assert_eq!(poll.events[0].cursor, "1"); + + // The default `seed_restart_floor` is a no-op; exercise it alongside + // this wrapper's other pass-through trait methods. + broker.seed_restart_floor(999); + let subject_subscription = broker + .subscribe_for_subject(event_type, "0", None) + .expect("subscribe_for_subject must succeed"); + broker + .cancel(&subject_subscription.subscription_id) + .expect("cancel must succeed"); + } + #[test] fn invalid_cursor_is_rejected() { let catalog = make_catalog("dev.traverse.cursor", LifecycleStatus::Active); diff --git a/crates/traverse-runtime/src/events/durable.rs b/crates/traverse-runtime/src/events/durable.rs index d5785c2..f98a898 100644 --- a/crates/traverse-runtime/src/events/durable.rs +++ b/crates/traverse-runtime/src/events/durable.rs @@ -6,14 +6,108 @@ //! downgraded to in-memory-only delivery — and every timeout produces a //! structured audit record (spec 036 NFR-006). +use std::collections::HashMap; +use std::path::Path; use std::sync::mpsc; -use std::sync::{Arc, Condvar, Mutex, PoisonError}; +use std::sync::{Arc, Condvar, Mutex, PoisonError, RwLock}; use std::time::Duration; use serde::Serialize; -use super::journal::{DurableEventJournal, JournalError}; -use super::types::{EventBroker, EventError, Subscription, SubscriptionPoll, TraverseEvent}; +use super::broker::BrokerClock; +use super::journal::{DurableEventJournal, JournalConfig, JournalError}; +use super::types::{ + BrokerEvent, EventBroker, EventError, Subscription, SubscriptionPoll, TraverseEvent, +}; + +/// Prefix distinguishing durable-journal-backed subscription ids from the +/// inner broker's own `sub-*` ids, so [`DurableBroker::poll`] and +/// [`DurableBroker::cancel`] can route without a second lookup table. +const DURABLE_SUBSCRIPTION_PREFIX: &str = "durable-sub-"; + +/// Read-side of a durable journal: serves replay for cursors the live +/// broker's in-memory window no longer retains (spec 066 FR-005, FR-008). +/// [`RwLock`] is the production implementation, shared +/// with the write path via [`DurableBroker::open`]; tests inject a stub for +/// write-path-only scenarios that never exercise replay. +pub trait JournalSource: Send + Sync { + /// Replay up to `max_events` events strictly after `cursor`. + /// + /// # Errors + /// + /// Returns [`JournalError`] when the cursor is malformed or expired, or + /// the durable read fails. + fn replay_from( + &self, + cursor: &str, + max_events: usize, + ) -> Result, JournalError>; +} + +impl JournalSource for RwLock { + fn replay_from( + &self, + cursor: &str, + max_events: usize, + ) -> Result, JournalError> { + self.read() + .unwrap_or_else(PoisonError::into_inner) + .replay_from(cursor, max_events) + } +} + +/// Write-side adapter that locks a shared journal for each append, letting +/// [`DurableBroker::open`] give the writer thread and the read path the same +/// underlying storage so cursors stay consistent between them. +struct SharedJournalSink(Arc>); + +impl JournalSink for SharedJournalSink { + fn append_event(&mut self, event: &TraverseEvent) -> Result { + self.0 + .write() + .unwrap_or_else(PoisonError::into_inner) + .append(event) + } + + fn append_revocation(&mut self, revoked_cursor: &str) -> Result { + self.0 + .write() + .unwrap_or_else(PoisonError::into_inner) + .append_revocation(revoked_cursor) + } +} + +struct DurableSubscriptionState { + event_type: String, + subject_id: Option, + cursor: String, +} + +#[derive(Default)] +struct DurableSubscriptions { + next_id: u64, + entries: HashMap, +} + +fn map_journal_read_error(event_type: &str, error: JournalError) -> EventError { + match error { + JournalError::CursorExpired { + oldest_available_cursor, + } => EventError::CursorExpired { + event_type: event_type.to_string(), + oldest_available_cursor, + }, + JournalError::InvalidCursor(msg) => EventError::InvalidCursor(msg), + JournalError::Io(msg) | JournalError::InvalidConfig(msg) => EventError::JournalRead(msg), + JournalError::Corrupt { + path, + line, + message, + } => EventError::JournalRead(format!( + "corrupt journal record at {path}:{line}: {message}" + )), + } +} /// Durable sink the writer thread appends through. [`DurableEventJournal`] /// is the production implementation; tests inject slow or failing sinks to @@ -103,22 +197,42 @@ enum WriterJob { /// [`EventBroker`] decorator that makes every published event durable before /// it is delivered: the event is journaled with fsync-before-acknowledgement /// (066 FR-006) and only then forwarded to the inner broker for live -/// delivery. A write that exceeds the configured timeout rejects the event +/// delivery, adopting the journal-assigned cursor as the inner broker's own +/// cursor for that event (spec 066 FR-007) so the two stay numerically +/// consistent. A write that exceeds the configured timeout rejects the event /// with `journal_write_timeout` (067 FR-003/FR-004); if the abandoned write /// completes later, the writer durably revokes it so it can never surface /// through replay. +/// +/// `subscribe`/`poll` normally delegate to the inner broker's fast, +/// in-memory delivery. When a requested cursor is older than the inner +/// broker's retention window, [`DurableBroker`] falls back to the durable +/// journal (spec 066 FR-005, FR-008): if the journal still retains the +/// cursor, a durable-mode subscription is created that reads through +/// [`JournalSource::replay_from`] for the rest of its lifetime (applying the +/// identical `subject_id` filter as live delivery, FR-003); if the journal +/// has also reclaimed that history, `cursor_expired` is returned with the +/// journal's own oldest available cursor. pub struct DurableBroker { inner: B, jobs: mpsc::Sender, write_timeout: Duration, audit: Arc, + source: Arc, + subscriptions: Mutex, } impl DurableBroker { - /// Wrap `inner` with a durable write path backed by `sink`. + /// Wrap `inner` with a durable write path backed by `sink`, and a + /// journal-backed replay fallback backed by `source`. Production callers + /// should use [`DurableBroker::open`], which guarantees `sink` and + /// `source` share the same underlying journal storage; this lower-level + /// constructor exists so tests can inject independent write- and + /// read-side doubles. pub fn new( inner: B, sink: impl JournalSink + 'static, + source: Arc, config: DurableBrokerConfig, audit: Arc, ) -> Self { @@ -132,8 +246,54 @@ impl DurableBroker { jobs, write_timeout: config.write_timeout, audit, + source, + subscriptions: Mutex::new(DurableSubscriptions::default()), } } + + /// Opens (or creates and recovers) a durable journal at `root` and wraps + /// `inner` with a write path and journal-backed replay fallback sharing + /// that same storage, so cursors stay consistent across live delivery + /// and durable replay (spec 066 FR-007). + /// + /// # Errors + /// + /// Returns [`JournalError`] when the journal cannot be opened or + /// recovered (066 FR-009). + pub fn open( + root: &Path, + inner: B, + journal_config: JournalConfig, + broker_config: DurableBrokerConfig, + audit: Arc, + clock: Arc, + ) -> Result { + let journal = Arc::new(RwLock::new(DurableEventJournal::open( + root, + journal_config, + clock, + )?)); + // A freshly constructed `inner` has no memory of history that + // predates this process (spec 066 FR-007 restart continuity): seed + // it with the journal's own latest cursor so it correctly defers + // cursors it cannot itself vouch for to durable replay, rather than + // optimistically accepting them because it has simply never seen + // this event type before. + inner.seed_restart_floor( + journal + .read() + .unwrap_or_else(PoisonError::into_inner) + .latest_cursor(), + ); + let source: Arc = Arc::clone(&journal) as Arc; + Ok(Self::new( + inner, + SharedJournalSink(journal), + source, + broker_config, + audit, + )) + } } impl EventBroker for DurableBroker { @@ -159,7 +319,7 @@ impl EventBroker for DurableBroker { drop(state); match outcome { - AckState::Done(cursor) => match self.inner.publish(event) { + AckState::Done(cursor) => match self.inner.publish_with_cursor(event, &cursor) { Ok(()) => Ok(()), Err(error) => { // Durably written but undeliverable: revoke so replay @@ -191,7 +351,7 @@ impl EventBroker for DurableBroker { } fn subscribe(&self, event_type: &str, from_cursor: &str) -> Result { - self.inner.subscribe(event_type, from_cursor) + self.subscribe_for_subject(event_type, from_cursor, None) } fn subscribe_for_subject( @@ -200,8 +360,53 @@ impl EventBroker for DurableBroker { from_cursor: &str, subject_id: Option<&str>, ) -> Result { - self.inner + match self + .inner .subscribe_for_subject(event_type, from_cursor, subject_id) + { + Ok(subscription) => Ok(subscription), + Err(EventError::CursorExpired { + event_type: expired_event_type, + .. + }) => { + // The in-memory window no longer retains this cursor; check + // whether the durable journal still does (066 FR-005, + // FR-008). `max_events: 0` only validates the cursor. + match self.source.replay_from(from_cursor, 0) { + Ok(_) => { + let cursor = normalize_cursor(from_cursor)?; + let mut subscriptions = self + .subscriptions + .lock() + .unwrap_or_else(PoisonError::into_inner); + subscriptions.next_id = subscriptions.next_id.saturating_add(1); + let subscription_id = + format!("{DURABLE_SUBSCRIPTION_PREFIX}{}", subscriptions.next_id); + subscriptions.entries.insert( + subscription_id.clone(), + DurableSubscriptionState { + event_type: event_type.to_string(), + subject_id: subject_id.map(str::to_owned), + cursor: cursor.clone(), + }, + ); + Ok(Subscription { + subscription_id, + event_type: event_type.to_string(), + cursor, + }) + } + Err(JournalError::CursorExpired { + oldest_available_cursor, + }) => Err(EventError::CursorExpired { + event_type: expired_event_type, + oldest_available_cursor, + }), + Err(other) => Err(map_journal_read_error(event_type, other)), + } + } + Err(other) => Err(other), + } } fn poll( @@ -209,14 +414,90 @@ impl EventBroker for DurableBroker { subscription_id: &str, max_events: usize, ) -> Result { - self.inner.poll(subscription_id, max_events) + if !subscription_id.starts_with(DURABLE_SUBSCRIPTION_PREFIX) { + return self.inner.poll(subscription_id, max_events); + } + + let mut subscriptions = self + .subscriptions + .lock() + .unwrap_or_else(PoisonError::into_inner); + let subscription = subscriptions + .entries + .get_mut(subscription_id) + .ok_or_else(|| EventError::SubscriptionNotFound(subscription_id.to_string()))?; + + let mut delivered = Vec::new(); + let mut cursor = subscription.cursor.clone(); + if max_events > 0 { + loop { + let batch = self + .source + .replay_from(&cursor, max_events) + .map_err(|error| map_journal_read_error(&subscription.event_type, error))?; + if batch.is_empty() { + break; + } + let batch_len = batch.len(); + for (record_cursor, event) in batch { + cursor = record_cursor; + if event.event_type != subscription.event_type { + continue; + } + if subscription + .subject_id + .as_deref() + .is_some_and(|subject| event.subject_id.as_deref() != Some(subject)) + { + continue; + } + delivered.push(BrokerEvent { + cursor: cursor.clone(), + event, + }); + if delivered.len() >= max_events { + break; + } + } + if delivered.len() >= max_events || batch_len < max_events { + break; + } + } + } + subscription.cursor.clone_from(&cursor); + + Ok(SubscriptionPoll { + subscription_id: subscription_id.to_string(), + event_type: subscription.event_type.clone(), + cursor, + events: delivered, + }) } fn cancel(&self, subscription_id: &str) -> Result<(), EventError> { + if subscription_id.starts_with(DURABLE_SUBSCRIPTION_PREFIX) { + let mut subscriptions = self + .subscriptions + .lock() + .unwrap_or_else(PoisonError::into_inner); + return if subscriptions.entries.remove(subscription_id).is_some() { + Ok(()) + } else { + Err(EventError::SubscriptionNotFound( + subscription_id.to_string(), + )) + }; + } self.inner.cancel(subscription_id) } } +fn normalize_cursor(raw: &str) -> Result { + raw.parse::() + .map(|parsed| parsed.to_string()) + .map_err(|_| EventError::InvalidCursor(raw.to_string())) +} + fn run_writer( mut sink: impl JournalSink, jobs: &mpsc::Receiver, @@ -382,6 +663,39 @@ mod tests { std::env::temp_dir().join(format!("traverse-durable-{name}-{}", Uuid::new_v4())) } + /// Read-side stub for write-path-only tests that never exercise durable + /// replay: every read returns empty results, so any accidental fallback + /// attempt fails loudly with `CursorExpired` rather than silently + /// succeeding. + struct NullJournalSource; + + impl JournalSource for NullJournalSource { + fn replay_from( + &self, + _cursor: &str, + _max_events: usize, + ) -> Result, JournalError> { + Err(JournalError::CursorExpired { + oldest_available_cursor: "0".to_string(), + }) + } + } + + fn null_source() -> Arc { + Arc::new(NullJournalSource) + } + + fn open_shared_journal(root: &std::path::Path) -> Arc> { + Arc::new(RwLock::new( + DurableEventJournal::open( + root, + JournalConfig::default(), + Arc::new(crate::events::broker::SystemClock), + ) + .expect("journal must open"), + )) + } + #[test] fn journal_sink_impl_appends_and_revokes_through_the_real_journal() { let root = test_root("sink-impl"); @@ -406,19 +720,16 @@ mod tests { #[test] fn durable_publish_journals_then_delivers() { let root = test_root("happy"); - let journal = DurableEventJournal::open( - &root, - JournalConfig::default(), - Arc::new(crate::events::broker::SystemClock), - ) - .expect("journal must open"); let audit = Arc::new(RecordingAudit::default()); - let broker = DurableBroker::new( + let broker = DurableBroker::open( + &root, inner_broker(), - journal, + JournalConfig::default(), DurableBrokerConfig::default(), audit.clone(), - ); + Arc::new(crate::events::broker::SystemClock), + ) + .expect("durable broker must open"); let subscription = broker .subscribe(EVENT_TYPE, "0") @@ -447,18 +758,15 @@ mod tests { #[test] fn durable_subject_subscription_delegates_to_inner_broker() { let root = test_root("subject-subscription"); - let journal = DurableEventJournal::open( + let broker = DurableBroker::open( &root, - JournalConfig::default(), - Arc::new(crate::events::broker::SystemClock), - ) - .expect("journal must open"); - let broker = DurableBroker::new( inner_broker(), - journal, + JournalConfig::default(), DurableBrokerConfig::default(), Arc::new(RecordingAudit::default()), - ); + Arc::new(crate::events::broker::SystemClock), + ) + .expect("durable broker must open"); let mut event = test_event(); event.subject_id = Some("subject-match".to_string()); broker.publish(event).expect("publish must succeed"); @@ -486,8 +794,13 @@ mod tests { fail_revocation: false, }; let audit = Arc::new(RecordingAudit::default()); - let broker = - DurableBroker::new(inner_broker(), sink, DurableBrokerConfig::default(), audit); + let broker = DurableBroker::new( + inner_broker(), + sink, + null_source(), + DurableBrokerConfig::default(), + audit, + ); let mut unregistered = test_event(); unregistered.event_type = "dev.traverse.test.unknown".to_string(); @@ -519,6 +832,7 @@ mod tests { let broker = DurableBroker::new( inner_broker(), sink, + null_source(), DurableBrokerConfig { write_timeout: Duration::from_millis(50), }, @@ -570,8 +884,13 @@ mod tests { fail_revocation: false, }; let audit = Arc::new(RecordingAudit::default()); - let broker = - DurableBroker::new(inner_broker(), sink, DurableBrokerConfig::default(), audit); + let broker = DurableBroker::new( + inner_broker(), + sink, + null_source(), + DurableBrokerConfig::default(), + audit, + ); gate_tx .send(Err(JournalError::Io("disk gone".to_string()))) @@ -597,6 +916,7 @@ mod tests { let broker = DurableBroker::new( inner_broker(), sink, + null_source(), DurableBrokerConfig { write_timeout: Duration::from_millis(50), }, @@ -622,4 +942,877 @@ mod tests { assert_eq!(failure_audit.kind, "journal_revocation_failed"); assert!(failure_audit.detail.contains("revocation rejected")); } + + // --- journal-backed replay fallback (spec 066 FR-005, FR-007, FR-008; final #659 slice) --- + + #[derive(Debug)] + struct ManualClock(StdMutex); + + impl ManualClock { + fn new() -> Self { + Self(StdMutex::new(std::time::SystemTime::now())) + } + + fn advance(&self, by: Duration) { + if let Ok(mut guard) = self.0.lock() + && let Some(next) = guard.checked_add(by) + { + *guard = next; + } + } + } + + impl crate::events::broker::BrokerClock for ManualClock { + fn now(&self) -> std::time::SystemTime { + self.0 + .lock() + .ok() + .map_or(std::time::SystemTime::UNIX_EPOCH, |guard| *guard) + } + } + + /// Deterministically simulates a journal that has also reclaimed the + /// requested history, independent of real segment rollover/pruning + /// mechanics (already covered by `journal.rs`'s own test suite). + struct AlwaysExpiredSource; + + impl JournalSource for AlwaysExpiredSource { + fn replay_from( + &self, + _cursor: &str, + _max_events: usize, + ) -> Result, JournalError> { + Err(JournalError::CursorExpired { + oldest_available_cursor: "5".to_string(), + }) + } + } + + fn short_retention_inner( + clock: Arc, + ) -> crate::events::broker::InProcessBroker { + crate::events::broker::InProcessBroker::with_clock( + active_catalog(), + crate::events::broker::BrokerConfig { + retention_window: Duration::from_millis(10), + max_queue_len: 16, + }, + clock, + ) + .expect("short-retention broker must build") + } + + #[test] + fn poll_falls_back_to_the_durable_journal_once_the_in_memory_window_expires() { + let root = test_root("fallback-live"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let audit = Arc::new(RecordingAudit::default()); + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + audit, + ); + + // Publish two events: cursor "0" is a sentinel that never expires + // (it means "start of retention"), so triggering real expiry + // requires resuming from a genuine prior cursor ("1") that is now + // older than the latest retained cursor ("2"). + broker + .publish(test_event()) + .expect("first publish must succeed"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + + // Age the in-memory buffer out; `subscribe` must transparently fall + // back to the durable journal within this single call rather than + // surfacing the inner broker's `CursorExpired` to the caller. + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribing on an expired cursor must fall back to the durable journal"); + assert!( + subscription + .subscription_id + .starts_with(DURABLE_SUBSCRIPTION_PREFIX), + "a durable fallback subscription must use the durable-mode id prefix" + ); + + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("durable poll must succeed"); + assert_eq!( + poll.events.len(), + 1, + "only the event strictly after cursor \"1\" must replay" + ); + + // Publish a second event *after* the durable subscription was + // created and prove it still replays: durable-mode subscriptions + // are not a one-time catchup, they keep working for new events too. + broker + .publish(test_event()) + .expect("second publish must succeed"); + let poll_again = broker + .poll(&subscription.subscription_id, 10) + .expect("second durable poll must succeed"); + assert_eq!(poll_again.events.len(), 1, "the new event must also replay"); + + broker + .cancel(&subscription.subscription_id) + .expect("cancelling a durable-mode subscription must succeed"); + let after_cancel = broker.poll(&subscription.subscription_id, 10); + assert!(matches!( + after_cancel, + Err(EventError::SubscriptionNotFound(_)) + )); + } + + #[test] + fn durable_fallback_applies_the_identical_subject_filter_as_live_delivery() { + let root = test_root("fallback-subject"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + // A leading sentinel-only publish so the real assertions below can + // resume from a genuine prior cursor ("1"), not the "0" sentinel + // that never expires. + broker + .publish(test_event()) + .expect("leading publish must succeed"); + let mut matching = test_event(); + matching.subject_id = Some("subject-match".to_string()); + broker.publish(matching).expect("publish must succeed"); + let mut other = test_event(); + other.subject_id = Some("subject-other".to_string()); + broker.publish(other).expect("publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe_for_subject(EVENT_TYPE, "1", Some("subject-match")) + .expect("subject subscription must fall back to the durable journal"); + assert!( + subscription + .subscription_id + .starts_with(DURABLE_SUBSCRIPTION_PREFIX) + ); + + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("durable poll must succeed"); + assert_eq!( + poll.events.len(), + 1, + "only the matching subject must replay" + ); + assert_eq!( + poll.events[0].event.subject_id.as_deref(), + Some("subject-match") + ); + } + + #[test] + fn durable_fallback_surfaces_cursor_expired_when_the_journal_has_also_reclaimed_history() { + let clock = Arc::new(ManualClock::new()); + let (gate_tx, gate_rx) = channel(); + let (revoked_tx, _revoked_rx) = channel(); + let sink = ScriptedSink { + gate: gate_rx, + revocations: revoked_tx, + fail_revocation: false, + }; + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + sink, + Arc::new(AlwaysExpiredSource) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + gate_tx.send(Ok("1".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("first publish must succeed"); + gate_tx.send(Ok("2".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let err = broker + .subscribe(EVENT_TYPE, "1") + .expect_err("both the in-memory and durable layers have expired this cursor"); + assert_eq!( + err, + EventError::CursorExpired { + event_type: EVENT_TYPE.to_string(), + oldest_available_cursor: "5".to_string(), + }, + "the durable journal's own oldest-available cursor must be surfaced" + ); + } + + #[test] + fn durable_fallback_propagates_non_cursor_journal_read_errors() { + struct FailingSource; + impl JournalSource for FailingSource { + fn replay_from( + &self, + _cursor: &str, + _max_events: usize, + ) -> Result, JournalError> { + Err(JournalError::Io("disk unavailable".to_string())) + } + } + + let clock = Arc::new(ManualClock::new()); + let (gate_tx, gate_rx) = channel(); + let (revoked_tx, _revoked_rx) = channel(); + let sink = ScriptedSink { + gate: gate_rx, + revocations: revoked_tx, + fail_revocation: false, + }; + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + sink, + Arc::new(FailingSource) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + gate_tx.send(Ok("1".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("first publish must succeed"); + gate_tx.send(Ok("2".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let err = broker + .subscribe(EVENT_TYPE, "1") + .expect_err("a durable read failure must not be silently swallowed"); + assert!(matches!(err, EventError::JournalRead(_)), "{err}"); + } + + #[test] + fn poll_on_an_unknown_durable_subscription_id_returns_subscription_not_found() { + let root = test_root("fallback-unknown-poll"); + let journal = open_shared_journal(&root); + let broker = DurableBroker::new( + inner_broker(), + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + let err = broker + .poll(&format!("{DURABLE_SUBSCRIPTION_PREFIX}999"), 10) + .expect_err("unknown durable subscription id must be rejected"); + assert!(matches!(err, EventError::SubscriptionNotFound(_))); + + let cancel_err = broker + .cancel(&format!("{DURABLE_SUBSCRIPTION_PREFIX}999")) + .expect_err("cancelling an unknown durable subscription id must be rejected"); + assert!(matches!(cancel_err, EventError::SubscriptionNotFound(_))); + } + + #[test] + fn poll_with_zero_max_events_on_a_durable_subscription_returns_no_events() { + let root = test_root("fallback-zero-poll"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + broker + .publish(test_event()) + .expect("first publish must succeed"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribe must fall back to the durable journal"); + + let poll = broker + .poll(&subscription.subscription_id, 0) + .expect("poll with max_events=0 must succeed"); + assert!(poll.events.is_empty()); + assert_eq!(poll.cursor, "1", "an unread cursor must not advance"); + } + + #[test] + fn cursor_survives_a_full_broker_restart_at_the_same_journal_root() { + let root = test_root("restart-continuity"); + let cursor_before_restart = { + let broker = DurableBroker::open( + &root, + inner_broker(), + JournalConfig::default(), + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + Arc::new(crate::events::broker::SystemClock), + ) + .expect("durable broker must open"); + + let subscription = broker + .subscribe(EVENT_TYPE, "0") + .expect("subscribe must succeed"); + broker + .publish(test_event()) + .expect("first publish must succeed"); + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("poll must succeed"); + assert_eq!(poll.events.len(), 1); + // A second event that the subscriber never got around to + // consuming before the (simulated) restart, so resuming its + // cursor genuinely has a gap only the durable journal can fill. + broker + .publish(test_event()) + .expect("second publish must succeed"); + poll.cursor + // `broker` and its in-process buffers are dropped here, + // simulating a full process restart: nothing in memory + // survives, only what was fsynced to `root`. + }; + + // Reopen at the same root with an entirely fresh in-memory broker — + // the durable journal is the only thing that persisted. + let restarted = DurableBroker::open( + &root, + inner_broker(), + JournalConfig::default(), + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + Arc::new(crate::events::broker::SystemClock), + ) + .expect("durable broker must reopen at the same root"); + + // The fresh in-memory broker has never seen this event type, so + // without a seeded restart floor it could not tell this cursor is + // behind the durable tip; resuming it must fall back to the + // durable journal and pick up the event missed before restart. + let subscription = restarted + .subscribe(EVENT_TYPE, &cursor_before_restart) + .expect("resuming the pre-restart cursor must succeed after restart"); + assert!( + subscription + .subscription_id + .starts_with(DURABLE_SUBSCRIPTION_PREFIX) + ); + + let poll = restarted + .poll(&subscription.subscription_id, 10) + .expect("poll after restart must succeed"); + assert_eq!( + poll.events.len(), + 1, + "the event published before restart but never consumed must replay" + ); + + restarted + .publish(test_event()) + .expect("publish after restart must succeed"); + let poll = restarted + .poll(&subscription.subscription_id, 10) + .expect("second poll after restart must succeed"); + assert_eq!( + poll.events.len(), + 1, + "the event published after restart must replay from the resumed cursor" + ); + } + + #[test] + fn open_surfaces_an_invalid_journal_config() { + let root = test_root("open-invalid-config"); + let err = DurableBroker::open( + &root, + inner_broker(), + JournalConfig { + max_segment_bytes: 0, + ..JournalConfig::default() + }, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + Arc::new(crate::events::broker::SystemClock), + ) + .err() + .expect("an invalid journal config must not open"); + assert!(matches!(err, JournalError::InvalidConfig(_)), "{err}"); + } + + #[test] + fn subscribe_propagates_non_cursor_errors_from_the_inner_broker_unchanged() { + let root = test_root("subscribe-non-cursor-error"); + let broker = DurableBroker::open( + &root, + inner_broker(), + JournalConfig::default(), + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + Arc::new(crate::events::broker::SystemClock), + ) + .expect("durable broker must open"); + + let err = broker + .subscribe("dev.traverse.test.unregistered", "0") + .expect_err( + "an unregistered event type must be rejected without consulting the journal", + ); + assert!(matches!(err, EventError::UnregisteredEventType(_)), "{err}"); + } + + #[test] + fn durable_poll_returns_no_events_when_already_caught_up_to_the_journal_tip() { + let root = test_root("fallback-caught-up"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + broker + .publish(test_event()) + .expect("first publish must succeed"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribe must fall back to the durable journal"); + + // Immediately caught up: the durable journal has nothing after + // cursor "2" yet, so `replay_from` returns an empty batch on the + // very first loop iteration. + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("first poll must succeed"); + assert_eq!(poll.events.len(), 1, "only the missed event must replay"); + let poll_again = broker + .poll(&subscription.subscription_id, 10) + .expect("second poll must succeed"); + assert!( + poll_again.events.is_empty(), + "nothing new since the last poll must yield an empty batch" + ); + } + + #[test] + fn durable_poll_respects_max_events_and_can_be_paged() { + let root = test_root("fallback-paging"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + for _ in 0..4 { + broker.publish(test_event()).expect("publish must succeed"); + } + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribe must fall back to the durable journal"); + + let first_page = broker + .poll(&subscription.subscription_id, 2) + .expect("first page must succeed"); + assert_eq!( + first_page.events.len(), + 2, + "the page must stop at max_events" + ); + assert_eq!(first_page.cursor, "3"); + + let second_page = broker + .poll(&subscription.subscription_id, 2) + .expect("second page must succeed"); + assert_eq!(second_page.events.len(), 1, "only one event remains"); + assert_eq!(second_page.cursor, "4"); + } + + #[test] + fn durable_poll_skips_records_of_other_event_types() { + const OTHER_EVENT_TYPE: &str = "dev.traverse.test.durable.other"; + let root = test_root("fallback-other-type"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let catalog = active_catalog(); + catalog + .register(EventCatalogEntry { + event_type: OTHER_EVENT_TYPE.to_string(), + version: "1.0.0".to_string(), + owner: "test.capability".to_string(), + lifecycle_status: LifecycleStatus::Active, + consumer_count: 0, + }) + .expect("second catalog entry must register"); + let inner = crate::events::broker::InProcessBroker::with_clock( + catalog, + crate::events::broker::BrokerConfig { + retention_window: Duration::from_millis(10), + max_queue_len: 16, + }, + clock.clone() as Arc, + ) + .expect("broker must build"); + let broker = DurableBroker::new( + inner, + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + broker + .publish(test_event()) + .expect("first publish must succeed"); + let mut other = test_event(); + other.event_type = OTHER_EVENT_TYPE.to_string(); + broker + .publish(other) + .expect("other-type publish must succeed"); + broker + .publish(test_event()) + .expect("third publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribe must fall back to the durable journal"); + let poll = broker + .poll(&subscription.subscription_id, 10) + .expect("poll must succeed"); + assert_eq!( + poll.events.len(), + 1, + "the interleaved other-type record must be skipped" + ); + assert_eq!(poll.events[0].event.event_type, EVENT_TYPE); + } + + #[test] + fn durable_poll_surfaces_cursor_expired_when_the_journal_prunes_mid_subscription() { + struct ExpiresAfterFirstCallSource(std::sync::atomic::AtomicUsize); + + impl JournalSource for ExpiresAfterFirstCallSource { + fn replay_from( + &self, + _cursor: &str, + _max_events: usize, + ) -> Result, JournalError> { + let call = self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if call == 0 { + Ok(Vec::new()) + } else { + Err(JournalError::CursorExpired { + oldest_available_cursor: "9".to_string(), + }) + } + } + } + + let clock = Arc::new(ManualClock::new()); + let (gate_tx, gate_rx) = channel(); + let (revoked_tx, _revoked_rx) = channel(); + let sink = ScriptedSink { + gate: gate_rx, + revocations: revoked_tx, + fail_revocation: false, + }; + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + sink, + Arc::new(ExpiresAfterFirstCallSource( + std::sync::atomic::AtomicUsize::new(0), + )) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + gate_tx.send(Ok("1".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("first publish must succeed"); + gate_tx.send(Ok("2".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribe must fall back to the durable journal (first source call)"); + + let err = broker.poll(&subscription.subscription_id, 10).expect_err( + "the journal pruning this cursor mid-subscription must surface cursor_expired", + ); + assert_eq!( + err, + EventError::CursorExpired { + event_type: EVENT_TYPE.to_string(), + oldest_available_cursor: "9".to_string(), + } + ); + } + + #[test] + fn durable_poll_propagates_a_corrupt_journal_read_as_journal_read() { + struct CorruptSource; + impl JournalSource for CorruptSource { + fn replay_from( + &self, + _cursor: &str, + _max_events: usize, + ) -> Result, JournalError> { + Err(JournalError::Corrupt { + path: "segment-1.jsonl".to_string(), + line: 3, + message: "truncated record".to_string(), + }) + } + } + + let clock = Arc::new(ManualClock::new()); + let (gate_tx, gate_rx) = channel(); + let (revoked_tx, _revoked_rx) = channel(); + let sink = ScriptedSink { + gate: gate_rx, + revocations: revoked_tx, + fail_revocation: false, + }; + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + sink, + Arc::new(CorruptSource) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + gate_tx.send(Ok("1".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("first publish must succeed"); + gate_tx.send(Ok("2".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let err = broker + .subscribe(EVENT_TYPE, "1") + .expect_err("a corrupt journal read must not be silently swallowed"); + assert!( + matches!(&err, EventError::JournalRead(msg) if msg.contains("truncated record")), + "{err}" + ); + } + + #[test] + fn null_source_fails_loudly_instead_of_silently_serving_a_fallback() { + let clock = Arc::new(ManualClock::new()); + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + SharedJournalSink(open_shared_journal(&test_root("null-source-fallback"))), + null_source(), + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + broker + .publish(test_event()) + .expect("first publish must succeed"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + clock.advance(Duration::from_secs(1)); + + let err = broker.subscribe(EVENT_TYPE, "1").expect_err( + "a broker with no real durable source must not silently accept a stale cursor", + ); + assert!(matches!(err, EventError::CursorExpired { .. }), "{err}"); + } + + #[test] + fn map_journal_read_error_translates_invalid_cursor() { + struct InvalidCursorSource; + impl JournalSource for InvalidCursorSource { + fn replay_from( + &self, + _cursor: &str, + _max_events: usize, + ) -> Result, JournalError> { + Err(JournalError::InvalidCursor("not a real cursor".to_string())) + } + } + + let clock = Arc::new(ManualClock::new()); + let (gate_tx, gate_rx) = channel(); + let (revoked_tx, _revoked_rx) = channel(); + let sink = ScriptedSink { + gate: gate_rx, + revocations: revoked_tx, + fail_revocation: false, + }; + let broker = DurableBroker::new( + short_retention_inner(clock.clone()), + sink, + Arc::new(InvalidCursorSource) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + gate_tx.send(Ok("1".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("first publish must succeed"); + gate_tx.send(Ok("2".to_string())).expect("gate must accept"); + broker + .publish(test_event()) + .expect("second publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let err = broker + .subscribe(EVENT_TYPE, "1") + .expect_err("an invalid durable cursor must not be silently accepted"); + assert!(matches!(err, EventError::InvalidCursor(_)), "{err}"); + } + + #[test] + fn shared_journal_sink_revokes_a_durably_written_but_undeliverable_event() { + // Uses the real, production `SharedJournalSink` (via `DurableBroker::open`) + // rather than a scripted test double, so the revocation actually + // exercises the journal-backed write path (067 FR-004). + let root = test_root("shared-sink-revocation"); + let broker = DurableBroker::open( + &root, + inner_broker(), + JournalConfig::default(), + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + Arc::new(crate::events::broker::SystemClock), + ) + .expect("durable broker must open"); + + let mut unregistered = test_event(); + unregistered.event_type = "dev.traverse.test.unknown".to_string(); + let err = broker + .publish(unregistered) + .expect_err("an unregistered event type must be rejected by the inner broker"); + assert!(matches!(err, EventError::UnregisteredEventType(_)), "{err}"); + + // The event was durably written before delivery failed, then + // revoked; a fresh reader over the same root must never see it. + let reader = DurableEventJournal::open( + &root, + JournalConfig::default(), + Arc::new(crate::events::broker::SystemClock), + ) + .expect("journal must reopen"); + let replayed = reader.replay_from("0", 10).expect("replay must succeed"); + assert!( + replayed.is_empty(), + "the revoked, undeliverable event must never replay" + ); + } + + #[test] + fn durable_poll_fetches_additional_batches_when_a_full_batch_matches_nothing() { + const OTHER_EVENT_TYPE: &str = "dev.traverse.test.durable.other-batching"; + let root = test_root("fallback-multi-batch"); + let clock = Arc::new(ManualClock::new()); + let journal = open_shared_journal(&root); + let catalog = active_catalog(); + catalog + .register(EventCatalogEntry { + event_type: OTHER_EVENT_TYPE.to_string(), + version: "1.0.0".to_string(), + owner: "test.capability".to_string(), + lifecycle_status: LifecycleStatus::Active, + consumer_count: 0, + }) + .expect("second catalog entry must register"); + let inner = crate::events::broker::InProcessBroker::with_clock( + catalog, + crate::events::broker::BrokerConfig { + retention_window: Duration::from_millis(10), + max_queue_len: 16, + }, + clock.clone() as Arc, + ) + .expect("broker must build"); + let broker = DurableBroker::new( + inner, + SharedJournalSink(Arc::clone(&journal)), + Arc::clone(&journal) as Arc, + DurableBrokerConfig::default(), + Arc::new(RecordingAudit::default()), + ); + + broker + .publish(test_event()) + .expect("leading publish must succeed"); + // Two consecutive other-type records: with max_events=1, each + // `replay_from` batch is entirely filtered out, forcing the poll + // loop to fetch another batch rather than stopping after one. + let mut other_one = test_event(); + other_one.event_type = OTHER_EVENT_TYPE.to_string(); + broker + .publish(other_one) + .expect("first other-type publish must succeed"); + let mut other_two = test_event(); + other_two.event_type = OTHER_EVENT_TYPE.to_string(); + broker + .publish(other_two) + .expect("second other-type publish must succeed"); + broker + .publish(test_event()) + .expect("trailing publish must succeed"); + + clock.advance(Duration::from_secs(1)); + let subscription = broker + .subscribe(EVENT_TYPE, "1") + .expect("subscribe must fall back to the durable journal"); + + let poll = broker + .poll(&subscription.subscription_id, 1) + .expect("poll must succeed across multiple internal batches"); + assert_eq!(poll.events.len(), 1); + assert_eq!(poll.events[0].event.event_type, EVENT_TYPE); + assert_eq!( + poll.cursor, "4", + "cursor must advance past the skipped batch" + ); + } } diff --git a/crates/traverse-runtime/src/events/journal.rs b/crates/traverse-runtime/src/events/journal.rs index ffd422d..e142cf5 100644 --- a/crates/traverse-runtime/src/events/journal.rs +++ b/crates/traverse-runtime/src/events/journal.rs @@ -344,6 +344,16 @@ impl DurableEventJournal { } } + /// The most recent cursor ever durably assigned, or `0` when nothing has + /// been written yet. [`super::durable::DurableBroker::open`] uses this to + /// seed a freshly constructed in-memory broker's restart floor (spec 066 + /// FR-007), so it correctly defers cursors it cannot itself vouch for to + /// durable replay instead of accepting them by default. + #[must_use] + pub(crate) fn latest_cursor(&self) -> u64 { + self.next_seq.saturating_sub(1) + } + /// Reclaim expired history by deleting whole sealed segments only — never /// rewriting or truncating in place (067 FR-002). A segment is deleted /// only once every event in it falls outside the retention window; the diff --git a/crates/traverse-runtime/src/events/types.rs b/crates/traverse-runtime/src/events/types.rs index f15c19f..a045f35 100644 --- a/crates/traverse-runtime/src/events/types.rs +++ b/crates/traverse-runtime/src/events/types.rs @@ -68,6 +68,8 @@ pub enum EventError { /// Durable journal write exceeded the configured timeout; the event was /// rejected, not delivered (067 FR-003/FR-004). JournalWriteTimeout(String), + /// Durable journal read failed while serving replay (066 FR-005/FR-009). + JournalRead(String), } impl std::fmt::Display for EventError { @@ -87,6 +89,7 @@ impl std::fmt::Display for EventError { Self::InvalidRetentionWindow(msg) => write!(f, "invalid retention window: {msg}"), Self::JournalWrite(msg) => write!(f, "journal write failed: {msg}"), Self::JournalWriteTimeout(msg) => write!(f, "journal_write_timeout: {msg}"), + Self::JournalRead(msg) => write!(f, "journal read failed: {msg}"), } } } @@ -103,6 +106,36 @@ pub trait EventBroker: Send + Sync { /// or [`EventError::LifecycleViolation`] if the catalog entry is not `Active`. fn publish(&self, event: TraverseEvent) -> Result<(), EventError>; + /// Publish an event, threading a pre-assigned durable cursor through to + /// the broker's own cursor space when the implementation supports cursor + /// injection. [`DurableBroker`](super::durable::DurableBroker) uses this + /// to keep live-delivery cursors numerically consistent with the durable + /// journal's cursor space (spec 066 FR-007), so a cursor issued during + /// live polling remains valid when later resolved through durable + /// replay. The default implementation ignores `cursor` and behaves like + /// [`Self::publish`]. + /// + /// # Errors + /// + /// Returns the same errors as [`Self::publish`]. + fn publish_with_cursor(&self, event: TraverseEvent, cursor: &str) -> Result<(), EventError> { + let _ = cursor; + self.publish(event) + } + + /// Establishes a floor below which any cursor, for any event type this + /// broker has not itself observed, is treated as potentially predating + /// this broker instance's own history (spec 066 FR-007: restart cursor + /// continuity). [`DurableBroker::open`](super::durable::DurableBroker::open) + /// calls this once at construction, seeded from the durable journal's + /// latest cursor, so a freshly constructed in-memory broker correctly + /// defers stale-looking cursors to durable replay after a restart + /// instead of accepting them outright. The default implementation is a + /// no-op. + fn seed_restart_floor(&self, floor: u64) { + let _ = floor; + } + /// Create a subscription for the given `event_type` starting from `from_cursor`. /// /// `from_cursor` is an opaque cursor string previously returned by [`poll`](Self::poll). @@ -264,6 +297,7 @@ mod tests { EventError::InvalidRetentionWindow("bad".to_string()), EventError::JournalWrite("disk gone".to_string()), EventError::JournalWriteTimeout("exceeded 2000ms".to_string()), + EventError::JournalRead("disk gone".to_string()), ]; for err in cases {