camel_core/lifecycle/adapters/

in_memory.rs

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use async_trait::async_trait;
use camel_api::MetricsCollector;
use tokio::sync::{Mutex, RwLock};

use crate::lifecycle::domain::DomainError;

use crate::lifecycle::application::ports::{
    CommandDedupPort, EventPublisherPort, ProjectionStorePort, RouteRepositoryPort,
    RouteStatusProjection, RuntimeEventJournalPort, RuntimeUnitOfWorkPort,
};
use crate::lifecycle::domain::{RouteRuntimeAggregate, RouteRuntimeState, RuntimeEvent};

#[derive(Default, Clone)]
pub struct InMemoryRouteRepository {
    routes: Arc<RwLock<HashMap<String, RouteRuntimeAggregate>>>,
}

#[async_trait]
impl RouteRepositoryPort for InMemoryRouteRepository {
    async fn load(&self, route_id: &str) -> Result<Option<RouteRuntimeAggregate>, DomainError> {
        let routes = self.routes.read().await;
        Ok(routes.get(route_id).cloned())
    }

    async fn save(&self, aggregate: RouteRuntimeAggregate) -> Result<(), DomainError> {
        let mut routes = self.routes.write().await;
        routes.insert(aggregate.route_id().to_string(), aggregate);
        Ok(())
    }

    async fn save_if_version(
        &self,
        aggregate: RouteRuntimeAggregate,
        expected_version: u64,
    ) -> Result<(), DomainError> {
        let mut routes = self.routes.write().await;
        let route_id = aggregate.route_id().to_string();
        let current = routes.get(&route_id).ok_or_else(|| {
            DomainError::InvalidState(format!(
                "optimistic lock conflict for route '{route_id}': route not found"
            ))
        })?;

        if current.version() != expected_version {
            return Err(DomainError::InvalidState(format!(
                "optimistic lock conflict for route '{route_id}': expected version {expected_version}, actual {}",
                current.version()
            )));
        }

        routes.insert(route_id, aggregate);
        Ok(())
    }

    async fn delete(&self, route_id: &str) -> Result<(), DomainError> {
        let mut routes = self.routes.write().await;
        routes.remove(route_id);
        Ok(())
    }
}

#[derive(Default, Clone)]
pub struct InMemoryProjectionStore {
    statuses: Arc<RwLock<HashMap<String, RouteStatusProjection>>>,
}

#[async_trait]
impl ProjectionStorePort for InMemoryProjectionStore {
    async fn upsert_status(&self, status: RouteStatusProjection) -> Result<(), DomainError> {
        let mut statuses = self.statuses.write().await;
        statuses.insert(status.route_id.clone(), status);
        Ok(())
    }

    async fn get_status(
        &self,
        route_id: &str,
    ) -> Result<Option<RouteStatusProjection>, DomainError> {
        let statuses = self.statuses.read().await;
        Ok(statuses.get(route_id).cloned())
    }

    async fn list_statuses(&self) -> Result<Vec<RouteStatusProjection>, DomainError> {
        let statuses = self.statuses.read().await;
        Ok(statuses.values().cloned().collect())
    }

    async fn remove_status(&self, route_id: &str) -> Result<(), DomainError> {
        let mut statuses = self.statuses.write().await;
        statuses.remove(route_id);
        Ok(())
    }
}

#[derive(Default, Clone)]
pub struct InMemoryEventPublisher {
    events: Arc<RwLock<Vec<RuntimeEvent>>>,
}

impl InMemoryEventPublisher {
    pub async fn snapshot(&self) -> Vec<RuntimeEvent> {
        self.events.read().await.clone()
    }
}

#[async_trait]
impl EventPublisherPort for InMemoryEventPublisher {
    async fn publish(&self, events: &[RuntimeEvent]) -> Result<(), DomainError> {
        let mut stored = self.events.write().await;
        stored.extend(events.iter().cloned());
        Ok(())
    }
}

#[derive(Default, Clone)]
pub struct InMemoryCommandDedup {
    seen: Arc<RwLock<HashSet<String>>>,
}

#[async_trait]
impl CommandDedupPort for InMemoryCommandDedup {
    async fn first_seen(&self, command_id: &str) -> Result<bool, DomainError> {
        let mut seen = self.seen.write().await;
        Ok(seen.insert(command_id.to_string()))
    }

    async fn forget_seen(&self, command_id: &str) -> Result<(), DomainError> {
        let mut seen = self.seen.write().await;
        seen.remove(command_id);
        Ok(())
    }
}

#[derive(Clone)]
pub struct InMemoryRuntimeStore {
    inner: Arc<Mutex<RuntimeStoreState>>,
    journal: Option<Arc<dyn RuntimeEventJournalPort>>,
    /// Shared late-bound metrics handle (dashboard-observability T3.1).
    /// Seeded once by `CamelContextBuilder::build()` — never a fresh
    /// collector — so every projection transition published through this
    /// store emits `MetricsCollector::set_route_state`.
    metrics: Option<Arc<dyn MetricsCollector>>,
}

#[derive(Default)]
struct RuntimeStoreState {
    routes: HashMap<String, RouteRuntimeAggregate>,
    statuses: HashMap<String, RouteStatusProjection>,
    events: Vec<RuntimeEvent>,
    seen: HashSet<String>,
    /// Route ids reconstructed by the most recent journal replay that have not
    /// yet been re-registered in this process. Drained by `take_recovered` so
    /// the first declarative-boot registration adopts the persisted aggregate
    /// while a later same-session duplicate is still rejected.
    recovered: HashSet<String>,
}

impl InMemoryRuntimeStore {
    pub fn with_journal(mut self, journal: Arc<dyn RuntimeEventJournalPort>) -> Self {
        self.journal = Some(journal);
        self
    }

    /// Seed the shared late-bound metrics handle. Called from
    /// `CamelContextBuilder::build()` with the same handle threaded to the
    /// route controller and the runtime bus (no new collector instances).
    pub fn with_metrics(mut self, metrics: Arc<dyn MetricsCollector>) -> Self {
        self.metrics = Some(metrics);
        self
    }

    /// Emit a route-state gauge transition for a projection write. `None`
    /// handle (tests, standalone stores) = no-op.
    fn emit_route_state(&self, route_id: &str, state: &str) {
        if let Some(metrics) = &self.metrics {
            metrics.set_route_state(route_id, state);
        }
    }

    pub async fn snapshot_events(&self) -> Vec<RuntimeEvent> {
        self.inner.lock().await.events.clone()
    }
}

fn upsert_replayed_route(
    state: &mut RuntimeStoreState,
    route_id: &str,
    next_state: RouteRuntimeState,
    status: &str,
    increment_version: bool,
) {
    let current_version = state
        .routes
        .get(route_id)
        .map(|agg| agg.version())
        .unwrap_or(0);
    let next_version = if increment_version {
        current_version.saturating_add(1)
    } else {
        current_version
    };
    state.routes.insert(
        route_id.to_string(),
        RouteRuntimeAggregate::from_snapshot(route_id, next_state, next_version),
    );
    state.statuses.insert(
        route_id.to_string(),
        RouteStatusProjection {
            route_id: route_id.to_string(),
            status: status.to_string(),
        },
    );
}

fn state_label(state: &RouteRuntimeState) -> &'static str {
    match state {
        RouteRuntimeState::Registered => "Registered",
        RouteRuntimeState::Starting => "Starting",
        RouteRuntimeState::Started => "Started",
        RouteRuntimeState::Suspended => "Suspended",
        RouteRuntimeState::Stopping => "Stopping",
        RouteRuntimeState::Stopped => "Stopped",
        RouteRuntimeState::Failed(_) => "Failed",
    }
}

fn apply_replayed_event(state: &mut RuntimeStoreState, event: &RuntimeEvent) {
    match event {
        RuntimeEvent::RouteRegistered { route_id } => {
            state.routes.insert(
                route_id.clone(),
                RouteRuntimeAggregate::new(route_id.clone()),
            );
            state.statuses.insert(
                route_id.clone(),
                RouteStatusProjection {
                    route_id: route_id.clone(),
                    status: "Registered".to_string(),
                },
            );
        }
        RuntimeEvent::RouteRemoved { route_id } => {
            state.routes.remove(route_id);
            state.statuses.remove(route_id);
        }
        _ => {
            // Use the domain's state machine to derive the next state from the event.
            let Some(next_state) = RouteRuntimeAggregate::state_from_event(event) else {
                return;
            };
            let route_id = match event {
                RuntimeEvent::RouteStartRequested { route_id }
                | RuntimeEvent::RouteStarted { route_id }
                | RuntimeEvent::RouteFailed { route_id, .. }
                | RuntimeEvent::RouteStopped { route_id }
                | RuntimeEvent::RouteSuspended { route_id }
                | RuntimeEvent::RouteResumed { route_id }
                | RuntimeEvent::RouteReloaded { route_id } => route_id,
                _ => return,
            };
            let status = state_label(&next_state);
            let increment_version = !matches!(
                (event, state.routes.get(route_id).map(|agg| agg.state())),
                (
                    RuntimeEvent::RouteStarted { .. },
                    Some(RouteRuntimeState::Starting)
                )
            );
            upsert_replayed_route(state, route_id, next_state, status, increment_version);
        }
    }
}

impl Default for InMemoryRuntimeStore {
    fn default() -> Self {
        Self {
            inner: Arc::new(Mutex::new(RuntimeStoreState::default())),
            journal: None,
            metrics: None,
        }
    }
}

#[async_trait]
impl RouteRepositoryPort for InMemoryRuntimeStore {
    async fn load(&self, route_id: &str) -> Result<Option<RouteRuntimeAggregate>, DomainError> {
        let guard = self.inner.lock().await;
        Ok(guard.routes.get(route_id).cloned())
    }

    async fn save(&self, aggregate: RouteRuntimeAggregate) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        guard
            .routes
            .insert(aggregate.route_id().to_string(), aggregate);
        Ok(())
    }

    async fn save_if_version(
        &self,
        aggregate: RouteRuntimeAggregate,
        expected_version: u64,
    ) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        let route_id = aggregate.route_id().to_string();
        let current = guard.routes.get(&route_id).ok_or_else(|| {
            DomainError::InvalidState(format!(
                "optimistic lock conflict for route '{route_id}': route not found"
            ))
        })?;

        if current.version() != expected_version {
            return Err(DomainError::InvalidState(format!(
                "optimistic lock conflict for route '{route_id}': expected version {expected_version}, actual {}",
                current.version()
            )));
        }

        guard.routes.insert(route_id, aggregate);
        Ok(())
    }

    async fn delete(&self, route_id: &str) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        guard.routes.remove(route_id);
        guard.recovered.remove(route_id);
        Ok(())
    }

    async fn take_recovered(&self, route_id: &str) -> Result<bool, DomainError> {
        let mut guard = self.inner.lock().await;
        Ok(guard.recovered.remove(route_id))
    }
}

#[async_trait]
impl ProjectionStorePort for InMemoryRuntimeStore {
    async fn upsert_status(&self, status: RouteStatusProjection) -> Result<(), DomainError> {
        let route_id = status.route_id.clone();
        let state = status.status.clone();
        let mut guard = self.inner.lock().await;
        guard.statuses.insert(status.route_id.clone(), status);
        self.emit_route_state(&route_id, &state);
        Ok(())
    }

    async fn get_status(
        &self,
        route_id: &str,
    ) -> Result<Option<RouteStatusProjection>, DomainError> {
        let guard = self.inner.lock().await;
        Ok(guard.statuses.get(route_id).cloned())
    }

    async fn list_statuses(&self) -> Result<Vec<RouteStatusProjection>, DomainError> {
        let guard = self.inner.lock().await;
        Ok(guard.statuses.values().cloned().collect())
    }

    async fn remove_status(&self, route_id: &str) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        guard.statuses.remove(route_id);
        // Undeployed route: drop its gauge series so a scrape reflects
        // only routes that exist.
        if let Some(metrics) = &self.metrics {
            metrics.clear_route_state(route_id);
        }
        Ok(())
    }
}

#[async_trait]
impl EventPublisherPort for InMemoryRuntimeStore {
    async fn publish(&self, events: &[RuntimeEvent]) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        if let Some(journal) = &self.journal {
            journal.append_batch(events).await?;
        }
        guard.events.extend(events.iter().cloned());
        Ok(())
    }
}

#[async_trait]
impl CommandDedupPort for InMemoryRuntimeStore {
    async fn first_seen(&self, command_id: &str) -> Result<bool, DomainError> {
        let mut guard = self.inner.lock().await;
        if !guard.seen.insert(command_id.to_string()) {
            return Ok(false);
        }

        if let Some(journal) = &self.journal
            && let Err(err) = journal.append_command_id(command_id).await
        {
            guard.seen.remove(command_id);
            return Err(err);
        }

        Ok(true)
    }

    async fn forget_seen(&self, command_id: &str) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        let removed = guard.seen.remove(command_id);
        if removed && let Some(journal) = &self.journal {
            journal.remove_command_id(command_id).await?;
        }
        Ok(())
    }
}

/// Journal-derived boot nonce from the recorded command IDs of the durable
/// dedup store: `1 + max(P)` where `P` collects, for every recorded command
/// ID whose final two `':'`-separated segments both parse as `u64`, its
/// penultimate segment. Tail scanning (right-to-left via `rsplit`) needs no
/// prefix or segment-count parsing, so legacy IDs written for routes whose
/// route IDs contain colons are covered by construction.
///
/// No qualifying recorded ID yields `Ok(0)`. A penultimate value of
/// `u64::MAX` (only reachable via an adversarial recorded command ID)
/// exhausts the deterministic nonce space and fails closed: the error names
/// the offending recorded ID and instructs the operator to clean or rotate
/// the journal.
fn derive_boot_nonce<'a>(ids: impl IntoIterator<Item = &'a str>) -> Result<u64, DomainError> {
    let mut max_penultimate: Option<(u64, &str)> = None;
    for id in ids {
        let mut segments = id.rsplit(':');
        let (Some(last), Some(penultimate)) = (segments.next(), segments.next()) else {
            continue;
        };
        let (Ok(_), Ok(nonce_seg)) = (last.parse::<u64>(), penultimate.parse::<u64>()) else {
            continue;
        };
        if max_penultimate.as_ref().is_none_or(|&(m, _)| nonce_seg > m) {
            max_penultimate = Some((nonce_seg, id));
        }
    }

    match max_penultimate {
        None => Ok(0),
        Some((u64::MAX, offending)) => Err(DomainError::InvalidState(format!(
            "deterministic boot nonce space exhausted by recorded command ID '{offending}'; clean or rotate the journal"
        ))),
        Some((m, _)) => Ok(m + 1),
    }
}

#[async_trait]
impl RuntimeUnitOfWorkPort for InMemoryRuntimeStore {
    async fn persist_upsert(
        &self,
        aggregate: RouteRuntimeAggregate,
        expected_version: Option<u64>,
        projection: RouteStatusProjection,
        events: &[RuntimeEvent],
    ) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        if let Some(expected) = expected_version {
            let route_id = aggregate.route_id().to_string();
            let current = guard.routes.get(&route_id).ok_or_else(|| {
                DomainError::InvalidState(format!(
                    "optimistic lock conflict for route '{route_id}': route not found"
                ))
            })?;
            if current.version() != expected {
                return Err(DomainError::InvalidState(format!(
                    "optimistic lock conflict for route '{route_id}': expected version {expected}, actual {}",
                    current.version()
                )));
            }
        }

        if let Some(journal) = &self.journal {
            journal.append_batch(events).await?;
        }

        guard
            .routes
            .insert(aggregate.route_id().to_string(), aggregate);
        let route_id = projection.route_id.clone();
        let state = projection.status.clone();
        guard
            .statuses
            .insert(projection.route_id.clone(), projection);
        guard.events.extend(events.iter().cloned());
        self.emit_route_state(&route_id, &state);
        Ok(())
    }

    async fn persist_delete(
        &self,
        route_id: &str,
        events: &[RuntimeEvent],
    ) -> Result<(), DomainError> {
        let mut guard = self.inner.lock().await;
        if let Some(journal) = &self.journal {
            journal.append_batch(events).await?;
        }
        guard.routes.remove(route_id);
        guard.statuses.remove(route_id);
        guard.events.extend(events.iter().cloned());
        if let Some(metrics) = &self.metrics {
            metrics.clear_route_state(route_id);
        }
        Ok(())
    }

    async fn recover_from_journal(&self) -> Result<(), DomainError> {
        let Some(journal) = &self.journal else {
            return Ok(());
        };

        let replayed_events = journal.load_all().await?;
        let replayed_command_ids = journal.load_command_ids().await?;

        let mut guard = self.inner.lock().await;
        guard.routes.clear();
        guard.statuses.clear();
        guard.events.clear();
        guard.seen.clear();
        guard.recovered.clear();

        for event in &replayed_events {
            apply_replayed_event(&mut guard, event);
        }
        // Every aggregate that survived replay is adoptable exactly once by the
        // first re-registration of its id (declarative boot), so a durable
        // journal no longer turns a clean restart into an AlreadyExists storm.
        guard.recovered = guard.routes.keys().cloned().collect();
        // Seed the route-state gauges from the recovered projections:
        // replay writes statuses directly, bypassing the emit sites.
        if let Some(metrics) = &self.metrics {
            for status in guard.statuses.values() {
                metrics.set_route_state(&status.route_id, &status.status);
            }
        }
        guard.events = replayed_events;
        for command_id in replayed_command_ids {
            guard.seen.insert(command_id);
        }
        Ok(())
    }

    async fn recovered_boot_nonce(&self) -> Result<u64, DomainError> {
        let guard = self.inner.lock().await;
        // After replay, `seen` holds exactly the replayed durable command IDs.
        derive_boot_nonce(guard.seen.iter().map(String::as_str))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;

    #[derive(Clone)]
    struct ReplayJournal {
        events: Vec<RuntimeEvent>,
    }

    #[async_trait]
    impl RuntimeEventJournalPort for ReplayJournal {
        async fn append_batch(&self, _events: &[RuntimeEvent]) -> Result<(), DomainError> {
            Ok(())
        }

        async fn load_all(&self) -> Result<Vec<RuntimeEvent>, DomainError> {
            Ok(self.events.clone())
        }
    }

    #[tokio::test]
    async fn repo_roundtrip_works() {
        let repo = InMemoryRouteRepository::default();
        repo.save(RouteRuntimeAggregate::new("r1")).await.unwrap();
        assert!(repo.load("r1").await.unwrap().is_some());

        let updated = RouteRuntimeAggregate::from_snapshot(
            "r1",
            crate::lifecycle::domain::RouteRuntimeState::Started,
            1,
        );
        repo.save_if_version(updated.clone(), 0).await.unwrap();
        let loaded = repo.load("r1").await.unwrap().unwrap();
        assert_eq!(loaded.version(), 1);

        let conflict = repo.save_if_version(updated, 0).await.unwrap_err();
        assert!(
            conflict.to_string().contains("optimistic lock conflict"),
            "unexpected conflict error: {conflict}"
        );

        repo.delete("r1").await.unwrap();
        assert!(repo.load("r1").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn projection_roundtrip_works() {
        let store = InMemoryProjectionStore::default();
        store
            .upsert_status(RouteStatusProjection {
                route_id: "r1".into(),
                status: "Started".into(),
            })
            .await
            .unwrap();

        let status = store.get_status("r1").await.unwrap();
        assert!(status.is_some());
        assert_eq!(status.unwrap().status, "Started");
        store.remove_status("r1").await.unwrap();
        assert!(store.get_status("r1").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn event_publisher_stores_events() {
        let publisher = InMemoryEventPublisher::default();
        publisher
            .publish(&[RuntimeEvent::RouteStarted {
                route_id: "r1".into(),
            }])
            .await
            .unwrap();

        let events = publisher.snapshot().await;
        assert_eq!(events.len(), 1);
    }

    #[tokio::test]
    async fn command_dedup_detects_duplicates() {
        let dedup = InMemoryCommandDedup::default();
        assert!(dedup.first_seen("c1").await.unwrap());
        assert!(!dedup.first_seen("c1").await.unwrap());
        dedup.forget_seen("c1").await.unwrap();
        assert!(dedup.first_seen("c1").await.unwrap());
        assert!(dedup.first_seen("c2").await.unwrap());
    }

    #[tokio::test]
    async fn runtime_store_uow_persists_all_three_writes() {
        let store = InMemoryRuntimeStore::default();
        let aggregate = RouteRuntimeAggregate::new("uow-r1");
        let projection = RouteStatusProjection {
            route_id: "uow-r1".to_string(),
            status: "Registered".to_string(),
        };
        let events = vec![RuntimeEvent::RouteRegistered {
            route_id: "uow-r1".to_string(),
        }];

        store
            .persist_upsert(aggregate, None, projection.clone(), &events)
            .await
            .unwrap();

        assert!(store.load("uow-r1").await.unwrap().is_some());
        assert_eq!(
            store.get_status("uow-r1").await.unwrap().unwrap(),
            projection
        );
        assert_eq!(store.snapshot_events().await, events);
    }

    #[tokio::test]
    async fn runtime_store_uow_enforces_expected_version() {
        let store = InMemoryRuntimeStore::default();
        let initial = RouteRuntimeAggregate::new("uow-r2");
        let initial_projection = RouteStatusProjection {
            route_id: "uow-r2".to_string(),
            status: "Registered".to_string(),
        };
        store
            .persist_upsert(
                initial,
                None,
                initial_projection,
                &[RuntimeEvent::RouteRegistered {
                    route_id: "uow-r2".to_string(),
                }],
            )
            .await
            .unwrap();

        let started = RouteRuntimeAggregate::from_snapshot(
            "uow-r2",
            crate::lifecycle::domain::RouteRuntimeState::Started,
            1,
        );
        let err = store
            .persist_upsert(
                started,
                Some(99),
                RouteStatusProjection {
                    route_id: "uow-r2".to_string(),
                    status: "Started".to_string(),
                },
                &[RuntimeEvent::RouteStarted {
                    route_id: "uow-r2".to_string(),
                }],
            )
            .await
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("optimistic lock conflict"),
            "unexpected error: {err}"
        );
    }

    #[tokio::test]
    async fn replay_start_requested_only_advances_version_once() {
        let store = InMemoryRuntimeStore::default().with_journal(Arc::new(ReplayJournal {
            events: vec![
                RuntimeEvent::RouteRegistered {
                    route_id: "replay-r1".to_string(),
                },
                RuntimeEvent::RouteStartRequested {
                    route_id: "replay-r1".to_string(),
                },
            ],
        }));

        store.recover_from_journal().await.unwrap();
        let aggregate = store.load("replay-r1").await.unwrap().unwrap();

        assert_eq!(aggregate.state(), &RouteRuntimeState::Starting);
        assert_eq!(aggregate.version(), 1);
    }

    #[tokio::test]
    async fn replay_start_requested_then_started_keeps_single_command_version() {
        let store = InMemoryRuntimeStore::default().with_journal(Arc::new(ReplayJournal {
            events: vec![
                RuntimeEvent::RouteRegistered {
                    route_id: "replay-r2".to_string(),
                },
                RuntimeEvent::RouteStartRequested {
                    route_id: "replay-r2".to_string(),
                },
                RuntimeEvent::RouteStarted {
                    route_id: "replay-r2".to_string(),
                },
            ],
        }));

        store.recover_from_journal().await.unwrap();
        let aggregate = store.load("replay-r2").await.unwrap().unwrap();

        assert_eq!(aggregate.state(), &RouteRuntimeState::Started);
        assert_eq!(aggregate.version(), 1);
    }

    #[tokio::test]
    async fn recover_from_journal_marks_replayed_routes_adoptable_once() {
        let store = InMemoryRuntimeStore::default().with_journal(Arc::new(ReplayJournal {
            events: vec![
                RuntimeEvent::RouteRegistered {
                    route_id: "replay-r3".to_string(),
                },
                RuntimeEvent::RouteStarted {
                    route_id: "replay-r3".to_string(),
                },
            ],
        }));

        store.recover_from_journal().await.unwrap();

        // First re-registration adopts the recovered aggregate.
        assert!(store.take_recovered("replay-r3").await.unwrap());
        // Marker is consumed: a same-session duplicate is no longer adoptable.
        assert!(!store.take_recovered("replay-r3").await.unwrap());
        // A route that never existed in the journal is never adoptable.
        assert!(!store.take_recovered("never-seen").await.unwrap());
    }

    #[tokio::test]
    async fn recover_from_journal_leaves_no_marker_for_removed_routes() {
        let store = InMemoryRuntimeStore::default().with_journal(Arc::new(ReplayJournal {
            events: vec![
                RuntimeEvent::RouteRegistered {
                    route_id: "replay-r4".to_string(),
                },
                RuntimeEvent::RouteRemoved {
                    route_id: "replay-r4".to_string(),
                },
            ],
        }));

        store.recover_from_journal().await.unwrap();

        // A route removed before the restart is gone from the store and thus
        // not adoptable — a fresh registration proceeds as first-time.
        assert!(store.load("replay-r4").await.unwrap().is_none());
        assert!(!store.take_recovered("replay-r4").await.unwrap());
    }

    #[test]
    fn derive_boot_nonce_empty_is_zero() {
        assert_eq!(derive_boot_nonce([]), Ok(0));
    }

    #[test]
    fn derive_boot_nonce_strictly_above_recorded_penultimates() {
        let ids = [
            "context:start:r:7:0",
            "context:start:r:3:5",
            "context:stop:r:7:1",
        ];
        // 1 + max penultimate 7
        assert_eq!(derive_boot_nonce(ids), Ok(8));
    }

    #[test]
    fn derive_boot_nonce_ignores_non_numeric_tails() {
        // Legacy four-segment IDs whose final-two rule fails: `hello`/`0`
        // (`hello` does not parse) and a lone trailing segment.
        let ids = ["context:start:hello:0", "context:start:foo"];
        assert_eq!(derive_boot_nonce(ids), Ok(0));
    }

    #[test]
    fn derive_boot_nonce_legacy_colon_route_is_forbidden() {
        // Legacy write for route `foo:0`: the tail scan collects penultimate
        // `0`, so the derived nonce must be strictly above it — never `Ok(0)`.
        assert_eq!(derive_boot_nonce(["context:start:foo:0:0"]), Ok(1));
    }

    #[test]
    fn derive_boot_nonce_max_penultimate_fails_closed() {
        let offending = "context:start:r:18446744073709551615:0";
        let err = derive_boot_nonce([offending]).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains(offending),
            "error must name the offending recorded ID verbatim: {msg}"
        );
        assert!(
            msg.contains("clean or rotate the journal"),
            "error must instruct the operator to clean or rotate the journal: {msg}"
        );
    }

    /// Task 3.1 follow-up (holistic review): journal recovery seeds the
    /// route-state gauges from recovered projections — the replay loop
    /// writes statuses directly, bypassing the emit sites, so the seed
    /// after the loop is the only path a recovered route's series has.
    #[tokio::test]
    async fn recovery_seeds_route_state_gauges() {
        struct StateRecorder(std::sync::Mutex<Vec<(String, String)>>);
        impl MetricsCollector for StateRecorder {
            fn increment_errors(&self, _r: &str, _e: &str) {}
            fn record_exchange_duration(&self, _r: &str, _s: std::time::Duration) {}
            fn set_route_state(&self, route: &str, state: &str) {
                self.0
                    .lock()
                    .expect("states lock") // allow-unwrap
                    .push((route.to_string(), state.to_string()));
            }
            fn increment_exchanges(&self, _r: &str) {}
            fn set_queue_depth(&self, _q: &str, _d: usize) {}
            fn record_circuit_breaker_change(&self, _r: &str, _f: &str, _t: &str) {}
        }
        let recorder = Arc::new(StateRecorder(std::sync::Mutex::new(Vec::new())));
        let store = InMemoryRuntimeStore::default()
            .with_metrics(recorder.clone() as Arc<dyn MetricsCollector>)
            .with_journal(Arc::new(ReplayJournal {
                events: vec![
                    RuntimeEvent::RouteRegistered {
                        route_id: "seed-r1".to_string(),
                    },
                    RuntimeEvent::RouteStarted {
                        route_id: "seed-r1".to_string(),
                    },
                ],
            }));

        store.recover_from_journal().await.unwrap();

        let states = recorder.0.lock().expect("states lock").clone();
        assert_eq!(
            states,
            vec![("seed-r1".to_string(), "Started".to_string())],
            "recovery must seed the route-state gauge for each recovered projection"
        );
    }
}