openquiver-runtime 0.1.0

//! Rebuilding a run from its journal.
//!
//! This is the whole of "resume": there is no saved state to load, because the
//! journal *is* the state. Walking it forward reaches exactly the position the
//! run was in, and the nodes it already passed are not run again — their outputs
//! are read back from what they recorded.

use std::collections::BTreeMap;

use openquiver_compile::CompiledPlan;
use openquiver_graph::{NodeName, PinRef};
use openquiver_store::Event;

use crate::journal::Record;
use crate::{RunValue, RuntimeError};

/// Where execution stands.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Frontier {
    /// Nothing left to do.
    End,
    /// About to run this node.
    At(NodeName),
    /// This node was entered and never left, so whether it happened is unknown.
    InDoubt(NodeName),
}

pub(crate) struct Replayed {
    /// Every output produced so far, by the pin that produced it.
    pub values: BTreeMap<PinRef, RunValue>,
    pub frontier: Frontier,
    /// Whether the run recorded its own end.
    pub finished: bool,
}

/// Walks a journal forward to the position it describes.
///
/// Every shape a journal cannot legitimately have is refused rather than read as
/// best it can be. A journal missing its start reads as a run that finished
/// having done nothing; a journal that continues past its end reads as a run
/// that finished having done half. Both would be recorded as successes.
pub(crate) fn replay(
    events: &[Event],
    plan: &CompiledPlan,
    entry: &NodeName,
) -> Result<Replayed, RuntimeError> {
    let mut values = BTreeMap::new();
    let mut frontier = Frontier::End;
    let mut started = false;
    let mut finished = false;

    for event in events {
        if finished {
            return Err(RuntimeError::BadJournal("it continues after the run ended"));
        }
        match Record::decode(&event.kind, &event.payload)? {
            Record::Started(_) if started => {
                return Err(RuntimeError::BadJournal("it starts twice"));
            }
            Record::Started(start) => {
                // The graph may have been edited while the run was suspended.
                // Replaying against the edited version is silently destructive:
                // a deleted wire stops the run early and reports success.
                if start.plan != plan.fingerprint() {
                    return Err(RuntimeError::DifferentPlan);
                }
                started = true;
                frontier = Frontier::At(entry.clone());
            }
            _ if !started => {
                return Err(RuntimeError::BadJournal(
                    "it does not begin with the start of a run",
                ));
            }
            Record::Entered(entered) => frontier = Frontier::InDoubt(entered.node),
            Record::Exited(exited) => {
                let node = plan
                    .node(exited.node.as_str())
                    .ok_or_else(|| RuntimeError::NotInPlan(exited.node.as_str().to_owned()))?;
                for (pin, value) in exited.outputs {
                    values.insert(PinRef::new(exited.node.clone(), pin), value);
                }
                frontier = match exited.exec.and_then(|pin| node.exec_next.get(&pin)) {
                    Some(next) => Frontier::At(next.clone()),
                    None => Frontier::End,
                };
            }
            // Inert on replay: it is a record of something a person saw, not of
            // anything the run has to know.
            Record::Printed(_) => {}
            Record::Finished => {
                finished = true;
                frontier = Frontier::End;
            }
        }
    }

    if !started {
        return Err(RuntimeError::BadJournal(
            "it does not begin with the start of a run",
        ));
    }

    Ok(Replayed {
        values,
        frontier,
        finished,
    })
}