Skip to content

feat(runtime): journal-backed replay through subscribe/poll (#659) #709

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
enricopiovesan merged 1 commit into from
Jul 15, 2026
Merged

feat(runtime): journal-backed replay through subscribe/poll (#659) #709

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
306 changes: 227 additions & 79 deletions crates/traverse-runtime/src/events/broker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ struct BrokerState {
buffers: HashMap<String, VecDeque<BufferedEvent>>,
seen_event_ids: HashMap<String, HashSet<String>>,
subscriptions: HashMap<SubscriptionId, SubscriptionState>,
/// See [`EventBroker::seed_restart_floor`].
restart_floor: u64,
}

/// Synchronous, in-memory implementation of [`EventBroker`].
Expand Down Expand Up @@ -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<u64>,
) -> 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<u64, EventError> {
Expand Down Expand Up @@ -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()
{
Expand Down Expand Up @@ -299,92 +404,36 @@ 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
///
/// - [`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`.
Expand Down Expand Up @@ -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<Subscription, EventError> {
self.0.subscribe(event_type, from_cursor)
}
fn subscribe_for_subject(
&self,
event_type: &str,
from_cursor: &str,
subject_id: Option<&str>,
) -> Result<Subscription, EventError> {
self.0
.subscribe_for_subject(event_type, from_cursor, subject_id)
}
fn poll(
&self,
subscription_id: &str,
max_events: usize,
) -> Result<SubscriptionPoll, EventError> {
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);
Expand Down
Loading
Loading