resuma/exec/
durable.rs

1//! Durable storage — Resuma's own KV + checkpoint persistence (`.resuma/durable/`).
2
3use std::fs;
4use std::path::{Path, PathBuf};
5
6use parking_lot::RwLock;
7use serde::{de::DeserializeOwned, Serialize};
8use serde_json::Value;
9
10use crate::core::{Result, ResumaError};
11
12use super::types::{GraphId, GraphSnapshot, WorkerEvent};
13
14static ROOT: RwLock<Option<PathBuf>> = RwLock::new(None);
15
16/// Configure durable storage root (default: `.resuma/durable` under cwd).
17pub fn configure(root: impl AsRef<Path>) {
18    let p = root.as_ref().to_path_buf();
19    let _ = fs::create_dir_all(&p);
20    *ROOT.write() = Some(p);
21}
22
23fn root_dir() -> PathBuf {
24    ROOT.read()
25        .clone()
26        .unwrap_or_else(|| PathBuf::from(".resuma/durable"))
27}
28
29fn key_path(namespace: &str, key: &str) -> PathBuf {
30    let safe_ns = sanitize(namespace);
31    let safe_key = sanitize(key);
32    root_dir().join(safe_ns).join(format!("{safe_key}.json"))
33}
34
35fn sanitize(s: &str) -> String {
36    s.chars()
37        .map(|c| {
38            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
39                c
40            } else {
41                '_'
42            }
43        })
44        .collect()
45}
46
47/// Get a JSON value from durable storage.
48pub fn get(namespace: &str, key: &str) -> Option<Value> {
49    let path = key_path(namespace, key);
50    let data = fs::read_to_string(path).ok()?;
51    serde_json::from_str(&data).ok()
52}
53
54/// Set a JSON value in durable storage (atomic: temp file + fsync + rename).
55pub fn set(namespace: &str, key: &str, value: &Value) -> Result<()> {
56    use std::io::Write;
57    let path = key_path(namespace, key);
58    if let Some(parent) = path.parent() {
59        fs::create_dir_all(parent).map_err(ResumaError::Io)?;
60    }
61    let data = serde_json::to_string_pretty(value)?;
62    let tmp = path.with_extension("json.tmp");
63    {
64        let mut f = fs::File::create(&tmp).map_err(ResumaError::Io)?;
65        f.write_all(data.as_bytes()).map_err(ResumaError::Io)?;
66        f.sync_all().map_err(ResumaError::Io)?;
67    }
68    fs::rename(&tmp, &path).map_err(ResumaError::Io)
69}
70
71/// Typed get/set helpers.
72pub fn get_typed<T: DeserializeOwned>(namespace: &str, key: &str) -> Result<Option<T>> {
73    match get(namespace, key) {
74        Some(v) => Ok(Some(serde_json::from_value(v)?)),
75        None => Ok(None),
76    }
77}
78
79pub fn set_typed<T: Serialize>(namespace: &str, key: &str, value: &T) -> Result<()> {
80    set(namespace, key, &serde_json::to_value(value)?)
81}
82
83const GRAPHS_NS: &str = "graphs";
84const EVENTS_NS: &str = "events";
85const CHECKPOINTS_NS: &str = "checkpoints";
86
87/// Persist graph snapshot for replay across restarts.
88pub fn persist_graph(snapshot: &GraphSnapshot) -> Result<()> {
89    set(GRAPHS_NS, &snapshot.id.0, &serde_json::to_value(snapshot)?)
90}
91
92pub fn load_graph(id: &GraphId) -> Option<GraphSnapshot> {
93    get(GRAPHS_NS, &id.0).and_then(|v| serde_json::from_value(v).ok())
94}
95
96/// Append-only event log on disk.
97pub fn persist_events(id: &GraphId, events: &[WorkerEvent]) -> Result<()> {
98    set(EVENTS_NS, &id.0, &serde_json::to_value(events)?)
99}
100
101pub fn load_events(id: &GraphId) -> Option<Vec<WorkerEvent>> {
102    get(EVENTS_NS, &id.0).and_then(|v| serde_json::from_value(v).ok())
103}
104
105/// Checkpoint worker state mid-execution.
106pub fn save_checkpoint(id: &GraphId, state: &super::state::StateStore) -> Result<()> {
107    set(
108        CHECKPOINTS_NS,
109        &id.0,
110        &serde_json::to_value(state.snapshot())?,
111    )
112}
113
114pub fn load_checkpoint(id: &GraphId) -> Option<super::state::StateStore> {
115    let map = get(CHECKPOINTS_NS, &id.0)?;
116    let store = super::state::StateStore::default();
117    if let Some(obj) = map.as_object() {
118        for (k, v) in obj {
119            store.set(k.clone(), v.clone());
120        }
121    }
122    Some(store)
123}
124
125const EXECUTIONS_NS: &str = "executions";
126
127/// Persisted execution metadata for pause/resume across restarts.
128#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
129pub struct ExecutionRecord {
130    pub graph_id: GraphId,
131    pub worker: String,
132    pub input: Value,
133    pub plan: super::types::ExecutionPlan,
134    pub profile: super::resources::ResourceProfile,
135    pub paused: bool,
136    /// Hard cancel — graph must not be resumed (distinct from cooperative pause).
137    #[serde(default)]
138    pub cancelled: bool,
139}
140
141pub fn save_execution_record(record: &ExecutionRecord) -> Result<()> {
142    set(
143        EXECUTIONS_NS,
144        &record.graph_id.0,
145        &serde_json::to_value(record)?,
146    )
147}
148
149pub fn load_execution_record(id: &GraphId) -> Option<ExecutionRecord> {
150    get(EXECUTIONS_NS, &id.0).and_then(|v| serde_json::from_value(v).ok())
151}
152
153const TOKENS_NS: &str = "tokens";
154
155/// Persist graph-scoped access token for SSE / UI controls.
156pub fn save_graph_token(id: &GraphId, token: &str) -> Result<()> {
157    set(TOKENS_NS, &id.0, &serde_json::json!({ "token": token }))
158}
159
160pub fn load_graph_token(id: &GraphId) -> Option<String> {
161    get(TOKENS_NS, &id.0).and_then(|v| v.get("token").and_then(|t| t.as_str()).map(str::to_string))
162}
resuma/exec/durable.rs

resuma/exec/
durable.rs
