openquiver_runtime/

lib.rs

//! Running a workflow, durably.
//!
//! A run's state *is* its journal. Nothing is held anywhere else: every node
//! writes what it is about to do before doing it, and what it produced
//! afterwards, and resuming is a walk over those records back to the position
//! the run was in. That inverts the usual role of a log — here it is the
//! authority rather than a commentary — and it is what lets a run outlive the
//! process, the terminal and the machine's next reboot.
//!
//! Long waits are the normal case rather than the exception. An agent turn takes
//! minutes, an approval can take days, and a run that dies with its window would
//! be useless for either.

mod journal;
mod nodes;
mod replay;
mod value;

use std::collections::BTreeMap;

use openquiver_acp::{AcpError, Agents};
use openquiver_compile::{CompiledPlan, Source};
use openquiver_graph::{NodeName, PinRef};
use openquiver_nodes::Recovery;
use openquiver_store::{RunId, RunStatus, Store, StoreError};

use journal::{Entered, Exited, Record, Started};
use nodes::Effects;
use replay::Frontier;

pub use journal::JournalError;
pub use value::RunValue;

/// What one step of a run did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Progress {
    Ran(NodeName),
    /// Execution reached the end and the run was closed.
    Done,
}

/// What to do about a node that was entered but never left.
///
/// The one question a journal cannot answer about itself. Between "it was
/// entered" and "it exited" the node may have created an issue, written a file
/// or paid for an agent turn — and it may have done none of that. Nothing
/// recorded afterwards settles it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InDoubt {
    /// Refuse to guess, and say which node is in question.
    Ask,
    /// Run it again, accepting that its effect may happen twice.
    Retry,
}

#[derive(Debug, thiserror::Error)]
pub enum RuntimeError {
    #[error("the journal could not be reached: {0}")]
    Store(#[from] StoreError),
    #[error(transparent)]
    Agent(#[from] AcpError),
    #[error(transparent)]
    Journal(#[from] JournalError),
    #[error("a journal entry could not be written: {0}")]
    Unwritable(#[from] serde_json::Error),
    #[error("this workflow has no starting event, so there is nothing to run")]
    NoEntry,
    #[error("this workflow has {0} starting events, and a run begins at one")]
    ManyEntries(usize),
    #[error("run `{0}` has already finished")]
    AlreadyFinished(String),
    #[error("this journal cannot be replayed: {0}")]
    BadJournal(&'static str),
    #[error(
        "this run was started on a different version of this workflow; \
         the graph has been edited since"
    )]
    DifferentPlan,
    #[error("this journal mentions `{0}`, which is not a node in this workflow")]
    NotInPlan(String),
    #[error("there is no way to run a `{0}` node")]
    NoSuchNodeType(String),
    #[error("`{node}` needs `{pin}`, but `{from}` has not run and cannot supply it")]
    NotYetRun {
        node: String,
        pin: String,
        from: String,
    },
    #[error("`{node}` needs `{pin}`, and nothing supplies it")]
    NoInput { node: String, pin: String },
    #[error("`{node}` expects {expected} on `{pin}`")]
    WrongValue {
        node: String,
        pin: String,
        expected: &'static str,
    },
    #[error(
        "`{node}` was entered and never left, so whether it finished is unknown; \
         resume again saying whether to run it"
    )]
    InDoubt { node: String },
}

impl RuntimeError {
    /// Whether resuming could ever get past this.
    ///
    /// A run stopped by something that will fail identically next time is
    /// finished, not waiting. Recording it as merely suspended puts it in the
    /// list of runs to pick up, where every attempt appends another
    /// `node.entered` to a journal that will never reach an end — and makes a
    /// permanently broken run indistinguishable from one waiting on a person.
    ///
    /// Matched exhaustively on purpose: a new way to fail must be classified,
    /// not inherit a default.
    fn is_permanent(&self) -> bool {
        match self {
            // The disk, the agent process and the network can all be tried
            // again, and trying again is exactly what resuming is for.
            RuntimeError::Store(_) | RuntimeError::Agent(_) => false,
            RuntimeError::Journal(_)
            | RuntimeError::Unwritable(_)
            | RuntimeError::NoEntry
            | RuntimeError::ManyEntries(_)
            | RuntimeError::AlreadyFinished(_)
            | RuntimeError::BadJournal(_)
            | RuntimeError::DifferentPlan
            | RuntimeError::NotInPlan(_)
            | RuntimeError::NoSuchNodeType(_)
            | RuntimeError::NotYetRun { .. }
            | RuntimeError::NoInput { .. }
            | RuntimeError::WrongValue { .. } => true,
            // Not raised by a running node — it is the answer `resume` gives
            // before a run is under way — so it never reaches this.
            RuntimeError::InDoubt { .. } => true,
        }
    }
}

/// One run in progress.
pub struct Runtime<'a> {
    store: &'a Store,
    plan: &'a CompiledPlan,
    run: RunId,
    effects: Effects<'a>,
    values: BTreeMap<PinRef, RunValue>,
    next: Option<NodeName>,
    finished: bool,
}

impl<'a> Runtime<'a> {
    /// Starts a new run and records that it started.
    pub async fn start(
        store: &'a Store,
        plan: &'a CompiledPlan,
        agents: &'a dyn Agents,
        workflow: &str,
    ) -> Result<Self, RuntimeError> {
        let entry = single_entry(plan)?;
        let run = store.start_run(workflow).await?;

        let runtime = Self {
            store,
            plan,
            run,
            effects: Effects {
                agents,
                sessions: BTreeMap::new(),
            },
            values: BTreeMap::new(),
            next: Some(entry),
            finished: false,
        };
        runtime
            .append(Record::Started(Started {
                plan: plan.fingerprint(),
            }))
            .await?;
        Ok(runtime)
    }

    /// Picks up a run from its journal.
    ///
    /// Nodes that recorded an exit are not run again; their outputs are read
    /// back from what they wrote. `in_doubt` answers the one case that cannot be
    /// settled from the record — and it is only consulted for node types that
    /// say repeating them is unsafe.
    pub async fn resume(
        store: &'a Store,
        plan: &'a CompiledPlan,
        agents: &'a dyn Agents,
        run: RunId,
        in_doubt: InDoubt,
    ) -> Result<Self, RuntimeError> {
        if !store.status(&run).await?.resumable() {
            return Err(RuntimeError::AlreadyFinished(run.as_str().to_owned()));
        }

        let entry = single_entry(plan)?;
        let events = store.events(&run).await?;
        let replayed = replay::replay(&events, plan, &entry)?;

        let next = match replayed.frontier {
            Frontier::End => None,
            Frontier::At(node) => Some(node),
            Frontier::InDoubt(node) => {
                let recovery = plan
                    .node(node.as_str())
                    .ok_or_else(|| RuntimeError::NotInPlan(node.as_str().to_owned()))?
                    .recovery;
                match (recovery, in_doubt) {
                    (Recovery::Retry, _) | (Recovery::Ask, InDoubt::Retry) => Some(node),
                    (Recovery::Ask, InDoubt::Ask) => {
                        return Err(RuntimeError::InDoubt {
                            node: node.as_str().to_owned(),
                        });
                    }
                }
            }
        };

        store.resume(&run).await?;
        Ok(Self {
            store,
            plan,
            run,
            effects: Effects {
                agents,
                sessions: BTreeMap::new(),
            },
            values: replayed.values,
            next,
            finished: replayed.finished,
        })
    }

    pub fn id(&self) -> &RunId {
        &self.run
    }

    /// Runs to the end, or to the first thing that stops it.
    pub async fn run(&mut self) -> Result<(), RuntimeError> {
        while let Progress::Ran(_) = self.step().await? {}
        Ok(())
    }

    /// Runs one node.
    ///
    /// The unit a run is killed and resumed at, and the unit a debugger steps
    /// by — the same primitive, because a breakpoint and a crash leave the
    /// journal in exactly the same shape.
    pub async fn step(&mut self) -> Result<Progress, RuntimeError> {
        let Some(name) = self.next.clone() else {
            self.close().await?;
            return Ok(Progress::Done);
        };

        match self.run_node(&name).await {
            Ok(()) => Ok(Progress::Ran(name)),
            Err(error) => {
                // The journal is left exactly as it is: the entry with no exit
                // is the evidence that this node is in doubt, and erasing it
                // would turn an honest uncertainty into a silent repeat.
                let recorded = if error.is_permanent() {
                    self.store.finish(&self.run, RunStatus::Failed).await
                } else {
                    self.store.suspend(&self.run).await
                };
                if let Err(problem) = recorded {
                    tracing::warn!(
                        run = self.run.as_str(),
                        "could not record how the run stopped: {problem}"
                    );
                }
                Err(error)
            }
        }
    }

    async fn run_node(&mut self, name: &NodeName) -> Result<(), RuntimeError> {
        let node = self
            .plan
            .node(name.as_str())
            .ok_or_else(|| RuntimeError::NotInPlan(name.as_str().to_owned()))?;
        // Before the entry, not after: resolving inputs touches nothing, so a
        // graph that cannot supply a pin should not leave a node looking as if
        // it might have run.
        let inputs = self.resolve(name, node)?;

        self.append(Record::Entered(Entered { node: name.clone() }))
            .await?;
        let outcome = nodes::perform(&mut self.effects, name, node, &inputs).await?;

        for record in outcome.records {
            self.append(record).await?;
        }
        self.append(Record::Exited(Exited {
            node: name.clone(),
            outputs: outcome.outputs.clone(),
            exec: outcome.exec.map(str::to_owned),
        }))
        .await?;

        for (pin, value) in outcome.outputs {
            self.values.insert(PinRef::new(name.clone(), pin), value);
        }
        self.next = outcome
            .exec
            .and_then(|pin| node.exec_next.get(pin).cloned());
        Ok(())
    }

    fn resolve(
        &self,
        name: &NodeName,
        node: &openquiver_compile::CompiledNode,
    ) -> Result<BTreeMap<String, RunValue>, RuntimeError> {
        node.inputs
            .iter()
            .map(|(pin, source)| {
                let value =
                    match source {
                        Source::Literal(literal) => RunValue::from(literal),
                        Source::Pin(from) => self.values.get(from).cloned().ok_or_else(|| {
                            RuntimeError::NotYetRun {
                                node: name.as_str().to_owned(),
                                pin: pin.clone(),
                                from: from.node().to_owned(),
                            }
                        })?,
                    };
                Ok((pin.clone(), value))
            })
            .collect()
    }

    /// Records the end of the run.
    ///
    /// The record is written once; the status is reconciled every time. They are
    /// two writes, and a run killed between them left the journal saying it had
    /// ended and the run row still saying `Running` — resumable for ever, picked
    /// up by every startup, and doing nothing each time.
    async fn close(&mut self) -> Result<(), RuntimeError> {
        if !self.finished {
            // In this order: the journal refuses entries to a finished run, and
            // it is the journal that says the run finished.
            self.append(Record::Finished).await?;
            self.finished = true;
        }
        self.store.finish(&self.run, RunStatus::Finished).await?;
        Ok(())
    }

    async fn append(&self, record: Record) -> Result<(), RuntimeError> {
        self.store
            .append(&self.run, record.kind(), &record.payload()?)
            .await?;
        Ok(())
    }
}

/// The node a run begins at.
///
/// Several event nodes in one graph is a milestone-2 question — the Run panel
/// will offer a choice — and guessing at one now would make which event fires
/// depend on alphabetical order.
fn single_entry(plan: &CompiledPlan) -> Result<NodeName, RuntimeError> {
    match plan.entries.as_slice() {
        [only] => Ok(only.clone()),
        [] => Err(RuntimeError::NoEntry),
        many => Err(RuntimeError::ManyEntries(many.len())),
    }
}