qualia_client_core/wellfair/
sync_protocol.rs

1//! Versioned, replay-safe sync-operation protocol and quarantined inbox.
2//!
3//! This is the delivery layer that lets Projects/Finance (and any domain) converge across
4//! nodes without duplicating money or obligations. It implements the master plan's SyncService
5//! (§4.2), link-protocol framing (§9.5), and the money-safety invariants (§17):
6//!
7//! - every operation is **versioned** (protocol + schema) and **content-hashed**;
8//! - the inbox is **quarantined**: untrusted frames are decoded into this DTO and validated
9//!   before anything is admitted — oversized, malformed, unsigned, wrong-hash, wrong-version,
10//!   and Sanctuary-classified frames are **rejected fail-closed**;
11//! - admission is **idempotent**: a replayed `operation_id` is recorded as `Duplicate`, never
12//!   applied twice;
13//! - [`merge_operations`] is **add-wins by operation id** and **order-independent**, so
14//!   duplicate/reordered/replayed frames converge to the same set — the same discipline the
15//!   domain layers (`finance::derived_balance`, `projects::derive_obligations`) use to derive
16//!   totals purely over the unique-id set.
17//!
18//! Full signature *verification* is the identity/key-vault layer's job (it holds the actor
19//! public keys); this layer verifies presence + integrity and enforces the routing lane.
20
21use std::collections::HashSet;
22use std::fs::{self, OpenOptions};
23use std::io::{BufRead, BufReader, Write};
24use std::path::{Path, PathBuf};
25
26use serde::{Deserialize, Serialize};
27use sha2::{Digest, Sha256};
28
29pub const CURRENT_PROTOCOL_VERSION: u16 = 1;
30pub const CURRENT_SCHEMA_VERSION: u16 = 1;
31/// Hard cap on a single serialized operation (defends the quarantine against oversized frames).
32pub const MAX_OPERATION_BYTES: usize = 64 * 1024;
33/// Hard cap on the payload summary carried inline.
34pub const MAX_SUMMARY_BYTES: usize = 16 * 1024;
35
36pub const SYNC_INBOX_FILE: &str = "wellfair/sync_inbox.jsonl";
37
38/// A single versioned, content-addressed sync operation (the wire DTO).
39#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
40pub struct SyncOperation {
41    pub protocol_version: u16,
42    pub schema_version: u16,
43    /// Stable operation identifier — the dedup/idempotency anchor.
44    pub operation_id: String,
45    pub record_id: String,
46    pub kind: String,
47    /// SHA-256 (hex) of `payload_summary` — integrity check on receipt.
48    pub content_hash: String,
49    /// Lamport clock for causal ordering across nodes.
50    pub lamport: u64,
51    #[serde(default, skip_serializing_if = "Option::is_none")]
52    pub parent_op_id: Option<String>,
53    pub actor_did: String,
54    /// Routing lane: "Public" | "Restricted" | "Classified".
55    pub sensitivity: String,
56    /// Approved projection / journal summary carried inline (no sensitive plaintext for Sanctuary).
57    pub payload_summary: String,
58    pub committed_unix: u32,
59    /// Detached signature (hex) over [`SyncOperation::signing_payload`]. Presence is required
60    /// (fail-closed); cryptographic verification is performed by the identity layer.
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    pub signature: Option<String>,
63}
64
65/// Outcome of validating/admitting an inbound operation.
66#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(rename_all = "snake_case", tag = "state", content = "reason")]
68pub enum AdmitOutcome {
69    /// Passed all checks and is newly admitted.
70    Validated,
71    /// A prior operation with this id was already admitted; ignored (idempotent replay).
72    Duplicate,
73    /// Failed a validation check; carries the reason.
74    Rejected(String),
75}
76
77impl AdmitOutcome {
78    pub fn is_validated(&self) -> bool {
79        matches!(self, AdmitOutcome::Validated)
80    }
81    pub fn is_rejected(&self) -> bool {
82        matches!(self, AdmitOutcome::Rejected(_))
83    }
84}
85
86pub fn sha256_hex(bytes: &[u8]) -> String {
87    hex::encode(Sha256::digest(bytes))
88}
89
90impl SyncOperation {
91    /// Build a well-formed, content-hashed operation (signature filled in by the host layer).
92    #[allow(clippy::too_many_arguments)]
93    pub fn new(
94        operation_id: impl Into<String>,
95        record_id: impl Into<String>,
96        kind: impl Into<String>,
97        actor_did: impl Into<String>,
98        sensitivity: impl Into<String>,
99        payload_summary: impl Into<String>,
100        lamport: u64,
101        committed_unix: u32,
102    ) -> Self {
103        let payload_summary = payload_summary.into();
104        let content_hash = sha256_hex(payload_summary.as_bytes());
105        Self {
106            protocol_version: CURRENT_PROTOCOL_VERSION,
107            schema_version: CURRENT_SCHEMA_VERSION,
108            operation_id: operation_id.into(),
109            record_id: record_id.into(),
110            kind: kind.into(),
111            content_hash,
112            lamport,
113            parent_op_id: None,
114            actor_did: actor_did.into(),
115            sensitivity: sensitivity.into(),
116            payload_summary,
117            committed_unix,
118            signature: None,
119        }
120    }
121
122    /// The bytes a signature must cover: id + record + content hash (binds identity to content).
123    pub fn signing_payload(&self) -> Vec<u8> {
124        format!(
125            "{}|{}|{}",
126            self.operation_id, self.record_id, self.content_hash
127        )
128        .into_bytes()
129    }
130
131    pub fn with_signature(mut self, signature_hex: impl Into<String>) -> Self {
132        self.signature = Some(signature_hex.into());
133        self
134    }
135}
136
137/// Validate an inbound operation fail-closed. `seen_ids` is the set of already-admitted
138/// operation ids (for replay detection). Returns the admission outcome without persisting.
139pub fn validate_operation(op: &SyncOperation, seen_ids: &HashSet<String>) -> AdmitOutcome {
140    // Version gate: refuse anything we don't understand.
141    if op.protocol_version != CURRENT_PROTOCOL_VERSION {
142        return AdmitOutcome::Rejected(format!(
143            "unsupported protocol version {} (expected {CURRENT_PROTOCOL_VERSION})",
144            op.protocol_version
145        ));
146    }
147    if op.schema_version != CURRENT_SCHEMA_VERSION {
148        return AdmitOutcome::Rejected(format!(
149            "unsupported schema version {} (expected {CURRENT_SCHEMA_VERSION})",
150            op.schema_version
151        ));
152    }
153    // Size bounds (quarantine defense).
154    if op.payload_summary.len() > MAX_SUMMARY_BYTES {
155        return AdmitOutcome::Rejected("payload summary exceeds size cap".into());
156    }
157    match serde_json::to_string(op) {
158        Ok(s) if s.len() > MAX_OPERATION_BYTES => {
159            return AdmitOutcome::Rejected("operation exceeds size cap".into());
160        }
161        Err(e) => return AdmitOutcome::Rejected(format!("operation not serializable: {e}")),
162        _ => {}
163    }
164    // Signature must be present (fail closed); full verification is the identity layer's job.
165    match &op.signature {
166        Some(sig) if !sig.is_empty() => {}
167        _ => return AdmitOutcome::Rejected("missing signature (fail closed)".into()),
168    }
169    // Integrity: the content hash must match the carried payload.
170    if op.content_hash != sha256_hex(op.payload_summary.as_bytes()) {
171        return AdmitOutcome::Rejected("content hash does not match payload".into());
172    }
173    // Routing lane: Sanctuary/Classified operations must never traverse the ordinary inbox (§5.2).
174    if op.sensitivity == "Classified" {
175        return AdmitOutcome::Rejected(
176            "Classified/Sanctuary operations are excluded from the ordinary sync lane".into(),
177        );
178    }
179    // Replay: a previously-admitted id is idempotently ignored.
180    if seen_ids.contains(&op.operation_id) {
181        return AdmitOutcome::Duplicate;
182    }
183    AdmitOutcome::Validated
184}
185
186/// Next Lamport clock value given the local counter and an observed remote value.
187pub fn lamport_next(local: u64, observed: u64) -> u64 {
188    local.max(observed).saturating_add(1)
189}
190
191/// Merge two operation sets **add-wins by operation id** (never re-apply), returning a
192/// deterministically ordered union (by Lamport clock, then id). Idempotent and order-independent.
193pub fn merge_operations(
194    existing: &[SyncOperation],
195    incoming: &[SyncOperation],
196) -> Vec<SyncOperation> {
197    let mut merged: Vec<SyncOperation> = Vec::with_capacity(existing.len() + incoming.len());
198    for op in existing.iter().chain(incoming.iter()) {
199        if !merged.iter().any(|e| e.operation_id == op.operation_id) {
200            merged.push(op.clone());
201        }
202    }
203    merged.sort_by(|a, b| {
204        a.lamport
205            .cmp(&b.lamport)
206            .then_with(|| a.operation_id.cmp(&b.operation_id))
207    });
208    merged
209}
210
211/// A persisted inbox record: the operation plus its admission outcome and time.
212#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
213pub struct InboxRecord {
214    pub operation: SyncOperation,
215    pub outcome: AdmitOutcome,
216    pub admitted_unix: u32,
217}
218
219/// Durable quarantined inbox (append-only jsonl). Admission validates, dedupes by operation id,
220/// and records the outcome; only `Validated` records represent applicable operations.
221pub struct SyncInbox {
222    path: PathBuf,
223}
224
225impl SyncInbox {
226    pub fn open(storage_root: impl AsRef<Path>) -> std::io::Result<Self> {
227        let path = storage_root.as_ref().join(SYNC_INBOX_FILE);
228        if let Some(parent) = path.parent() {
229            fs::create_dir_all(parent)?;
230        }
231        if !path.exists() {
232            OpenOptions::new().create(true).write(true).open(&path)?;
233        }
234        Ok(Self { path })
235    }
236
237    fn load_all(&self) -> std::io::Result<Vec<InboxRecord>> {
238        let file = fs::File::open(&self.path)?;
239        let reader = BufReader::new(file);
240        let mut records = Vec::new();
241        for line in reader.lines() {
242            let line = line?;
243            if line.trim().is_empty() {
244                continue;
245            }
246            if let Ok(rec) = serde_json::from_str::<InboxRecord>(&line) {
247                records.push(rec);
248            }
249        }
250        Ok(records)
251    }
252
253    /// The set of operation ids already admitted as Validated (for replay detection).
254    fn admitted_ids(records: &[InboxRecord]) -> HashSet<String> {
255        records
256            .iter()
257            .filter(|r| r.outcome.is_validated())
258            .map(|r| r.operation.operation_id.clone())
259            .collect()
260    }
261
262    /// Validate and durably record an inbound operation. Idempotent: a replayed id yields
263    /// `Duplicate` and is not applied again. Returns the admission outcome.
264    pub fn admit(&self, op: &SyncOperation, now_unix: u32) -> std::io::Result<AdmitOutcome> {
265        let existing = self.load_all()?;
266        let seen = Self::admitted_ids(&existing);
267        let outcome = validate_operation(op, &seen);
268        let record = InboxRecord {
269            operation: op.clone(),
270            outcome: outcome.clone(),
271            admitted_unix: now_unix,
272        };
273        let line =
274            serde_json::to_string(&record).map_err(|e| std::io::Error::other(e.to_string()))?;
275        let mut file = OpenOptions::new().append(true).open(&self.path)?;
276        writeln!(file, "{line}")?;
277        file.sync_all()?;
278        Ok(outcome)
279    }
280
281    /// All admitted-`Validated` operations, in Lamport order (the applicable set).
282    pub fn validated_operations(&self) -> std::io::Result<Vec<SyncOperation>> {
283        let mut ops: Vec<SyncOperation> = self
284            .load_all()?
285            .into_iter()
286            .filter(|r| r.outcome.is_validated())
287            .map(|r| r.operation)
288            .collect();
289        // Collapse any accidental duplicates and order deterministically.
290        Ok(merge_operations(&ops.split_off(0), &[]))
291    }
292
293    pub fn list_recent(&self, limit: usize) -> std::io::Result<Vec<InboxRecord>> {
294        let mut all = self.load_all()?;
295        if all.len() > limit {
296            all.drain(0..all.len() - limit);
297        }
298        all.reverse();
299        Ok(all)
300    }
301}
302
303#[cfg(test)]
304mod tests {
305    use super::*;
306
307    fn signed(id: &str, kind: &str, summary: &str, lamport: u64) -> SyncOperation {
308        SyncOperation::new(
309            id,
310            format!("urn:wellfair:{kind}:{id}"),
311            kind,
312            "did:wf:remote",
313            "Restricted",
314            summary,
315            lamport,
316            1_700_000_000,
317        )
318        .with_signature("deadbeef")
319    }
320
321    #[test]
322    fn valid_operation_admitted() {
323        let seen = HashSet::new();
324        let op = signed("op1", "ledger_entry", "{\"amount_cents\":100}", 1);
325        assert_eq!(validate_operation(&op, &seen), AdmitOutcome::Validated);
326    }
327
328    #[test]
329    fn replayed_id_is_duplicate() {
330        let mut seen = HashSet::new();
331        seen.insert("op1".to_string());
332        let op = signed("op1", "ledger_entry", "x", 1);
333        assert_eq!(validate_operation(&op, &seen), AdmitOutcome::Duplicate);
334    }
335
336    #[test]
337    fn missing_signature_rejected() {
338        let seen = HashSet::new();
339        let mut op = signed("op2", "ledger_entry", "x", 1);
340        op.signature = None;
341        assert!(validate_operation(&op, &seen).is_rejected());
342    }
343
344    #[test]
345    fn tampered_content_hash_rejected() {
346        let seen = HashSet::new();
347        let mut op = signed("op3", "ledger_entry", "original", 1);
348        op.payload_summary = "tampered".into(); // hash no longer matches
349        assert!(validate_operation(&op, &seen).is_rejected());
350    }
351
352    #[test]
353    fn classified_operation_rejected_from_ordinary_lane() {
354        let seen = HashSet::new();
355        let mut op = signed("op4", "sanctuary_note", "x", 1);
356        op.sensitivity = "Classified".into();
357        assert!(validate_operation(&op, &seen).is_rejected());
358    }
359
360    #[test]
361    fn wrong_protocol_version_rejected() {
362        let seen = HashSet::new();
363        let mut op = signed("op5", "ledger_entry", "x", 1);
364        op.protocol_version = 99;
365        assert!(validate_operation(&op, &seen).is_rejected());
366    }
367
368    #[test]
369    fn oversized_summary_rejected() {
370        let seen = HashSet::new();
371        let big = "a".repeat(MAX_SUMMARY_BYTES + 1);
372        let op = signed("op6", "ledger_entry", &big, 1);
373        assert!(validate_operation(&op, &seen).is_rejected());
374    }
375
376    #[test]
377    fn lamport_next_is_monotonic() {
378        assert_eq!(lamport_next(3, 5), 6);
379        assert_eq!(lamport_next(7, 2), 8);
380        assert!(lamport_next(u64::MAX, u64::MAX) >= u64::MAX);
381    }
382
383    #[test]
384    fn merge_is_idempotent_and_order_independent() {
385        let a = vec![
386            signed("a", "ledger_entry", "1", 2),
387            signed("b", "ledger_entry", "2", 1),
388        ];
389        let incoming = vec![
390            signed("b", "ledger_entry", "2", 1),
391            signed("c", "ledger_entry", "3", 3),
392            signed("b", "ledger_entry", "2", 1), // duplicate
393        ];
394        let merged = merge_operations(&a, &incoming);
395        assert_eq!(merged.len(), 3);
396        // Lamport order: b(1), a(2), c(3)
397        assert_eq!(merged[0].operation_id, "b");
398        assert_eq!(merged[1].operation_id, "a");
399        assert_eq!(merged[2].operation_id, "c");
400        // Order independence + idempotency.
401        let other = merge_operations(&incoming, &a);
402        assert_eq!(merged, other);
403        let twice = merge_operations(&merge_operations(&a, &incoming), &incoming);
404        assert_eq!(merged, twice);
405    }
406
407    #[test]
408    fn inbox_dedupes_replayed_operations() {
409        let dir = tempfile::tempdir().unwrap();
410        let inbox = SyncInbox::open(dir.path()).unwrap();
411        let op = signed("op-replay", "ledger_entry", "{\"amount_cents\":500}", 1);
412
413        assert_eq!(inbox.admit(&op, 10).unwrap(), AdmitOutcome::Validated);
414        // Replay of the same op id is idempotent.
415        assert_eq!(inbox.admit(&op, 11).unwrap(), AdmitOutcome::Duplicate);
416        assert_eq!(inbox.admit(&op, 12).unwrap(), AdmitOutcome::Duplicate);
417
418        // Only one validated operation exists despite three admissions.
419        assert_eq!(inbox.validated_operations().unwrap().len(), 1);
420    }
421
422    #[test]
423    fn inbox_survives_reopen_and_orders_by_lamport() {
424        let dir = tempfile::tempdir().unwrap();
425        {
426            let inbox = SyncInbox::open(dir.path()).unwrap();
427            inbox
428                .admit(&signed("z", "ledger_entry", "z", 5), 1)
429                .unwrap();
430            inbox
431                .admit(&signed("y", "ledger_entry", "y", 2), 1)
432                .unwrap();
433        }
434        let reopened = SyncInbox::open(dir.path()).unwrap();
435        let ops = reopened.validated_operations().unwrap();
436        assert_eq!(ops.len(), 2);
437        assert_eq!(ops[0].operation_id, "y"); // lamport 2 before 5
438        assert_eq!(ops[1].operation_id, "z");
439    }
440
441    #[test]
442    fn two_node_partition_rejoin_converges() {
443        // Node A and Node B each admit a disjoint op plus a shared op; after exchanging
444        // operation sets both nodes hold the identical validated set.
445        let shared = signed("shared", "contribution", "s", 1);
446        let a_only = signed("a1", "contribution", "a", 2);
447        let b_only = signed("b1", "contribution", "b", 3);
448
449        let node_a = vec![shared.clone(), a_only.clone()];
450        let node_b = vec![shared.clone(), b_only.clone()];
451
452        let a_after = merge_operations(&node_a, &node_b);
453        let b_after = merge_operations(&node_b, &node_a);
454        assert_eq!(a_after, b_after);
455        assert_eq!(a_after.len(), 3);
456    }
457}
qualia_client_core/wellfair/sync_protocol.rs

qualia_client_core/wellfair/
sync_protocol.rs
