reddb_server/runtime/
integrity_tombstone.rs

1//! Issue #765 / S6 — opt-in SHA-256 integrity tombstones for input streams.
2//!
3//! Per ADR 0029 ("Integrity") and PRD #759: input streams support opt-in
4//! end-to-end SHA-256. The client streams a rolling hash over the row
5//! payloads and emits the expected digest in the terminal frame; the server
6//! recomputes the same hash and compares. Because S4 commits per chunk, rows
7//! are already durable when a mismatch is detected — rollback is impossible.
8//! Instead the server marks the affected RID range with an **integrity
9//! tombstone** in the collection metadata. Default reads filter tombstoned
10//! RIDs out of result sets.
11//!
12//! This module owns the durable representation (a JSON list persisted under
13//! a single `red_config` key) and the pure helpers the runtime + stream
14//! handler use. The runtime caches the parsed ranges in-memory so the common
15//! no-tombstone read path pays only a single relaxed atomic load.
16
17use crate::storage::unified::UnifiedStore;
18use crate::{RedDBError, RedDBResult};
19use reddb_types::Value;
20
21/// Single dot-notation key under which the whole tombstone list is stored.
22/// [`load_ranges`] uses the fallible store-config reader, which resolves the
23/// latest row by entity id and reads both resident and paged generations.
24pub const TOMBSTONE_KEY: &str = "stream.integrity.tombstones";
25
26/// Verification mode requested for an input stream. `None` is the default
27/// (`stream.integrity.default_verify`) and incurs no hashing overhead.
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
29pub enum VerifyMode {
30    #[default]
31    None,
32    Sha256,
33}
34
35impl VerifyMode {
36    /// Parse the wire token. Unknown / empty values fall back to `None` so a
37    /// malformed opt-in never terminates a stream that would otherwise run.
38    pub fn parse(token: &str) -> VerifyMode {
39        match token.trim().to_ascii_lowercase().as_str() {
40            "sha256" => VerifyMode::Sha256,
41            _ => VerifyMode::None,
42        }
43    }
44
45    pub fn is_enabled(self) -> bool {
46        matches!(self, VerifyMode::Sha256)
47    }
48}
49
50/// One tombstoned RID range, inclusive on both ends, scoped to a collection.
51/// RIDs (entity logical ids) are globally unique, but we still carry the
52/// table so reads filter precisely and forensic tooling can attribute the
53/// range to its origin.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct TombstoneRange {
56    pub table: String,
57    pub lo: u64,
58    pub hi: u64,
59}
60
61impl TombstoneRange {
62    pub fn new(table: impl Into<String>, lo: u64, hi: u64) -> Self {
63        Self {
64            table: table.into(),
65            lo,
66            hi,
67        }
68    }
69
70    /// True when `rid` falls inside this tombstoned range. RIDs (entity
71    /// logical ids) are drawn from a single global counter and never reused,
72    /// so a RID identifies exactly one row across the whole store — the read
73    /// filter can therefore match on RID alone and stays correct even for
74    /// projections that drop the `collection` system field. The `table` is
75    /// retained for the error envelope and forensic attribution.
76    pub fn covers_rid(&self, rid: u64) -> bool {
77        self.lo <= rid && rid <= self.hi
78    }
79}
80
81/// Serialize the range list to the compact JSON array persisted in
82/// `red_config`. Table names are SQL identifiers (validated at OpenStream),
83/// so the only escaping needed is the JSON string quote — kept explicit here
84/// rather than pulling a serializer for a three-field record.
85pub fn serialize_ranges(ranges: &[TombstoneRange]) -> String {
86    let mut out = String::with_capacity(2 + ranges.len() * 40);
87    out.push('[');
88    for (i, r) in ranges.iter().enumerate() {
89        if i > 0 {
90            out.push(',');
91        }
92        out.push_str("{\"table\":\"");
93        for ch in r.table.chars() {
94            if ch == '"' || ch == '\\' {
95                out.push('\\');
96            }
97            out.push(ch);
98        }
99        out.push_str(&format!("\",\"lo\":{},\"hi\":{}}}", r.lo, r.hi));
100    }
101    out.push(']');
102    out
103}
104
105/// A malformed persisted list cannot safely be treated as an empty list:
106/// that would expose rows already quarantined after a failed integrity check.
107pub fn parse_ranges(json: &str) -> RedDBResult<Vec<TombstoneRange>> {
108    let invalid = || RedDBError::Query("invalid persisted stream integrity tombstones".into());
109    let value: crate::json::Value =
110        crate::json::from_slice(json.as_bytes()).map_err(|_| invalid())?;
111    let arr = value.as_array().ok_or_else(invalid)?;
112    let mut out = Vec::with_capacity(arr.len());
113    for entry in arr {
114        let table = entry
115            .get("table")
116            .and_then(crate::json::Value::as_str)
117            .ok_or_else(invalid)?;
118        let lo = entry
119            .get("lo")
120            .and_then(parse_range_bound)
121            .ok_or_else(invalid)?;
122        let hi = entry
123            .get("hi")
124            .and_then(parse_range_bound)
125            .ok_or_else(invalid)?;
126        if lo > hi {
127            return Err(invalid());
128        }
129        out.push(TombstoneRange::new(table, lo, hi));
130    }
131    Ok(out)
132}
133
134// Persisted RID bounds are integer literals. Reject floating point values
135// instead of truncating a fractional/corrupted bound into another row's ID.
136fn parse_range_bound(value: &crate::json::Value) -> Option<u64> {
137    match value {
138        crate::json::Value::Integer(value) => u64::try_from(*value).ok(),
139        crate::json::Value::Decimal(value) => value.parse().ok(),
140        _ => None,
141    }
142}
143
144/// Load the current persisted ranges, propagating paging, codec and I/O errors.
145pub fn load_ranges(store: &UnifiedStore) -> RedDBResult<Vec<TombstoneRange>> {
146    match store.try_get_config(TOMBSTONE_KEY)? {
147        None => Ok(Vec::new()),
148        Some(Value::Text(json)) => parse_ranges(&json),
149        Some(_) => Err(RedDBError::Query(
150            "invalid persisted stream integrity tombstones".into(),
151        )),
152    }
153}
154
155/// Persist the full range list through the ordinary WAL-ordered config path.
156pub fn persist_ranges(store: &UnifiedStore, ranges: &[TombstoneRange]) -> RedDBResult<()> {
157    let json = serialize_ranges(ranges);
158    store.try_set_config_tree(TOMBSTONE_KEY, &crate::serde_json::Value::String(json))?;
159    Ok(())
160}
161
162/// Extract a record's RID (logical entity id) as exposed by SELECT scans.
163/// Mirrors the system-field convention in `record_search` where every
164/// scanned record carries `rid` as an unsigned integer.
165pub fn record_rid(record: &crate::storage::query::unified::UnifiedRecord) -> Option<u64> {
166    match record.get("rid")? {
167        Value::Integer(v) if *v >= 0 => Some(*v as u64),
168        Value::UnsignedInteger(v) => Some(*v),
169        _ => None,
170    }
171}
172
173/// True when a record is tombstoned by any range, matched on RID alone
174/// (RIDs are globally unique — see [`TombstoneRange::covers_rid`]). A record
175/// without a resolvable RID is never filtered (fail-open on read — a scan
176/// that cannot identify a row must not silently drop it).
177pub fn record_tombstoned(
178    ranges: &[TombstoneRange],
179    record: &crate::storage::query::unified::UnifiedRecord,
180) -> bool {
181    let Some(rid) = record_rid(record) else {
182        return false;
183    };
184    ranges.iter().any(|r| r.covers_rid(rid))
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    #[test]
192    fn serialize_parse_round_trip() {
193        let ranges = vec![
194            TombstoneRange::new("orders", 10, 12),
195            TombstoneRange::new("events", 5, 5),
196        ];
197        let json = serialize_ranges(&ranges);
198        let parsed = parse_ranges(&json).expect("valid ranges");
199        assert_eq!(parsed, ranges);
200    }
201
202    #[test]
203    fn parse_rejects_malformed_entries_without_losing_other_ranges() {
204        let json = r#"[{"table":"a","lo":1,"hi":3},{"table":"b","lo":4}]"#;
205        assert!(parse_ranges(json).is_err());
206        assert!(parse_ranges(r#"[{"table":"a","lo":3,"hi":1}]"#).is_err());
207    }
208
209    #[test]
210    fn parse_garbage_is_an_error() {
211        assert!(parse_ranges("not json").is_err());
212        assert!(parse_ranges("{}").is_err());
213        assert!(parse_ranges("[]").expect("empty list").is_empty());
214    }
215
216    #[test]
217    fn covers_rid_is_inclusive() {
218        let r = TombstoneRange::new("t", 4, 6);
219        assert!(r.covers_rid(4));
220        assert!(r.covers_rid(5));
221        assert!(r.covers_rid(6));
222        assert!(!r.covers_rid(3));
223        assert!(!r.covers_rid(7));
224    }
225
226    #[test]
227    fn verify_mode_parse_is_lenient() {
228        assert_eq!(VerifyMode::parse("sha256"), VerifyMode::Sha256);
229        assert_eq!(VerifyMode::parse("SHA256"), VerifyMode::Sha256);
230        assert_eq!(VerifyMode::parse("none"), VerifyMode::None);
231        assert_eq!(VerifyMode::parse(""), VerifyMode::None);
232        assert_eq!(VerifyMode::parse("bogus"), VerifyMode::None);
233        assert!(VerifyMode::Sha256.is_enabled());
234        assert!(!VerifyMode::None.is_enabled());
235    }
236
237    #[test]
238    fn table_name_with_quote_is_escaped() {
239        // Identifiers are validated at OpenStream so this can't occur in
240        // practice, but the serializer must not produce invalid JSON.
241        let ranges = vec![TombstoneRange::new("a\"b", 1, 2)];
242        let json = serialize_ranges(&ranges);
243        let parsed = parse_ranges(&json).expect("valid ranges");
244        assert_eq!(parsed, ranges);
245    }
246    #[test]
247    fn integrity_exact_numeric_ranges_reject_fractional_and_overflowed_ids() {
248        for rid in [(1_u64 << 53) + 1, (i64::MAX as u64) + 1, u64::MAX] {
249            let expected = vec![TombstoneRange::new("events", rid, rid)];
250            assert_eq!(
251                parse_ranges(&serialize_ranges(&expected)).expect("exact integer bounds"),
252                expected
253            );
254        }
255        for invalid in [
256            r#"[{"table":"events","lo":1.5,"hi":2}]"#,
257            r#"[{"table":"events","lo":1,"hi":2.5}]"#,
258            r#"[{"table":"events","lo":-1,"hi":2}]"#,
259            r#"[{"table":"events","lo":1,"hi":18446744073709551616}]"#,
260        ] {
261            assert!(
262                parse_ranges(invalid).is_err(),
263                "invalid RID bounds: {invalid}"
264            );
265        }
266    }
267}
reddb_server/runtime/integrity_tombstone.rs

reddb_server/runtime/
integrity_tombstone.rs
