harn-vm 0.10.155

//! Checkpoint system for resilient pipeline execution.
//!
//! Provides durable checkpoint reads, serialized mutations, and atomic insertion.
//! Checkpoints are persisted to `<state-root>/checkpoints/<pipeline>.json`
//! and survive pipeline crashes/timeouts. On resume, a pipeline can skip
//! already-processed items by checking `checkpoint_get`.
//!
//! The per-pipeline state is per-thread: `register_checkpoint_builtins`
//! installs a fresh [`CheckpointState`] into a thread-local cell so the
//! `#[harn_builtin]`-emitted handler fns can read/mutate it without
//! capturing closures (which would block the macro path). Each VM that
//! registers checkpoint builtins overrides the cell for its thread —
//! since the Harn VM is single-threaded per execution, this is the
//! intended scoping (each pipeline run installs its own state once at
//! VM setup, then the handlers see it for the duration of the run).

use std::cell::RefCell;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use crate::stdlib::macros::{harn_builtin, VmBuiltinDef};
use crate::value::{VmError, VmValue};
use crate::vm::Vm;

struct CheckpointState {
    data: BTreeMap<String, serde_json::Value>,
    path: PathBuf,
}

impl CheckpointState {
    fn new(base_dir: &Path, pipeline_name: &str) -> Self {
        Self::at_state_root(&crate::runtime_paths::state_root(base_dir), pipeline_name)
    }

    fn at_state_root(state_root: &Path, pipeline_name: &str) -> Self {
        Self {
            data: BTreeMap::new(),
            path: state_root
                .join("checkpoints")
                .join(format!("{pipeline_name}.json")),
        }
    }

    fn load(&mut self) -> Result<(), String> {
        self.data = match std::fs::read_to_string(&self.path) {
            // serde's own message can quote stored values, so report only
            // the failure class and position.
            Ok(contents) => serde_json::from_str(&contents).map_err(|e| {
                let reason = match e.classify() {
                    serde_json::error::Category::Data => "not a JSON object",
                    _ => "malformed JSON",
                };
                format!(
                    "checkpoint decode error in {}: {reason} at line {}, column {}",
                    self.path.display(),
                    e.line(),
                    e.column()
                )
            })?,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                self.confirm_absent()?;
                BTreeMap::new()
            }
            Err(e) => return Err(format!("checkpoint read error: {e}")),
        };
        Ok(())
    }

    /// Accept "not found" as an empty store only when the path is genuinely
    /// absent beneath a directory.
    ///
    /// Unix reports a non-directory ancestor as `ENOTDIR`, but Windows reports
    /// it as `ERROR_PATH_NOT_FOUND`, which std maps to `NotFound` — the same
    /// kind a missing `checkpoints/` directory produces. Without this check a
    /// store that cannot be reached reads as empty on Windows, and a resumed
    /// pipeline would redo work its checkpoint had recorded as spent.
    fn confirm_absent(&self) -> Result<(), String> {
        for ancestor in self.path.ancestors().skip(1) {
            match std::fs::metadata(ancestor) {
                Ok(metadata) if metadata.is_dir() => return Ok(()),
                Ok(_) => {
                    return Err(format!(
                        "checkpoint read error: {} is not a directory",
                        ancestor.display()
                    ))
                }
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
                Err(e) => return Err(format!("checkpoint read error: {e}")),
            }
        }
        Ok(())
    }

    fn lock(&self) -> Result<std::fs::File, String> {
        let path = self.path.with_extension("lock");
        std::fs::create_dir_all(path.parent().expect("checkpoint path has a parent"))
            .map_err(|error| format!("checkpoint lock directory error: {error}"))?;
        let file = std::fs::OpenOptions::new()
            .create(true)
            .truncate(false)
            .read(true)
            .write(true)
            .open(&path)
            .map_err(|error| format!("checkpoint lock open error: {error}"))?;
        harn_flock::lock_with_deadline(
            &file,
            &path,
            harn_flock::LockMode::Exclusive,
            std::time::Duration::from_secs(5),
        )
        .map_err(|error| format!("checkpoint lock error: {error}"))?;
        Ok(file)
    }

    fn save(&self) -> Result<(), String> {
        let json = serde_json::to_string_pretty(&self.data)
            .map_err(|e| format!("checkpoint save error: {e}"))?;
        crate::atomic_io::atomic_write(&self.path, json.as_bytes())
            .map_err(|e| format!("checkpoint write error: {e}"))?;
        Ok(())
    }

    fn get(&mut self, key: &str) -> Result<VmValue, String> {
        self.load()?;
        Ok(match self.data.get(key) {
            Some(v) => json_to_vm(v),
            None => VmValue::Nil,
        })
    }

    fn set(&mut self, key: String, value: serde_json::Value) -> Result<(), String> {
        let _lock = self.lock()?;
        self.load()?;
        self.data.insert(key, value);
        self.persist()
    }

    fn insert(&mut self, key: String, value: serde_json::Value) -> Result<VmValue, String> {
        let _lock = self.lock()?;
        self.load()?;
        let inserted = !self.data.contains_key(&key);
        if inserted {
            self.data.insert(key.clone(), value);
            self.persist()?;
        }
        Ok(VmValue::dict(BTreeMap::from([
            ("inserted".to_owned(), VmValue::Bool(inserted)),
            ("value".to_owned(), json_to_vm(&self.data[&key])),
        ])))
    }

    fn clear(&mut self) -> Result<(), String> {
        let _lock = self.lock()?;
        match std::fs::remove_file(&self.path) {
            Ok(()) => (),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => (),
            Err(e) => return Err(format!("checkpoint clear error: {e}")),
        }
        self.data.clear();
        Ok(())
    }

    fn list(&mut self) -> Result<Vec<String>, String> {
        self.load()?;
        Ok(self.data.keys().cloned().collect())
    }

    fn exists(&mut self, key: &str) -> Result<bool, String> {
        self.load()?;
        Ok(self.data.contains_key(key))
    }

    fn delete(&mut self, key: &str) -> Result<(), String> {
        let _lock = self.lock()?;
        self.load()?;
        self.data.remove(key);
        self.persist()
    }

    fn persist(&mut self) -> Result<(), String> {
        let result = self.save();
        if result.is_err() {
            // A failed write is not a successful in-memory checkpoint.
            // Reload the durable owner before accepting another operation.
            self.data.clear();
        }
        result
    }
}

thread_local! {
    /// Active checkpoint state for the current thread's pipeline run.
    /// Set by `register_checkpoint_builtins`; read by the
    /// `#[harn_builtin]` handler fns. `None` before the first install
    /// (i.e. when the checkpoint builtins haven't been registered for
    /// this VM context) — in that case the handlers return a clear
    /// runtime error rather than a panic.
    static CHECKPOINT_STATE: RefCell<Option<CheckpointState>> = const { RefCell::new(None) };
}

fn with_state<R>(
    fn_name: &'static str,
    f: impl FnOnce(&mut CheckpointState) -> Result<R, VmError>,
) -> Result<R, VmError> {
    CHECKPOINT_STATE.with(|cell| {
        let mut guard = cell.borrow_mut();
        let state = guard.as_mut().ok_or_else(|| {
            VmError::Runtime(format!(
                "{fn_name}: checkpoint builtins not registered for this VM"
            ))
        })?;
        f(state)
    })
}

use crate::value::vm_to_storage_json as vm_to_json;

fn json_to_vm(jv: &serde_json::Value) -> VmValue {
    match jv {
        serde_json::Value::Null => VmValue::Nil,
        serde_json::Value::Bool(b) => VmValue::Bool(*b),
        serde_json::Value::Number(n) => {
            if let Some(i) = n.as_i64() {
                VmValue::Int(i)
            } else {
                VmValue::Float(n.as_f64().unwrap_or(0.0))
            }
        }
        serde_json::Value::String(s) => VmValue::String(arcstr::ArcStr::from(s.as_str())),
        serde_json::Value::Array(arr) => {
            VmValue::List(std::sync::Arc::new(arr.iter().map(json_to_vm).collect()))
        }
        serde_json::Value::Object(map) => {
            let mut m = BTreeMap::new();
            for (k, v) in map {
                m.insert(k.clone(), json_to_vm(v));
            }
            VmValue::dict(m)
        }
    }
}

/// Sanitize a pipeline name for use as a filename.
/// Rejects path traversal attempts and invalid characters.
fn sanitize_pipeline_name(name: &str) -> String {
    let base = std::path::Path::new(name)
        .file_name()
        .and_then(|f| f.to_str())
        .unwrap_or("default");
    if base.is_empty() || base == "." || base == ".." {
        return "default".to_string();
    }
    base.to_string()
}

/// Register checkpoint builtins on a VM.
///
/// The pipeline name is used to namespace checkpoint files. If not provided,
/// defaults to "default". State is installed into a thread-local cell that
/// the `#[harn_builtin]`-emitted handlers below read; subsequent calls on
/// the same thread overwrite the state for that thread (the Harn VM
/// executes single-threaded per run).
pub fn register_checkpoint_builtins(vm: &mut Vm, base_dir: &Path, pipeline_name: &str) {
    let safe_name = sanitize_pipeline_name(pipeline_name);
    register_checkpoint_state(vm, CheckpointState::new(base_dir, &safe_name));
}

pub(crate) fn register_checkpoint_builtins_at_state_root(
    vm: &mut Vm,
    state_root: &Path,
    pipeline_name: &str,
) {
    let safe_name = sanitize_pipeline_name(pipeline_name);
    register_checkpoint_state(vm, CheckpointState::at_state_root(state_root, &safe_name));
}

fn register_checkpoint_state(vm: &mut Vm, state: CheckpointState) {
    CHECKPOINT_STATE.with(|cell| {
        *cell.borrow_mut() = Some(state);
    });
    for def in MODULE_BUILTINS {
        vm.register_builtin_def(def);
    }
}

pub(crate) const MODULE_BUILTINS: &[&VmBuiltinDef] = &[
    &CHECKPOINT_IMPL_DEF,
    &CHECKPOINT_INSERT_IMPL_DEF,
    &CHECKPOINT_GET_IMPL_DEF,
    &CHECKPOINT_CLEAR_IMPL_DEF,
    &CHECKPOINT_LIST_IMPL_DEF,
    &CHECKPOINT_EXISTS_IMPL_DEF,
    &CHECKPOINT_DELETE_IMPL_DEF,
];

#[harn_builtin(
    exposure = "harness.runtime.checkpoint",
    effects = ["state.write@arg0"],
    sig = "checkpoint(key: string, value: any) -> nil",
    category = "checkpoint",
    doc = "Persist a checkpoint key/value pair to durable storage immediately."
)]
fn checkpoint_impl(args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    let key = args.first().map(|a| a.display()).unwrap_or_default();
    let value = args.get(1).unwrap_or(&VmValue::Nil);
    let json_val = vm_to_json(value)?;
    with_state("checkpoint", |state| {
        state.set(key, json_val).map_err(VmError::Runtime)
    })?;
    Ok(VmValue::Nil)
}

#[harn_builtin(
    exposure = "harness.runtime.checkpoint_insert",
    effects = ["state.mutate@arg0"],
    sig = "checkpoint_insert(key: string, value: unknown) -> {inserted: bool, value: unknown}",
    category = "checkpoint",
    doc = "Atomically retain the first checkpoint value and report whether this call inserted it."
)]
fn checkpoint_insert_impl(args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    let key = args
        .first()
        .map(|value| value.display())
        .unwrap_or_default();
    let value = vm_to_json(args.get(1).unwrap_or(&VmValue::Nil))?;
    with_state("checkpoint_insert", |state| {
        state.insert(key, value).map_err(VmError::Runtime)
    })
}

#[harn_builtin(
    exposure = "harness.runtime.checkpoint_get",
    effects = ["state.read@arg0"],
    sig = "checkpoint_get(key: string) -> any",
    category = "checkpoint",
    doc = "Read a persisted checkpoint value, or nil if the key is absent."
)]
fn checkpoint_get_impl(args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    let key = args.first().map(|a| a.display()).unwrap_or_default();
    with_state("checkpoint_get", |state| {
        state.get(&key).map_err(VmError::Runtime)
    })
}

#[harn_builtin(
    exposure = "harness.runtime.checkpoint_clear",
    effects = ["state.mutate@const=checkpoints"],
    sig = "checkpoint_clear() -> nil",
    category = "checkpoint",
    doc = "Clear every checkpoint for the active pipeline."
)]
fn checkpoint_clear_impl(_args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    with_state("checkpoint_clear", |state| {
        state.clear().map_err(VmError::Runtime)
    })?;
    Ok(VmValue::Nil)
}

#[harn_builtin(
    exposure = "harness.runtime.checkpoint_list",
    effects = ["state.read@const=checkpoints"],
    sig = "checkpoint_list() -> list",
    category = "checkpoint",
    doc = "Return every checkpoint key for the active pipeline."
)]
fn checkpoint_list_impl(_args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    with_state("checkpoint_list", |state| {
        let keys = state.list().map_err(VmError::Runtime)?;
        Ok(VmValue::List(std::sync::Arc::new(
            keys.into_iter()
                .map(|k| VmValue::String(arcstr::ArcStr::from(k)))
                .collect(),
        )))
    })
}

#[harn_builtin(
    exposure = "harness.runtime.checkpoint_exists",
    effects = ["state.read@arg0"],
    sig = "checkpoint_exists(key: string) -> bool",
    category = "checkpoint",
    doc = "Return true when the checkpoint key is present (even when its value is nil)."
)]
fn checkpoint_exists_impl(args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    let key = args.first().map(|a| a.display()).unwrap_or_default();
    with_state("checkpoint_exists", |state| {
        state
            .exists(&key)
            .map(VmValue::Bool)
            .map_err(VmError::Runtime)
    })
}

#[harn_builtin(
    exposure = "harness.runtime.checkpoint_delete",
    effects = ["state.mutate@arg0"],
    sig = "checkpoint_delete(key: string) -> nil",
    category = "checkpoint",
    doc = "Remove a single key from the checkpoint store."
)]
fn checkpoint_delete_impl(args: &[VmValue], _out: &mut String) -> Result<VmValue, VmError> {
    let key = args.first().map(|a| a.display()).unwrap_or_default();
    with_state("checkpoint_delete", |state| {
        state.delete(&key).map_err(VmError::Runtime)
    })?;
    Ok(VmValue::Nil)
}

#[cfg(test)]
mod tests;