aion-rs 0.31.0

//! What a boot's recovery pass cost, said in the log: the in-flight listing
//! (row-first, so it names how long "who is in flight?" took without
//! opening finished histories) and the replay tally (how many histories, how
//! many events, how long — each history's length at `debug` so the next
//! measurement points at the fat workflows instead of guessing).

use std::collections::HashSet;

use aion_core::{Event, WorkflowId};
use aion_store::EventStore;

use crate::EngineError;

/// The in-flight set a boot (or a shard adoption) brings back: the store's
/// active and paused lists, de-duplicated, with the listing's cost logged.
pub(super) async fn list_recoverable(
    store: &dyn EventStore,
) -> Result<Vec<WorkflowId>, EngineError> {
    let started = std::time::Instant::now();
    let mut recoverable = store.list_active().await?;
    recoverable.extend(store.list_paused().await?);
    let mut seen = HashSet::new();
    recoverable.retain(|workflow_id| seen.insert(workflow_id.clone()));
    tracing::info!(
        candidates = recoverable.len(),
        elapsed_ms = elapsed_ms(started),
        "startup recovery: in-flight workflows listed"
    );
    Ok(recoverable)
}

/// The replay pass's tally. It reports when dropped, so the summary line is
/// written on the error exit as well as the success one — a boot that fails
/// half-way still says what it had read by then.
pub(super) struct ReplayTally {
    started: std::time::Instant,
    histories_read: usize,
    events_read: usize,
}

impl ReplayTally {
    pub(super) fn start() -> Self {
        Self {
            started: std::time::Instant::now(),
            histories_read: 0,
            events_read: 0,
        }
    }

    pub(super) fn record(&mut self, workflow_id: &WorkflowId, history: &[Event]) {
        self.histories_read += 1;
        self.events_read += history.len();
        tracing::debug!(
            workflow_id = %workflow_id,
            events = history.len(),
            "startup recovery: history read"
        );
    }
}

impl Drop for ReplayTally {
    fn drop(&mut self) {
        tracing::info!(
            histories_read = self.histories_read,
            events_read = self.events_read,
            elapsed_ms = elapsed_ms(self.started),
            "startup recovery: resident workflows repopulated"
        );
    }
}

/// Whole milliseconds since `started`, saturating so a log field never panics.
pub(crate) fn elapsed_ms(started: std::time::Instant) -> u64 {
    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
}