onepipeline 0.22.2

//! Cross-DAG edges: `run:<run_id>#<node_id>`.
//!
//! A dependency that names another run is resolved by **reading that run's
//! ledger**, not this graph. Every unknown resolves toward *blocked* rather than
//! failed — an unknown run, a node that has not settled, and a node that settled
//! badly are all upstreams that may still arrive, and failing the consumer would
//! throw away a graph that is merely early.
//!
//! Once an upstream does arrive, the consumer records **how far that run had
//! got** when it did. If the upstream moves past that point afterwards, the
//! consumer reports it and does not re-run: the work it did was correct when it
//! was done, and whether it should be done again is the planner's judgement, not
//! this crate's.
//!
//! Reading is unlocked, exactly as every other reader of a journal is. A loop
//! that observes an upstream mid-write sees a prefix of it, which resolves
//! toward blocked and is re-read on the next pass.

// llmlint: ignore-file[invalid_states_unrepresentable] a dependency string, a run id, and
// a node id are the identifiers the plan schema spells and the journal payload carries;
// they are the same plain strings everywhere else in this crate, for the reason
// `src/plan.rs` records.

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Instant, SystemTime};

use serde_json::{json, Value};

use crate::error::Result;
use crate::graph::{Graph, NodeStatus};
use crate::journal::{self, Journal};
use crate::ledger::{self, RunPaths};

/// The prefix a cross-DAG reference starts with.
pub const PREFIX: &str = "run:";

/// The shape a reference must have, for the refusal to say so.
pub const SYNTAX: &str = "run:<run_id>#<node_id>";

/// One parsed `run:<run_id>#<node_id>`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Reference {
    /// The run whose ledger answers this edge.
    pub run: String,
    /// The node within it.
    pub node: String,
}

/// Parse a reference, or `None` if this dependency does not name another run.
///
/// Both halves must be non-empty: `run:#build` and `run:other#` name nothing
/// that could ever resolve, so they are malformed rather than merely pending.
///
/// The run half must also be a *run id* rather than a path. A plan is external
/// input and this half is joined onto the runs root to find the ledger that
/// answers the edge, so `run:../../elsewhere#node` would otherwise schedule this
/// run against a ledger outside the root it was pointed at. Refusing it here
/// makes the plan's own reading refuse it by name, through the
/// malformed-reference path that already exists.
pub fn parse(dependency: &str) -> Option<Reference> {
    let rest = dependency.strip_prefix(PREFIX)?;
    let (run, node) = rest.split_once('#')?;
    if node.is_empty() || !ledger::is_valid_run_id(run) {
        return None;
    }
    Some(Reference {
        run: run.to_string(),
        node: node.to_string(),
    })
}

/// Whether a dependency is a well-formed cross-DAG reference.
pub fn is_reference(dependency: &str) -> bool {
    parse(dependency).is_some()
}

/// Whether a dependency was *meant* to be one and is malformed.
///
/// Anything starting `run:` names no node of this graph, so reporting it as a
/// missing dependency would send a planner looking for a node they never wrote.
pub fn is_malformed(dependency: &str) -> bool {
    dependency.starts_with(PREFIX) && parse(dependency).is_none()
}

/// How far a run's ledger has got.
///
/// The count of records in its merged store. The journal is append-only, so this
/// only ever rises, and it rises whenever the upstream does anything at all —
/// which is exactly the question a watch asks. It is deliberately not a per-
/// stream `seq`: a run is written by more than one process, so no single stream's
/// sequence describes the run.
pub fn extent(root: &Path, run: &str) -> Option<u64> {
    let paths = RunPaths::under(root, run);
    if !paths.exists() {
        return None;
    }
    crate::loopstats::upstream_read();
    Some(ledger::read_lines(&paths.journal()).len() as u64)
}

/// How a node of another run last settled, as that run's ledger records it.
///
/// The *last* settlement wins: a node that failed once and succeeded in
/// a later one is done, which is the whole point of a planner retrying it.
/// A record this build cannot read is skipped, the same way every other reader
/// of a journal skips one.
/// The loop's own answer: read now, whatever was read before.
///
/// Never served from what is remembered below, because the loop is what the
/// freshness of a cross-DAG edge is promised against. Serving it a remembered
/// answer would add that answer's age to the interval the loop already waits, so
/// an upstream settling just after a read would go unseen for up to twice the
/// interval — and a consumer's own bound is one interval's worth of that. What
/// this does do is *refresh* what the quiet readers see, which is what makes
/// theirs free.
fn settled_status(root: &Path, reference: &Reference) -> Option<NodeStatus> {
    let status = read_settled_status(root, reference);
    remember(root, reference, status);
    status
}

/// The same answer for a reader that must not write, which may be one the loop
/// has already paid for.
///
/// Accepted up to twice [`UPSTREAM_EVERY`] old, which is deliberately looser than
/// the loop's own cadence rather than equal to it: the loop refreshes on that
/// cadence, so a window of exactly it would expire in the moment before each
/// refresh and buy one of these readers its own read for the gap. What it costs
/// is that a rendered row, or an edit's submission check, may name a cross-DAG
/// status a second behind the run's — where before it cost another run's whole
/// ledger, read again, for every record appended to this one's journal.
///
/// [`UPSTREAM_EVERY`]: crate::engine::UPSTREAM_EVERY
fn settled_status_recently(root: &Path, reference: &Reference) -> Option<NodeStatus> {
    if let Some(answered) = answered_within(root, reference, 2 * crate::engine::UPSTREAM_EVERY) {
        return answered;
    }
    settled_status(root, reference)
}

/// What this process has already read an upstream's ledger to learn, and when.
///
/// More than one part of a driver asks this same question. The reconcile loop
/// asks it on its own paced deadline, and every reader that resolves a graph's
/// edges quietly asks it beside them — a view rendering a row, and the summary
/// document's writer, which is asked once per record appended to a journal. Each
/// answer costs a read of **another run's whole ledger**, so two consumers asking
/// inside one interval is that ledger read twice over for one answer, and the
/// second read is one nobody chose to spend.
///
/// Filled by every read, and read back only by
/// [`settled_status_recently`] — the loop reads through [`settled_status`] and is
/// never served from here, so nothing this holds can make the answer the loop
/// acts on any older than its own interval already allows.
static ANSWERED: Mutex<Answers> = Mutex::new(BTreeMap::new());

/// One upstream node, under the runs root it was asked beneath.
type Asked = (PathBuf, String, String);

/// What that question answered, and when it was paid for.
type Answers = BTreeMap<Asked, (Instant, Option<NodeStatus>)>;

fn key(root: &Path, reference: &Reference) -> Asked {
    (
        root.to_path_buf(),
        reference.run.clone(),
        reference.node.clone(),
    )
}

/// A lock another thread poisoned still holds a sound answer: every writer here
/// replaces one whole entry, so nothing can be observed half-written.
fn answers() -> std::sync::MutexGuard<'static, Answers> {
    ANSWERED.lock().unwrap_or_else(|held| held.into_inner())
}

fn answered_within(
    root: &Path,
    reference: &Reference,
    age: std::time::Duration,
) -> Option<Option<NodeStatus>> {
    answers()
        .get(&key(root, reference))
        .filter(|(read_at, _)| read_at.elapsed() < age)
        .map(|(_, status)| *status)
}

fn remember(root: &Path, reference: &Reference, status: Option<NodeStatus>) {
    let mut answers = answers();
    // Dropped as they go stale rather than on a size, so what is held is one
    // entry per upstream a live graph is actually asking about.
    answers.retain(|_, (read_at, _)| read_at.elapsed() < 2 * crate::engine::UPSTREAM_EVERY);
    answers.insert(key(root, reference), (Instant::now(), status));
}

fn read_settled_status(root: &Path, reference: &Reference) -> Option<NodeStatus> {
    let paths = RunPaths::under(root, &reference.run);
    if !paths.exists() {
        return None;
    }
    crate::loopstats::upstream_read();
    ledger::read_lines(&paths.journal())
        .iter()
        .filter_map(|line| serde_json::from_str::<Value>(line).ok())
        .filter(|event| {
            event.get("kind").and_then(Value::as_str)
                == Some(journal::PipelineKind::NodeSettled.as_str())
        })
        .filter(|event| {
            event
                .get("labels")
                .and_then(|l| l.get("node"))
                .and_then(Value::as_str)
                == Some(reference.node.as_str())
        })
        .filter_map(|event| {
            event
                .get("payload")
                .and_then(|p| p.get("status"))
                .and_then(Value::as_str)
                .and_then(NodeStatus::parse)
        })
        .next_back()
}

/// Every well-formed cross-DAG reference a graph names, with the nodes naming it.
pub fn edges(graph: &Graph) -> BTreeMap<String, Vec<String>> {
    let mut edges: BTreeMap<String, Vec<String>> = BTreeMap::new();
    for node in graph.iter() {
        for dep in &node.deps {
            if is_reference(dep) {
                edges.entry(dep.clone()).or_default().push(node.id.clone());
            }
        }
    }
    edges
}

/// Resolves this run's cross-DAG edges and remembers what it has already said.
///
/// The memory is the run's **own journal**, not this process: a watch outlives
/// the process that captured it — an `adopt` starts a fresh one — so a baseline
/// held only in memory would be re-captured, and the report re-sent, every time
/// a driver restarted.
#[derive(Debug)]
pub struct Observer {
    root: PathBuf,
    /// Where each resolved upstream had got when it was first resolved.
    baselines: BTreeMap<String, u64>,
    /// The `(dependency, consumer)` pairs already reported as moved.
    reported: BTreeSet<(String, String)>,
}

impl Observer {
    /// An observer seeded from what this run has already recorded.
    pub fn new(
        root: &Path,
        baselines: BTreeMap<String, u64>,
        reported: BTreeSet<(String, String)>,
    ) -> Self {
        Self {
            root: root.to_path_buf(),
            baselines,
            reported,
        }
    }

    /// The observer a run's own folded state describes.
    pub fn of_run(paths: &RunPaths, state: &crate::projection::RunState) -> Self {
        let root = paths
            .dir
            .parent()
            .map_or_else(ledger::runs_root, Path::to_path_buf);
        Self::new(
            &root,
            state.cross_dag_baselines.clone(),
            state.cross_dag_reported.clone(),
        )
    }

    /// A cheap look at every upstream ledger this graph's edges are answered by.
    ///
    /// Length and modification time, and no read at all. A run's ledger is
    /// append-only and its length *is* the extent this observer measures against,
    /// so a `stat` answers the only question a re-read could: has that run done
    /// anything since we last looked? The reconcile loop waits on this, so an
    /// upstream that has not moved never *wakes* it — which is what keeps a
    /// consumer watching a quiet upstream from paying a pass for it. What still
    /// re-reads that ledger is the loop's own paced deadline, `UPSTREAM_EVERY`,
    /// because the freshness the edge is promised is a bound on how stale the
    /// answer may be rather than on how quiet the upstream is.
    pub fn marks(&self, graph: &Graph) -> BTreeMap<String, Option<(u64, SystemTime)>> {
        edges(graph)
            .into_keys()
            .filter_map(|dependency| parse(&dependency))
            .map(|reference| {
                let journal = RunPaths::under(&self.root, &reference.run).journal();
                let mark = std::fs::metadata(&journal).ok().map(|metadata| {
                    (
                        metadata.len(),
                        metadata.modified().unwrap_or(std::time::UNIX_EPOCH),
                    )
                });
                (reference.run, mark)
            })
            .collect()
    }

    /// Resolve every edge the graph names, recording what it learns.
    ///
    /// Returns the status each *dependency* resolved to, which is what the
    /// scheduler asks about a reference. Emits at most one `cross-dag-satisfied`
    /// per edge and one `upstream-modified` per edge and consumer, ever.
    pub fn resolve(
        &mut self,
        graph: &Graph,
        paths: &RunPaths,
        journal: &mut Journal,
    ) -> Result<BTreeMap<String, NodeStatus>> {
        let mut resolved = BTreeMap::new();
        for (dependency, consumers) in edges(graph) {
            let Some(reference) = parse(&dependency) else {
                continue;
            };
            let status = settled_status(&self.root, &reference);
            // Only `done` satisfies. Everything else — an unknown run, a node
            // that has not settled, one that failed or was skipped — leaves the
            // consumer waiting on an upstream that may still arrive.
            if status != Some(NodeStatus::Done) {
                resolved.insert(dependency, NodeStatus::Blocked);
                continue;
            }
            resolved.insert(dependency.clone(), NodeStatus::Done);

            let Some(extent) = extent(&self.root, &reference.run) else {
                continue;
            };
            let baseline = match self.baselines.get(&dependency) {
                Some(baseline) => *baseline,
                None => {
                    self.baselines.insert(dependency.clone(), extent);
                    // Recorded against the first consumer, so the baseline has a
                    // node to belong to in a stream every view reads by node.
                    if let Some(first) = consumers.first() {
                        journal.emit(
                            journal::PipelineKind::CrossDagSatisfied,
                            journal::labels(&paths.run, Some(first)),
                            journal::payload(&[
                                ("dependency", json!(dependency)),
                                ("last_seq", json!(extent)),
                            ]),
                        )?;
                    }
                    extent
                }
            };
            if extent <= baseline {
                continue;
            }
            for consumer in consumers {
                let pair = (dependency.clone(), consumer.clone());
                if !self.reported.insert(pair) {
                    continue;
                }
                // Reported, never acted on: the consumer's work was correct when
                // it was done, and whether it should be done again is the
                // planner's call.
                journal.emit(
                    journal::PipelineKind::UpstreamModified,
                    journal::labels(&paths.run, Some(&consumer)),
                    journal::payload(&[
                        ("dependency", json!(dependency)),
                        ("captured_last_seq", json!(baseline)),
                        ("observed_last_seq", json!(extent)),
                    ]),
                )?;
            }
        }
        Ok(resolved)
    }
}

/// Resolve a graph's edges without recording anything.
///
/// For the readers that must not write — a view, and the submission check an
/// edit is judged against. Both need the same answer the loop would get; neither
/// may append to the journal.
pub fn resolve_quietly(root: &Path, graph: &Graph) -> BTreeMap<String, NodeStatus> {
    edges(graph)
        .into_keys()
        .filter_map(|dependency| {
            let reference = parse(&dependency)?;
            let status = match settled_status_recently(root, &reference) {
                Some(NodeStatus::Done) => NodeStatus::Done,
                _ => NodeStatus::Blocked,
            };
            Some((dependency, status))
        })
        .collect()
}

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

    #[test]
    fn a_reference_needs_both_halves() {
        assert_eq!(
            parse("run:other#build"),
            Some(Reference {
                run: "other".into(),
                node: "build".into()
            })
        );
        // A node id may itself address a step within its node.
        assert_eq!(
            parse("run:other#ship/verify").map(|r| r.node),
            Some("ship/verify".to_string())
        );
        for malformed in [
            "run:other",
            "run:#build",
            "run:other#",
            "run:",
            "run:#",
            // A path, not a run id. The ledger that answers an edge is found by
            // joining this onto the runs root.
            "run:../elsewhere#build",
            "run:../../elsewhere#build",
            "run:a/b#build",
            "run:/absolute#build",
            "run:.#build",
            "run:..#build",
        ] {
            assert_eq!(parse(malformed), None, "{malformed} parsed");
            assert!(is_malformed(malformed), "{malformed} is not reported wrong");
        }
        // Not a reference at all, so not this module's business either way.
        assert_eq!(parse("build"), None);
        assert!(!is_malformed("build"));
    }

    #[test]
    fn an_unknown_run_has_no_extent_and_no_status() {
        let root =
            std::env::temp_dir().join(format!("onepipeline-crossdag-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).expect("a scratch root");
        assert_eq!(extent(&root, "nobody"), None);
        assert_eq!(
            settled_status(
                &root,
                &Reference {
                    run: "nobody".into(),
                    node: "build".into()
                }
            ),
            None
        );
        let _ = std::fs::remove_dir_all(&root);
    }
}