graphrefly/
versioning.rs

1//! Node runtime versioning (D109/D112/D113).
2//!
3//! This is runtime metadata over DATA boundaries, not a protocol message, storage
4//! generation, checkpoint envelope version, or public mutation surface.
5
6use std::fmt;
7use std::rc::Rc;
8
9use serde_json::{Number, Value};
10
11use crate::json::{strict_canonical_json_bytes, validate_strict_json_value};
12use crate::protocol::AnyValue;
13
14const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
15
16/// `NodeVersionHashFn` type alias.
17pub type NodeVersionHashFn = Rc<dyn Fn(&[u8]) -> String>;
18
19#[derive(Clone)]
20/// `NodeVersioningPolicy` variants.
21pub enum NodeVersioningPolicy {
22    /// `Disabled` variant.
23    Disabled,
24    /// `Level0` variant.
25    Level0,
26    /// `Level1` variant.
27    Level1 {
28        /// `hash` field for `Level1`.
29        hash: Option<NodeVersionHashFn>,
30    },
31}
32
33impl fmt::Debug for NodeVersioningPolicy {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        match self {
36            Self::Disabled => f.write_str("Disabled"),
37            Self::Level0 => f.write_str("Level0"),
38            Self::Level1 { hash } => f
39                .debug_struct("Level1")
40                .field("hash", &hash.as_ref().map(|_| "<installed>"))
41                .finish(),
42        }
43    }
44}
45
46#[derive(Clone)]
47/// `ResolvedNodeVersioningPolicy` variants.
48pub enum ResolvedNodeVersioningPolicy {
49    /// `Disabled` variant.
50    Disabled,
51    /// `Level0` variant.
52    Level0,
53    /// `Level1` variant.
54    Level1 {
55        /// `hash` field for `Level1`.
56        hash: NodeVersionHashFn,
57    },
58}
59
60impl fmt::Debug for ResolvedNodeVersioningPolicy {
61    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
62        match self {
63            Self::Disabled => f.write_str("Disabled"),
64            Self::Level0 => f.write_str("Level0"),
65            Self::Level1 { .. } => f.write_str("Level1 { hash: <installed> }"),
66        }
67    }
68}
69
70#[derive(Debug, Clone, PartialEq, Eq)]
71/// `NodeVersion` variants.
72pub enum NodeVersion {
73    /// `V0` variant.
74    V0 {
75        /// `counter` field for counter.
76        counter: u64,
77    },
78    /// `V1` variant.
79    V1 {
80        /// `counter` field for counter.
81        counter: u64,
82        /// `cid` field for cid.
83        cid: String,
84        /// `prev` field for prev.
85        prev: Option<String>,
86    },
87}
88
89#[derive(Debug, Clone)]
90pub(crate) enum RestoredNodeVersion {
91    Disabled,
92    Version(NodeVersion),
93}
94
95#[derive(Debug, Clone, PartialEq, Eq)]
96pub(crate) struct NodeVersioningError {
97    message: String,
98}
99
100impl NodeVersioningError {
101    fn new(message: impl Into<String>) -> Self {
102        Self {
103            message: message.into(),
104        }
105    }
106}
107
108impl fmt::Display for NodeVersioningError {
109    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
110        f.write_str(&self.message)
111    }
112}
113
114impl std::error::Error for NodeVersioningError {}
115
116pub(crate) fn resolve_node_versioning_policy(
117    policy: Option<NodeVersioningPolicy>,
118) -> ResolvedNodeVersioningPolicy {
119    match policy {
120        Some(NodeVersioningPolicy::Disabled) => ResolvedNodeVersioningPolicy::Disabled,
121        None | Some(NodeVersioningPolicy::Level0) => ResolvedNodeVersioningPolicy::Level0,
122        Some(NodeVersioningPolicy::Level1 { hash }) => ResolvedNodeVersioningPolicy::Level1 {
123            hash: hash.unwrap_or_else(default_node_version_hash_fn),
124        },
125    }
126}
127
128fn default_node_version_hash_fn() -> NodeVersionHashFn {
129    Rc::new(default_node_version_hash)
130}
131
132/// Creates or computes `default_node_version_hash`.
133pub fn default_node_version_hash(bytes: &[u8]) -> String {
134    format!("fnv1a64:{}", fnv1a64(bytes))
135}
136
137fn fnv1a64(bytes: &[u8]) -> String {
138    let mut hash = 0xcbf2_9ce4_8422_2325u64;
139    for byte in bytes {
140        hash ^= u64::from(*byte);
141        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
142    }
143    format!("{hash:016x}")
144}
145
146pub(crate) fn create_node_version(
147    policy: &ResolvedNodeVersioningPolicy,
148    initial: Option<&AnyValue>,
149) -> Result<Option<NodeVersion>, NodeVersioningError> {
150    match policy {
151        ResolvedNodeVersioningPolicy::Disabled => Ok(None),
152        ResolvedNodeVersioningPolicy::Level0 => Ok(Some(NodeVersion::V0 { counter: 0 })),
153        ResolvedNodeVersioningPolicy::Level1 { hash } => {
154            let cid = hash_node_data(hash, initial)?;
155            Ok(Some(NodeVersion::V1 {
156                counter: 0,
157                cid,
158                prev: None,
159            }))
160        }
161    }
162}
163
164pub(crate) fn assert_node_version_data_compatible(
165    policy: &ResolvedNodeVersioningPolicy,
166    value: &AnyValue,
167) -> Result<(), NodeVersioningError> {
168    if matches!(policy, ResolvedNodeVersioningPolicy::Level1 { .. }) {
169        let json = any_to_strict_json(value, "$")?;
170        let _ = strict_canonical_json_bytes(&json).map_err(|err| {
171            NodeVersioningError::new(format!(
172                "node versioning: DATA is not strict canonical JSON compatible (D112): {err}"
173            ))
174        })?;
175    }
176    Ok(())
177}
178
179pub(crate) fn advance_node_version(
180    current: Option<&NodeVersion>,
181    policy: &ResolvedNodeVersioningPolicy,
182    value: &AnyValue,
183) -> Result<Option<NodeVersion>, NodeVersioningError> {
184    match policy {
185        ResolvedNodeVersioningPolicy::Disabled => Ok(None),
186        ResolvedNodeVersioningPolicy::Level0 => Ok(Some(NodeVersion::V0 {
187            counter: next_counter(current)?,
188        })),
189        ResolvedNodeVersioningPolicy::Level1 { hash } => {
190            let previous = match current {
191                Some(NodeVersion::V1 { cid, .. }) => Some(cid.clone()),
192                _ => None,
193            };
194            Ok(Some(NodeVersion::V1 {
195                counter: next_counter(current)?,
196                cid: hash_node_data(hash, Some(value))?,
197                prev: previous,
198            }))
199        }
200    }
201}
202
203pub(crate) fn validate_node_version_json(
204    value: &Value,
205    path: &str,
206) -> Result<NodeVersion, NodeVersioningError> {
207    validate_strict_json_value(value, path).map_err(|err| {
208        NodeVersioningError::new(format!(
209            "restore_graph: node version metadata is not strict canonical JSON compatible (D112): {err}"
210        ))
211    })?;
212    let Value::Object(record) = value else {
213        return Err(NodeVersioningError::new(format!(
214            "restore_graph: {path} must be an object"
215        )));
216    };
217    match record.get("level").and_then(Value::as_u64) {
218        Some(0) => {
219            require_fields(record, &["level", "counter"], path)?;
220            let counter = read_counter(record.get("counter"), path)?;
221            Ok(NodeVersion::V0 { counter })
222        }
223        Some(1) => {
224            require_fields(record, &["level", "counter", "cid", "prev"], path)?;
225            let counter = read_counter(record.get("counter"), path)?;
226            let cid = record
227                .get("cid")
228                .and_then(Value::as_str)
229                .ok_or_else(|| {
230                    NodeVersioningError::new(format!("restore_graph: {path}.cid must be a string"))
231                })?
232                .to_owned();
233            let prev = match record.get("prev") {
234                Some(Value::Null) => None,
235                Some(Value::String(value)) => Some(value.clone()),
236                _ => {
237                    return Err(NodeVersioningError::new(format!(
238                        "restore_graph: {path}.prev must be a string or null"
239                    )));
240                }
241            };
242            if counter == 0 && prev.is_some() {
243                return Err(NodeVersioningError::new(format!(
244                    "restore_graph: {path}.prev must be null when counter is 0"
245                )));
246            }
247            if counter > 0 && prev.is_none() {
248                return Err(NodeVersioningError::new(format!(
249                    "restore_graph: {path}.prev must be a string when counter is > 0"
250                )));
251            }
252            Ok(NodeVersion::V1 { counter, cid, prev })
253        }
254        _ => Err(NodeVersioningError::new(format!(
255            "restore_graph: {path}.level must be 0 or 1"
256        ))),
257    }
258}
259
260pub(crate) fn node_version_to_json(version: &NodeVersion) -> Value {
261    let mut record = serde_json::Map::new();
262    match version {
263        NodeVersion::V0 { counter } => {
264            record.insert("level".to_owned(), Value::Number(Number::from(0)));
265            record.insert("counter".to_owned(), Value::Number(Number::from(*counter)));
266        }
267        NodeVersion::V1 { counter, cid, prev } => {
268            record.insert("level".to_owned(), Value::Number(Number::from(1)));
269            record.insert("counter".to_owned(), Value::Number(Number::from(*counter)));
270            record.insert("cid".to_owned(), Value::String(cid.clone()));
271            record.insert(
272                "prev".to_owned(),
273                prev.clone().map_or(Value::Null, Value::String),
274            );
275        }
276    }
277    Value::Object(record)
278}
279
280pub(crate) fn verify_restored_node_version(
281    policy: &ResolvedNodeVersioningPolicy,
282    restored: Option<&Value>,
283    has_data: bool,
284    cache: Option<&AnyValue>,
285    path: &str,
286) -> Result<RestoredNodeVersion, NodeVersioningError> {
287    let Some(restored) = restored else {
288        return match policy {
289            ResolvedNodeVersioningPolicy::Disabled => Ok(RestoredNodeVersion::Disabled),
290            _ => Err(NodeVersioningError::new(
291                "restore_graph: checkpoint node version metadata is required by the selected node versioning policy (D109)",
292            )),
293        };
294    };
295    let version = validate_node_version_json(restored, path)?;
296    match (policy, &version) {
297        (ResolvedNodeVersioningPolicy::Disabled, _) => Err(NodeVersioningError::new(
298            "restore_graph: checkpoint node version metadata is present but node versioning is disabled",
299        )),
300        (ResolvedNodeVersioningPolicy::Level0, NodeVersion::V0 { .. }) => {
301            Ok(RestoredNodeVersion::Version(version))
302        }
303        (ResolvedNodeVersioningPolicy::Level0, NodeVersion::V1 { .. }) => Err(
304            NodeVersioningError::new(
305                "restore_graph: checkpoint node version level 1 requires matching node versioning policy",
306            ),
307        ),
308        (ResolvedNodeVersioningPolicy::Level1 { .. }, NodeVersion::V0 { .. }) => Err(
309            NodeVersioningError::new(
310                "restore_graph: checkpoint node version level 0 requires matching node versioning policy",
311            ),
312        ),
313        (ResolvedNodeVersioningPolicy::Level1 { hash }, NodeVersion::V1 { counter, cid, .. }) => {
314            if !has_data && *counter > 0 {
315                return Err(NodeVersioningError::new(
316                    "restore_graph: checkpoint node version cid cannot be verified without current DATA under V1 versioning (D109)",
317                ));
318            }
319            let expected = hash_node_data(hash, if has_data { cache } else { None })?;
320            if &expected != cid {
321                return Err(NodeVersioningError::new(
322                    "restore_graph: checkpoint node version cid does not match the selected node versioning hash policy (D109)",
323                ));
324            }
325            Ok(RestoredNodeVersion::Version(version))
326        }
327    }
328}
329
330fn next_counter(current: Option<&NodeVersion>) -> Result<u64, NodeVersioningError> {
331    let Some(current) = current else {
332        return Ok(1);
333    };
334    current
335        .counter()
336        .checked_add(1)
337        .ok_or_else(|| {
338            NodeVersioningError::new(
339                "node versioning: counter overflow while advancing node runtime version (D109)",
340            )
341        })
342        .and_then(|counter| {
343            if counter > MAX_SAFE_INTEGER {
344                Err(NodeVersioningError::new(
345                    "node versioning: counter exceeded the strict JSON safe-integer range (D109)",
346                ))
347            } else {
348                Ok(counter)
349            }
350        })
351}
352
353fn require_fields(
354    record: &serde_json::Map<String, Value>,
355    fields: &[&str],
356    path: &str,
357) -> Result<(), NodeVersioningError> {
358    if record.len() != fields.len() || fields.iter().any(|field| !record.contains_key(*field)) {
359        return Err(NodeVersioningError::new(format!(
360            "restore_graph: {path} has unexpected node version fields"
361        )));
362    }
363    Ok(())
364}
365
366fn read_counter(value: Option<&Value>, path: &str) -> Result<u64, NodeVersioningError> {
367    let counter = value.and_then(Value::as_u64).ok_or_else(|| {
368        NodeVersioningError::new(format!(
369            "restore_graph: {path}.counter must be a non-negative safe integer"
370        ))
371    })?;
372    if counter > MAX_SAFE_INTEGER {
373        return Err(NodeVersioningError::new(format!(
374            "restore_graph: {path}.counter must be a non-negative safe integer"
375        )));
376    }
377    Ok(counter)
378}
379
380fn hash_node_data(
381    hash: &NodeVersionHashFn,
382    value: Option<&AnyValue>,
383) -> Result<String, NodeVersioningError> {
384    let json = match value {
385        Some(value) => any_to_strict_json(value, "$")?,
386        None => absent_v1_seed(),
387    };
388    let bytes = strict_canonical_json_bytes(&json).map_err(|err| {
389        NodeVersioningError::new(format!(
390            "node versioning: DATA is not strict canonical JSON compatible (D112): {err}"
391        ))
392    })?;
393    Ok(hash(&bytes))
394}
395
396fn absent_v1_seed() -> Value {
397    let mut record = serde_json::Map::new();
398    record.insert(
399        "@graphrefly/node-version".to_owned(),
400        Value::String("v1-absent".to_owned()),
401    );
402    Value::Object(record)
403}
404
405fn any_to_strict_json(value: &AnyValue, path: &str) -> Result<Value, NodeVersioningError> {
406    let out = if let Some(v) = value.downcast_ref::<Value>() {
407        v.clone()
408    } else if let Some(v) = value.downcast_ref::<String>() {
409        Value::String(v.clone())
410    } else if let Some(v) = value.downcast_ref::<bool>() {
411        Value::Bool(*v)
412    } else if let Some(v) = value.downcast_ref::<i32>() {
413        Value::Number(Number::from(*v))
414    } else if let Some(v) = value.downcast_ref::<i64>() {
415        Value::Number(Number::from(*v))
416    } else if let Some(v) = value.downcast_ref::<u32>() {
417        Value::Number(Number::from(*v))
418    } else if let Some(v) = value.downcast_ref::<u64>() {
419        Value::Number(Number::from(*v))
420    } else if let Some(v) = value.downcast_ref::<usize>() {
421        Value::Number(Number::from(*v as u64))
422    } else if let Some(v) = value.downcast_ref::<f64>() {
423        Number::from_f64(*v).map(Value::Number).ok_or_else(|| {
424            NodeVersioningError::new(format!(
425                "node versioning: DATA at {path} is not strict JSON compatible"
426            ))
427        })?
428    } else {
429        return Err(NodeVersioningError::new(format!(
430            "node versioning: DATA at {path} is not strict JSON compatible"
431        )));
432    };
433    validate_strict_json_value(&out, path).map_err(|err| {
434        NodeVersioningError::new(format!(
435            "node versioning: DATA is not strict canonical JSON compatible (D112): {err}"
436        ))
437    })?;
438    Ok(out)
439}
440
441impl NodeVersion {
442    /// Updates or reads `level`.
443    pub fn level(&self) -> u8 {
444        match self {
445            Self::V0 { .. } => 0,
446            Self::V1 { .. } => 1,
447        }
448    }
449
450    /// Updates or reads `counter`.
451    pub fn counter(&self) -> u64 {
452        match self {
453            Self::V0 { counter } | Self::V1 { counter, .. } => *counter,
454        }
455    }
456}
graphrefly/versioning.rs

graphrefly/
versioning.rs
