safemesh_crdt/
local.rs

1// Copyright (C) 2026 Ben Cassie
2// SPDX-License-Identifier: Apache-2.0
3//! Linux/local-filesystem packet-A adapter. All writers for a replica set must
4//! use the same directory and fixed configuration. Keep fence files in place.
5//! LocalReplica acknowledges in memory; DurableReplica commits before acknowledgement
6//! and offers checked ordinary restart of an existing committed store.
7use crate::{
8    ownership::*, Admission, Crdt, EventLog, GCounter, GCounterDelta, OrSet, OrSetDelta, Record,
9    RecordId, WireDecode, WireEncode, WireError, WireSchema,
10};
11use alloc::{format, string::String, vec::Vec};
12use std::{
13    fs::{self, File, OpenOptions, TryLockError},
14    io::{self, Read, Seek, SeekFrom, Write},
15    path::{Path, PathBuf},
16};
17
18#[derive(Debug)]
19pub enum LocalError {
20    Refused,
21    Exhausted,
22    RecoveryRequired,
23    Configuration,
24    History(WireError),
25    InvalidHistory,
26    Io(io::Error),
27}
28impl From<io::Error> for LocalError {
29    fn from(e: io::Error) -> Self {
30        Self::Io(e)
31    }
32}
33
34/// A generation captured by a caller. Renewal invalidates all older tickets.
35#[derive(Clone, Copy, Debug)]
36pub struct WriteTicket(u64);
37
38/// Owns the OS lock for its entire lifetime. No Clone or mutable state/log
39/// access is exposed. Contention yields a read-only instance whose writes fail.
40pub struct LocalReplica<C: Crdt> {
41    config: WriterConfig,
42    fence: File,
43    held: bool,
44    generation: u64,
45    state: C,
46    log: EventLog<C::Delta>,
47    last_sequence: u64,
48}
49
50impl<C: Crdt> Drop for LocalReplica<C> {
51    fn drop(&mut self) {
52        // Explicitly release our open-file-description lock before close. A
53        // concurrent fork/exec can briefly inherit an FD even with CLOEXEC.
54        let _ = self.fence.unlock();
55    }
56}
57
58impl<C: Crdt> LocalReplica<C>
59where
60    C::Delta: OwnedDelta + Clone + PartialEq,
61{
62    fn fresh(root: &Path, config: WriterConfig, state: C) -> Result<Self, LocalError> {
63        config.validate().map_err(|_| LocalError::Configuration)?;
64        fs::create_dir_all(root)?;
65        let mut fence = OpenOptions::new()
66            .read(true)
67            .write(true)
68            .create(true)
69            .truncate(false)
70            .open(root.join(format!("writer-{}.fence", config.writer)))?;
71        let held = match fence.try_lock() {
72            Ok(()) => true,
73            Err(TryLockError::WouldBlock) => false,
74            Err(TryLockError::Error(e)) => return Err(e.into()),
75        };
76        let mut bytes = Vec::new();
77        fence.read_to_end(&mut bytes)?;
78        let generation = if bytes.is_empty() && held {
79            // Reserve the store even before the first edit. Resetting an old
80            // allocation is never an implicit recovery policy.
81            fence.write_all(&config.writers.to_le_bytes())?;
82            fence.write_all(&config.writer.to_le_bytes())?;
83            fence.write_all(&1u64.to_le_bytes())?;
84            fence.sync_all()?;
85            1
86        } else {
87            if bytes.len() != 24 {
88                return Err(LocalError::RecoveryRequired);
89            }
90            let word = |i| u64::from_le_bytes(bytes[i..i + 8].try_into().unwrap());
91            if word(0) != config.writers || word(8) != config.writer {
92                return Err(LocalError::Configuration);
93            }
94            if held {
95                return Err(LocalError::RecoveryRequired);
96            }
97            word(16)
98        };
99        Ok(Self {
100            config,
101            fence,
102            held,
103            generation,
104            log: EventLog::for_crdt(&state),
105            state,
106            last_sequence: 0,
107        })
108    }
109
110    pub fn state(&self) -> &C {
111        &self.state
112    }
113    pub fn log(&self) -> &EventLog<C::Delta> {
114        &self.log
115    }
116    pub fn ticket(&self) -> WriteTicket {
117        WriteTicket(self.generation)
118    }
119    /// Ordinary allocation metadata, independent of the lock generation.
120    pub fn allocation_bytes(&self) -> Vec<u8> {
121        [
122            self.config.writers.to_le_bytes(),
123            self.config.writer.to_le_bytes(),
124            self.last_sequence.to_le_bytes(),
125        ]
126        .concat()
127    }
128    fn context(&self, ticket: WriteTicket, local: bool) -> WriteContext {
129        WriteContext {
130            config: self.config,
131            held: self.held,
132            generation: ticket.0,
133            current_generation: self.generation,
134            local,
135        }
136    }
137
138    /// Revoke outstanding tickets while retaining the same exclusive OS lock.
139    /// I/O uncertainty disables writes. This does not recover a replica.
140    pub fn renew(&mut self, ticket: WriteTicket) -> Result<WriteTicket, LocalError> {
141        if refuses(
142            self.context(ticket, true),
143            RecordId {
144                replica: self.config.writer,
145                sequence: 1,
146            },
147            OwnedPayload::Remove,
148        ) {
149            return Err(LocalError::Refused);
150        }
151        let next = next_sequence(self.generation).ok_or(LocalError::Exhausted)?;
152        self.held = false;
153        self.fence.seek(SeekFrom::Start(16))?;
154        self.fence.write_all(&next.to_le_bytes())?;
155        self.fence.sync_all()?;
156        self.generation = next;
157        self.held = true;
158        Ok(self.ticket())
159    }
160
161    /// Validate ownership and the local lease BEFORE calling M1 admission.
162    /// Remote redelivery and reordering do not require the receiver to own the
163    /// author's coordinate. They require the record to carry its author's space.
164    pub fn receive(
165        &mut self,
166        ticket: WriteTicket,
167        record: Record<C::Delta>,
168    ) -> Result<Admission, LocalError> {
169        self.admit(ticket, record, false)
170    }
171    fn admit(
172        &mut self,
173        ticket: WriteTicket,
174        record: Record<C::Delta>,
175        local: bool,
176    ) -> Result<Admission, LocalError> {
177        self.admit_committed(ticket, record, local, |_, _| Ok(()))
178    }
179    fn admit_committed(
180        &mut self,
181        ticket: WriteTicket,
182        record: Record<C::Delta>,
183        local: bool,
184        commit: impl FnOnce(&EventLog<C::Delta>, u64) -> Result<(), LocalError>,
185    ) -> Result<Admission, LocalError> {
186        if refuses(
187            self.context(ticket, local),
188            record.id,
189            record.delta.owned_payload(),
190        ) {
191            return Err(LocalError::Refused);
192        }
193        let mut candidate = self.log.clone();
194        let outcome = candidate.insert_record(record.clone());
195        if outcome != Admission::Accepted {
196            return Ok(outcome);
197        }
198        let sequence = if record.id.replica == self.config.writer {
199            self.last_sequence.max(record.id.sequence)
200        } else {
201            self.last_sequence
202        };
203        // Keep the OS lock but revoke writes before any fallible persistence.
204        // An error (including an ambiguous rename/sync) cannot be renewed away.
205        self.held = false;
206        commit(&candidate, sequence)?;
207        self.state.apply_delta(record.delta);
208        self.log = candidate;
209        self.last_sequence = sequence;
210        self.held = true;
211        Ok(outcome)
212    }
213    /// Append an existing delta in the configured writer's space. No allocation
214    /// or state mutation occurs on refusal, exhaustion, duplicate or collision.
215    pub fn append(
216        &mut self,
217        ticket: WriteTicket,
218        delta: C::Delta,
219    ) -> Result<Record<C::Delta>, LocalError> {
220        let sequence = next_sequence(self.last_sequence).ok_or(LocalError::Exhausted)?;
221        let record = Record {
222            id: RecordId {
223                replica: self.config.writer,
224                sequence,
225            },
226            delta,
227        };
228        match self.admit(ticket, record.clone(), true)? {
229            Admission::Accepted => Ok(record),
230            _ => Err(LocalError::Refused),
231        }
232    }
233}
234
235impl LocalReplica<GCounter> {
236    pub fn counter(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
237        config.validate().map_err(|_| LocalError::Configuration)?;
238        let n = usize::try_from(config.writers).map_err(|_| LocalError::Configuration)?;
239        Self::fresh(root, config, GCounter::new(n))
240    }
241    pub fn bump(
242        &mut self,
243        ticket: WriteTicket,
244        tally: u64,
245    ) -> Result<Record<GCounterDelta>, LocalError> {
246        self.append(
247            ticket,
248            GCounterDelta {
249                replica: self.config.writer as usize,
250                tally,
251            },
252        )
253    }
254}
255impl LocalReplica<OrSet<String, u64>> {
256    pub fn utf8_set(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
257        Self::fresh(root, config, OrSet::new())
258    }
259    pub fn add(
260        &mut self,
261        ticket: WriteTicket,
262        element: String,
263    ) -> Result<Record<OrSetDelta<String, u64>>, LocalError> {
264        let sequence = next_sequence(self.last_sequence).ok_or(LocalError::Exhausted)?;
265        let token = allocate_token(self.config.writers, self.config.writer, sequence)
266            .ok_or(LocalError::Exhausted)?;
267        self.append(ticket, OrSetDelta::Add { element, token })
268    }
269    pub fn remove(
270        &mut self,
271        ticket: WriteTicket,
272        element: &String,
273    ) -> Result<Record<OrSetDelta<String, u64>>, LocalError> {
274        self.append(
275            ticket,
276            OrSetDelta::Remove {
277                tokens: self.state.observed_tokens(element).into_iter().collect(),
278            },
279        )
280    }
281}
282
283#[path = "persistence.rs"]
284mod persistence;
285
286/// One committed local transaction. The suffix is the existing, unchanged log
287/// wire encoding; the 24-byte prefix is LocalReplica::allocation_bytes(). This
288/// storage container is not a new transport encoding. Reading it grants no
289/// write lease and does not implement packet C's checked writable restart.
290#[derive(Debug, PartialEq, Eq)]
291pub struct CommittedTransaction {
292    pub config: WriterConfig,
293    pub last_sequence: u64,
294    pub log_bytes: Vec<u8>,
295}
296impl CommittedTransaction {
297    pub fn read(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
298        config.validate().map_err(|_| LocalError::Configuration)?;
299        let bytes = fs::read(transaction_path(root, config))?;
300        if bytes.len() < 24 {
301            return Err(LocalError::RecoveryRequired);
302        }
303        let word = |i| u64::from_le_bytes(bytes[i..i + 8].try_into().unwrap());
304        if word(0) != config.writers || word(8) != config.writer {
305            return Err(LocalError::Configuration);
306        }
307        Ok(Self {
308            config,
309            last_sequence: word(16),
310            log_bytes: bytes[24..].to_vec(),
311        })
312    }
313}
314fn transaction_path(root: &Path, config: WriterConfig) -> PathBuf {
315    root.join(format!("writer-{}.transaction", config.writer))
316}
317
318/// Additive durable API for Linux local filesystems. The root must already
319/// exist durably and remain in place; all writers use the same fixed root and
320/// configuration. Each Accepted/Ok(record) follows full transaction replacement
321/// and file + directory sync. Any persistence error permanently disables this
322/// instance's writes, retaining its fence until drop. Use the explicit restart
323/// constructors for existing stores; fresh constructors never reset a store.
324pub struct DurableReplica<C: Crdt> {
325    inner: LocalReplica<C>,
326    path: PathBuf,
327}
328impl<C: Crdt> DurableReplica<C>
329where
330    C::Delta: OwnedDelta + Clone + PartialEq + WireEncode + WireSchema,
331{
332    fn fresh(root: &Path, config: WriterConfig, state: C) -> Result<Self, LocalError> {
333        let root = root.canonicalize()?;
334        // Never overwrite a transaction whose fence is missing.
335        match fs::metadata(transaction_path(&root, config)) {
336            Ok(_) => return Err(LocalError::RecoveryRequired),
337            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
338            Err(e) => return Err(e.into()),
339        }
340        let inner = LocalReplica::fresh(&root, config, state)?;
341        if !inner.held {
342            return Err(LocalError::Refused);
343        }
344        let path = transaction_path(&root, config);
345        let mut replica = Self { inner, path };
346        replica.inner.held = false;
347        Self::commit(&replica.path, config, &replica.inner.log, 0)?;
348        replica.inner.held = true;
349        Ok(replica)
350    }
351    fn commit(
352        path: &Path,
353        config: WriterConfig,
354        log: &EventLog<C::Delta>,
355        sequence: u64,
356    ) -> Result<(), LocalError> {
357        let mut bytes = [
358            config.writers.to_le_bytes(),
359            config.writer.to_le_bytes(),
360            sequence.to_le_bytes(),
361        ]
362        .concat();
363        bytes.extend(
364            log.to_wire_bytes()
365                .map_err(|e| LocalError::Io(io::Error::other(format!("{e:?}"))))?,
366        );
367        persistence::replace(path, &bytes)?;
368        Ok(())
369    }
370    pub fn state(&self) -> &C {
371        self.inner.state()
372    }
373    pub fn log(&self) -> &EventLog<C::Delta> {
374        self.inner.log()
375    }
376    pub fn allocation_bytes(&self) -> Vec<u8> {
377        self.inner.allocation_bytes()
378    }
379    pub fn ticket(&self) -> WriteTicket {
380        self.inner.ticket()
381    }
382    pub fn renew(&mut self, ticket: WriteTicket) -> Result<WriteTicket, LocalError> {
383        self.inner.renew(ticket)
384    }
385    fn admit(
386        &mut self,
387        ticket: WriteTicket,
388        record: Record<C::Delta>,
389        local: bool,
390    ) -> Result<Admission, LocalError> {
391        let path = &self.path;
392        let config = self.inner.config;
393        let outcome = self
394            .inner
395            .admit_committed(ticket, record, local, |log, sequence| {
396                Self::commit(path, config, log, sequence)
397            })?;
398        #[cfg(test)]
399        persistence::checkpoint(7)?;
400        Ok(outcome)
401    }
402    pub fn receive(
403        &mut self,
404        ticket: WriteTicket,
405        record: Record<C::Delta>,
406    ) -> Result<Admission, LocalError> {
407        self.admit(ticket, record, false)
408    }
409    pub fn append(
410        &mut self,
411        ticket: WriteTicket,
412        delta: C::Delta,
413    ) -> Result<Record<C::Delta>, LocalError> {
414        let sequence = next_sequence(self.inner.last_sequence).ok_or(LocalError::Exhausted)?;
415        let record = Record {
416            id: RecordId {
417                replica: self.inner.config.writer,
418                sequence,
419            },
420            delta,
421        };
422        match self.admit(ticket, record.clone(), true)? {
423            Admission::Accepted => Ok(record),
424            _ => Err(LocalError::Refused),
425        }
426    }
427}
428impl<C: Crdt> DurableReplica<C>
429where
430    C::Delta: OwnedDelta + Clone + PartialEq + WireEncode + WireDecode + WireSchema,
431{
432    fn restart(root: &Path, config: WriterConfig, state: C) -> Result<Self, LocalError> {
433        config.validate().map_err(|_| LocalError::Configuration)?;
434        let root = root.canonicalize()?;
435        // Opening without create is deliberate: missing ownership is not a new store.
436        let mut fence = OpenOptions::new()
437            .read(true)
438            .write(true)
439            .open(root.join(format!("writer-{}.fence", config.writer)))?;
440        match fence.try_lock() {
441            Ok(()) => {}
442            Err(TryLockError::WouldBlock) => return Err(LocalError::Refused),
443            Err(TryLockError::Error(e)) => return Err(e.into()),
444        }
445        let mut bytes = Vec::new();
446        fence.read_to_end(&mut bytes)?;
447        if bytes.len() != 24 {
448            return Err(LocalError::RecoveryRequired);
449        }
450        let word = |i| u64::from_le_bytes(bytes[i..i + 8].try_into().unwrap());
451        if word(0) != config.writers || word(8) != config.writer {
452            return Err(LocalError::Configuration);
453        }
454        let generation = word(16);
455        if generation == 0 {
456            return Err(LocalError::RecoveryRequired);
457        }
458        // The lock covers reading, checking and replaying the complete transaction.
459        let transaction = CommittedTransaction::read(&root, config)?;
460        let log = EventLog::<C::Delta>::from_wire_bytes_for(&transaction.log_bytes, &state)
461            .map_err(LocalError::History)?;
462        let mut inner = LocalReplica {
463            config,
464            fence,
465            held: true,
466            generation,
467            log: EventLog::for_crdt(&state),
468            state,
469            last_sequence: 0,
470        };
471        // Compose packet A's corpus-bound ownedStep with M1 admission/replay.
472        // This candidate is private until every record and allocation check passes.
473        for record in log.records() {
474            if inner.admit(inner.ticket(), record.clone(), false)? != Admission::Accepted {
475                return Err(LocalError::InvalidHistory);
476            }
477        }
478        if inner.last_sequence != transaction.last_sequence {
479            return Err(LocalError::InvalidHistory);
480        }
481        // A ticket from before restart must not authorize the newly acquired lease.
482        inner.renew(inner.ticket())?;
483        Ok(Self {
484            inner,
485            path: transaction_path(&root, config),
486        })
487    }
488}
489
490impl DurableReplica<GCounter> {
491    /// Reacquire ownership, validate the committed history and replay fresh state.
492    /// Any error returns no replica and grants no write ticket.
493    pub fn restart_counter(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
494        config.validate().map_err(|_| LocalError::Configuration)?;
495        let n = usize::try_from(config.writers).map_err(|_| LocalError::Configuration)?;
496        Self::restart(root, config, GCounter::new(n))
497    }
498    pub fn counter(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
499        config.validate().map_err(|_| LocalError::Configuration)?;
500        let n = usize::try_from(config.writers).map_err(|_| LocalError::Configuration)?;
501        Self::fresh(root, config, GCounter::new(n))
502    }
503    pub fn bump(
504        &mut self,
505        ticket: WriteTicket,
506        tally: u64,
507    ) -> Result<Record<GCounterDelta>, LocalError> {
508        self.append(
509            ticket,
510            GCounterDelta {
511                replica: self.inner.config.writer as usize,
512                tally,
513            },
514        )
515    }
516}
517impl DurableReplica<OrSet<String, u64>> {
518    /// Checked ordinary restart, including tombstones and token allocation.
519    pub fn restart_utf8_set(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
520        Self::restart(root, config, OrSet::new())
521    }
522    pub fn utf8_set(root: &Path, config: WriterConfig) -> Result<Self, LocalError> {
523        Self::fresh(root, config, OrSet::new())
524    }
525    pub fn add(
526        &mut self,
527        ticket: WriteTicket,
528        element: String,
529    ) -> Result<Record<OrSetDelta<String, u64>>, LocalError> {
530        let sequence = next_sequence(self.inner.last_sequence).ok_or(LocalError::Exhausted)?;
531        let token = allocate_token(
532            self.inner.config.writers,
533            self.inner.config.writer,
534            sequence,
535        )
536        .ok_or(LocalError::Exhausted)?;
537        self.append(ticket, OrSetDelta::Add { element, token })
538    }
539    pub fn remove(
540        &mut self,
541        ticket: WriteTicket,
542        element: &String,
543    ) -> Result<Record<OrSetDelta<String, u64>>, LocalError> {
544        self.append(
545            ticket,
546            OrSetDelta::Remove {
547                tokens: self
548                    .inner
549                    .state
550                    .observed_tokens(element)
551                    .into_iter()
552                    .collect(),
553            },
554        )
555    }
556}
557
558#[cfg(test)]
559mod durable_tests {
560    use super::*;
561    use alloc::string::ToString;
562    use std::{
563        process::Command,
564        sync::atomic::{AtomicU64, Ordering},
565    };
566
567    // IDs are scoped to a CRDT log; tokens are scoped to its set. Capture
568    // allocations at issuance, not received copies (which intentionally repeat).
569    fn unique<T: PartialEq>(values: &[T]) -> bool {
570        values
571            .iter()
572            .enumerate()
573            .all(|(i, x)| !values[..i].contains(x))
574    }
575    fn allocations<D>(records: &[Record<D>]) -> Vec<RecordId> {
576        records.iter().map(|r| r.id).collect()
577    }
578    fn tokens(records: &[Record<OrSetDelta<String, u64>>]) -> Vec<u64> {
579        records
580            .iter()
581            .filter_map(|r| match r.delta {
582                OrSetDelta::Add { token, .. } => Some(token),
583                OrSetDelta::Remove { .. } => None,
584            })
585            .collect()
586    }
587    fn check_allocations<T: PartialEq + Clone>(name: &str, before: &[T], after: &[T]) {
588        assert!(!before.is_empty() && !after.is_empty());
589        let mut all = before.to_vec();
590        all.extend_from_slice(after);
591        assert!(unique(&all), "{name}: allocation reused");
592        assert!(!before.iter().any(|x| after.contains(x)));
593        all.push(before[0].clone());
594        assert!(!unique(&all), "{name}: planted duplicate escaped collector");
595        std::println!(
596            "{name} before={} after={} overlap=0 planted-duplicate=detected",
597            before.len(),
598            after.len()
599        );
600    }
601    fn joined_exchange<C>(
602        a: &mut DurableReplica<C>,
603        b: &mut DurableReplica<C>,
604        empty: impl Fn() -> C,
605    ) where
606        C: Crdt + PartialEq + std::fmt::Debug,
607        C::Delta: OwnedDelta + Clone + PartialEq + WireEncode + WireDecode + WireSchema,
608    {
609        let ab: Vec<_> = a
610            .log()
611            .since(b.log().version())
612            .iter()
613            .map(|r| r.to_wire_bytes().unwrap())
614            .collect();
615        let ba: Vec<_> = b
616            .log()
617            .since(a.log().version())
618            .iter()
619            .map(|r| r.to_wire_bytes().unwrap())
620            .collect();
621        assert!(!ab.is_empty() && !ba.is_empty());
622        // Same two batches, opposite delivery orders, through M1 admission.
623        let mut finals = Vec::new();
624        for batches in [[&ab, &ba], [&ba, &ab]] {
625            let mut state = empty();
626            let mut log = EventLog::for_crdt(&state);
627            for batch in batches {
628                for packet in batch {
629                    let record = Record::<C::Delta>::from_wire_bytes(packet).unwrap();
630                    assert_eq!(
631                        log.admit_with(record, |d| state.apply_delta(d.clone())),
632                        Admission::Accepted
633                    );
634                }
635            }
636            finals.push(state);
637        }
638        assert_eq!(finals[0], finals[1]);
639        for (receiver, packets) in [(b, ab), (a, ba)] {
640            for packet in packets {
641                assert_eq!(
642                    receiver
643                        .receive(
644                            receiver.ticket(),
645                            Record::<C::Delta>::from_wire_bytes(&packet).unwrap()
646                        )
647                        .unwrap(),
648                    Admission::Accepted
649                );
650            }
651            assert_eq!(receiver.state(), &finals[0]);
652        }
653    }
654    fn joined_config(writer: u64) -> WriterConfig {
655        WriterConfig { writers: 2, writer }
656    }
657    fn joined_finish(root: &Path, loss: bool) {
658        let counter_root = root.join("counter");
659        let set_root = root.join("set");
660        let before_c = EventLog::<GCounterDelta>::from_wire_bytes_for(
661            &fs::read(root.join("counter-issued")).unwrap(),
662            &GCounter::new(2),
663        )
664        .unwrap();
665        let before_s = EventLog::<OrSetDelta<String, u64>>::from_wire_bytes_for(
666            &fs::read(root.join("set-issued")).unwrap(),
667            &OrSet::new(),
668        )
669        .unwrap();
670        assert_eq!(before_c.records().len(), 2);
671        assert_eq!(before_s.records().len(), 2);
672        let mut a = DurableReplica::restart_counter(&counter_root, joined_config(0)).unwrap();
673        let mut sa = DurableReplica::restart_utf8_set(&set_root, joined_config(0)).unwrap();
674        if loss {
675            assert_eq!(a.state().state(), &[0, 0]);
676            assert!(!sa.state().contains(&"café☕".into()));
677            assert!(!sa.state().contains(&"東京".into()));
678            assert!(a.log().records().is_empty() && sa.log().records().is_empty());
679            std::println!(
680                "joined LOSS: acknowledged counter=9 and UTF-8 edits absent after process restart"
681            );
682            return;
683        }
684        assert_eq!(a.state().state(), &[9, 0]);
685        for word in ["café☕", "東京"] {
686            assert!(sa.state().contains(&word.into()));
687        }
688        let mut b = DurableReplica::restart_counter(&counter_root, joined_config(1)).unwrap();
689        let mut sb = DurableReplica::restart_utf8_set(&set_root, joined_config(1)).unwrap();
690        let after_c = [
691            a.bump(a.ticket(), 12).unwrap(),
692            b.bump(b.ticket(), 7).unwrap(),
693        ];
694        let after_s = [
695            sa.add(sa.ticket(), "naïve".into()).unwrap(),
696            sb.add(sb.ticket(), "γειά".into()).unwrap(),
697        ];
698        check_allocations(
699            "counter IDs",
700            &allocations(before_c.records()),
701            &allocations(&after_c),
702        );
703        check_allocations(
704            "set IDs",
705            &allocations(before_s.records()),
706            &allocations(&after_s),
707        );
708        check_allocations("set tokens", &tokens(before_s.records()), &tokens(&after_s));
709        joined_exchange(&mut a, &mut b, || GCounter::new(2));
710        joined_exchange(&mut sa, &mut sb, OrSet::new);
711        assert_eq!(a.state(), b.state());
712        assert_eq!(a.state().state(), &[12, 7]);
713        assert_eq!(sa.state(), sb.state());
714        for word in ["café☕", "東京", "naïve", "γειά"] {
715            assert!(sa.state().contains(&word.into()));
716        }
717        for r in before_c.records() {
718            assert!(a.log().records().contains(r) && b.log().records().contains(r));
719        }
720        for r in before_s.records() {
721            assert!(sa.log().records().contains(r) && sb.log().records().contains(r));
722        }
723        let state_c = a.state().clone();
724        let state_s = sa.state().clone();
725        drop((a, b, sa, sb));
726        for writer in 0..2 {
727            assert_eq!(
728                DurableReplica::restart_counter(&counter_root, joined_config(writer))
729                    .unwrap()
730                    .state(),
731                &state_c
732            );
733            assert_eq!(
734                DurableReplica::restart_utf8_set(&set_root, joined_config(writer))
735                    .unwrap()
736                    .state(),
737                &state_s
738            );
739        }
740        std::println!("joined HAPPY: all acknowledged records survive; both exchange orders converge; second restart survives");
741    }
742
743    fn joined_start(root: &Path, loss: bool) {
744        for name in ["counter", "set"] {
745            fs::create_dir_all(root.join(name)).unwrap();
746        }
747        // Make the newly created store directories durable before edits.
748        fs::File::open(root).unwrap().sync_all().unwrap();
749        // Both replicas exist before the offline edits; neither exchanges yet.
750        let _b = DurableReplica::counter(&root.join("counter"), joined_config(1)).unwrap();
751        let _sb = DurableReplica::utf8_set(&root.join("set"), joined_config(1)).unwrap();
752        let mut c = DurableReplica::counter(&root.join("counter"), joined_config(0)).unwrap();
753        let mut s = DurableReplica::utf8_set(&root.join("set"), joined_config(0)).unwrap();
754        if loss {
755            // Deliberately bypass only the commit: same owned allocation and
756            // admission, falsely acknowledged by this negative-control caller.
757            c.inner.bump(c.ticket(), 5).unwrap();
758            c.inner.bump(c.ticket(), 9).unwrap();
759            s.inner.add(s.ticket(), "café☕".into()).unwrap();
760            s.inner.add(s.ticket(), "東京".into()).unwrap();
761        } else {
762            c.bump(c.ticket(), 5).unwrap();
763            c.bump(c.ticket(), 9).unwrap();
764            s.add(s.ticket(), "café☕".into()).unwrap();
765            s.add(s.ticket(), "東京".into()).unwrap();
766        }
767        // External audit only: restart never reads these files as recovery data.
768        fs::write(
769            root.join("counter-issued"),
770            c.log().to_wire_bytes().unwrap(),
771        )
772        .unwrap();
773        fs::write(root.join("set-issued"), s.log().to_wire_bytes().unwrap()).unwrap();
774        std::println!("joined ACK counter=5,9 set=café☕,東京 loss={loss}");
775        std::io::stdout().flush().unwrap();
776        // Intentionally bypass destructors: a real process ends after ACK.
777        std::process::exit(77);
778    }
779    static SERIAL: AtomicU64 = AtomicU64::new(0);
780    fn root() -> PathBuf {
781        let root = std::env::temp_dir().join(format!(
782            "durable-{}-{}",
783            std::process::id(),
784            SERIAL.fetch_add(1, Ordering::Relaxed)
785        ));
786        fs::create_dir_all(&root).unwrap();
787        root
788    }
789    fn config() -> WriterConfig {
790        WriterConfig {
791            writers: 2,
792            writer: 0,
793        }
794    }
795    fn read(root: &Path) -> CommittedTransaction {
796        CommittedTransaction::read(root, config()).unwrap()
797    }
798    fn fault(boundary: u8, crash: bool) {
799        persistence::FAULT.with(|f| f.set((boundary, crash)));
800    }
801    fn exercise(
802        kind: &str,
803        root: &Path,
804        boundary: u8,
805    ) -> (CommittedTransaction, CommittedTransaction) {
806        let old;
807        if kind == "counter" {
808            let mut r = DurableReplica::counter(root, config()).unwrap();
809            r.bump(r.ticket(), 5).unwrap();
810            old = read(root);
811            fault(boundary, true);
812            assert_eq!(r.bump(r.ticket(), 9).unwrap().id.sequence, 2);
813            assert_eq!(r.state().state(), &[9, 0]);
814        } else {
815            let mut r = DurableReplica::utf8_set(root, config()).unwrap();
816            r.add(r.ticket(), "café☕".into()).unwrap();
817            old = read(root);
818            fault(boundary, true);
819            assert_eq!(
820                r.remove(r.ticket(), &"café☕".into()).unwrap().id.sequence,
821                2
822            );
823            assert!(!r.state().contains(&"café☕".into()));
824            assert_eq!(r.state().tombstones().len(), 1);
825        }
826        // The public call has returned success to its caller.
827        std::println!("ACK {kind}");
828        std::io::stdout().flush().unwrap();
829        persistence::checkpoint(8).unwrap();
830        (old, read(root))
831    }
832    fn replay(kind: &str, transaction: &CommittedTransaction) {
833        assert_eq!(transaction.config, config());
834        assert_eq!(transaction.last_sequence, 2);
835        if kind == "counter" {
836            let mut state = GCounter::new(2);
837            let log =
838                EventLog::<GCounterDelta>::from_wire_bytes_for(&transaction.log_bytes, &state)
839                    .unwrap();
840            assert_eq!(log.records().len(), 2);
841            assert_eq!(log.to_wire_bytes().unwrap(), transaction.log_bytes);
842            for record in log.records() {
843                state.apply_delta(record.delta.clone());
844            }
845            assert_eq!(state.state(), &[9, 0]);
846        } else {
847            let mut state = OrSet::<String, u64>::new();
848            let log = EventLog::<OrSetDelta<String, u64>>::from_wire_bytes_for(
849                &transaction.log_bytes,
850                &state,
851            )
852            .unwrap();
853            assert_eq!(log.records().len(), 2);
854            assert_eq!(log.to_wire_bytes().unwrap(), transaction.log_bytes);
855            for record in log.records() {
856                state.apply_delta(record.delta.clone());
857            }
858            assert!(!state.contains(&"café☕".into()));
859            assert_eq!(
860                state.tombstones().iter().copied().collect::<Vec<_>>(),
861                alloc::vec![2]
862            );
863        }
864    }
865    #[test]
866    fn restart_still_works_and_fresh_allocation() {
867        let root = root();
868        let mut r = DurableReplica::counter(&root, config()).unwrap();
869        let before = r.bump(r.ticket(), 5).unwrap();
870        r.receive(
871            r.ticket(),
872            Record {
873                id: RecordId {
874                    replica: 1,
875                    sequence: 40,
876                },
877                delta: GCounterDelta {
878                    replica: 1,
879                    tally: 7,
880                },
881            },
882        )
883        .unwrap();
884        let old = r.ticket();
885        let state = r.state().state().to_vec();
886        let log = r.log().to_wire_bytes().unwrap();
887        let allocation = r.allocation_bytes();
888        drop(r);
889        let mut r = DurableReplica::restart_counter(&root, config()).unwrap();
890        assert_eq!(r.state().state(), state);
891        assert_eq!(r.log().to_wire_bytes().unwrap(), log);
892        assert_eq!(r.allocation_bytes(), allocation);
893        assert!(matches!(r.bump(old, 9), Err(LocalError::Refused)));
894        let after = r.bump(r.ticket(), 9).unwrap();
895        assert_eq!(
896            before.id,
897            RecordId {
898                replica: 0,
899                sequence: 1
900            }
901        );
902        assert_eq!(
903            after.id,
904            RecordId {
905                replica: 0,
906                sequence: 2
907            }
908        );
909        assert_eq!(r.state().state(), &[9, 7]);
910        std::println!(
911            "controls=7,2 counter replay={state:?} writable=true IDs {:?} -> {:?}",
912            before.id,
913            after.id
914        );
915        drop(r);
916        assert_eq!(
917            DurableReplica::restart_counter(&root, config())
918                .unwrap()
919                .state()
920                .state(),
921            &[9, 7]
922        );
923
924        let root = self::root();
925        let mut r = DurableReplica::utf8_set(&root, config()).unwrap();
926        let before = r.add(r.ticket(), "café☕".into()).unwrap();
927        let removed = r.remove(r.ticket(), &"café☕".into()).unwrap();
928        r.receive(
929            r.ticket(),
930            Record {
931                id: RecordId {
932                    replica: 1,
933                    sequence: 40,
934                },
935                delta: OrSetDelta::Add {
936                    element: "東京".into(),
937                    token: 81,
938                },
939            },
940        )
941        .unwrap();
942        let state = r.state().clone();
943        let log = r.log().to_wire_bytes().unwrap();
944        let allocation = r.allocation_bytes();
945        let old = r.ticket();
946        drop(r);
947        let mut r = DurableReplica::restart_utf8_set(&root, config()).unwrap();
948        assert_eq!(r.state(), &state);
949        assert_eq!(r.log().to_wire_bytes().unwrap(), log);
950        assert_eq!(r.allocation_bytes(), allocation);
951        assert!(matches!(
952            r.add(old, "stale".into()),
953            Err(LocalError::Refused)
954        ));
955        let after = r.add(r.ticket(), "café☕".into()).unwrap();
956        assert_eq!(
957            before.id,
958            RecordId {
959                replica: 0,
960                sequence: 1
961            }
962        );
963        assert_eq!(
964            removed.id,
965            RecordId {
966                replica: 0,
967                sequence: 2
968            }
969        );
970        assert_eq!(
971            after.id,
972            RecordId {
973                replica: 0,
974                sequence: 3
975            }
976        );
977        assert!(matches!(before.delta, OrSetDelta::Add { token: 2, .. }));
978        assert!(matches!(after.delta, OrSetDelta::Add { token: 6, .. }));
979        assert!(r.state().contains(&"café☕".into()));
980        assert!(r.state().contains(&"東京".into()));
981        assert!(r.state().tombstones().contains(&2));
982        std::println!("controls=7,2 set replay=equal including tombstone 2 and remote token 81; writable=true IDs {:?}, {:?} -> {:?}; tokens 2 -> 6", before.id, removed.id, after.id);
983        drop(r);
984        let r = DurableReplica::restart_utf8_set(&root, config()).unwrap();
985        assert!(r.state().contains(&"café☕".into()));
986        assert!(r.state().tombstones().contains(&2));
987    }
988    fn initialized(kind: &str) -> PathBuf {
989        let root = root();
990        if kind == "counter" {
991            let mut r = DurableReplica::counter(&root, config()).unwrap();
992            r.bump(r.ticket(), 5).unwrap();
993        } else {
994            let mut r = DurableReplica::utf8_set(&root, config()).unwrap();
995            r.add(r.ticket(), "café☕".into()).unwrap();
996        }
997        root
998    }
999    // Success proves the returned replica can commit a fresh edit.
1000    fn restart_and_write(kind: &str, root: &Path, config: WriterConfig) -> Result<(), LocalError> {
1001        if kind == "counter" {
1002            let mut r = DurableReplica::restart_counter(root, config)?;
1003            r.bump(r.ticket(), 9)?;
1004        } else {
1005            let mut r = DurableReplica::restart_utf8_set(root, config)?;
1006            r.add(r.ticket(), "new".into())?;
1007        }
1008        Ok(())
1009    }
1010    #[test]
1011    fn restart_corrupt_history() {
1012        for kind in ["counter", "set"] {
1013            let root = initialized(kind);
1014            let path = transaction_path(&root, config());
1015            let mut bytes = fs::read(&path).unwrap();
1016            // Damage the checksum; the well-formed payload permits revert control 8.
1017            *bytes.last_mut().unwrap() ^= 1;
1018            fs::write(&path, &bytes).unwrap();
1019            let result = restart_and_write(kind, &root, config());
1020            std::println!(
1021                "control=3 {kind} result={result:?} writable={}",
1022                result.is_ok()
1023            );
1024            assert!(matches!(
1025                result,
1026                Err(LocalError::History(WireError::IntegrityMismatch))
1027            ));
1028            assert_eq!(fs::read(&path).unwrap(), bytes);
1029        }
1030    }
1031    #[test]
1032    fn restart_mismatched_history() {
1033        for kind in ["counter", "set"] {
1034            let root = initialized(kind);
1035            let path = transaction_path(&root, config());
1036            let original = fs::read(&path).unwrap();
1037            let mut bytes = original.clone();
1038            bytes[..8].copy_from_slice(&3u64.to_le_bytes());
1039            fs::write(&path, &bytes).unwrap();
1040            let result = restart_and_write(kind, &root, config());
1041            std::println!(
1042                "control=4 {kind} result={result:?} writable={}",
1043                result.is_ok()
1044            );
1045            assert!(matches!(result, Err(LocalError::Configuration)));
1046            assert_eq!(fs::read(&path).unwrap(), bytes);
1047            fs::write(&path, original).unwrap();
1048            assert!(matches!(
1049                restart_and_write(
1050                    kind,
1051                    &root,
1052                    WriterConfig {
1053                        writers: 3,
1054                        writer: 0
1055                    }
1056                ),
1057                Err(LocalError::Configuration)
1058            ));
1059        }
1060        let root = initialized("counter");
1061        assert!(matches!(
1062            DurableReplica::restart_utf8_set(&root, config()),
1063            Err(LocalError::History(WireError::DeltaTypeMismatch))
1064        ));
1065    }
1066    #[test]
1067    fn restart_invalid_history() {
1068        for kind in ["counter", "set"] {
1069            let root = initialized(kind);
1070            let path = transaction_path(&root, config());
1071            // Serialize with the product: valid framing, invalid ownership.
1072            if kind == "counter" {
1073                let mut log = EventLog::for_crdt(&GCounter::new(2));
1074                assert_eq!(
1075                    log.insert_record(Record {
1076                        id: RecordId {
1077                            replica: 0,
1078                            sequence: 1
1079                        },
1080                        delta: GCounterDelta {
1081                            replica: 1,
1082                            tally: 5
1083                        },
1084                    }),
1085                    Admission::Accepted
1086                );
1087                DurableReplica::<GCounter>::commit(&path, config(), &log, 1).unwrap();
1088            } else {
1089                let mut log = EventLog::for_crdt(&OrSet::<String, u64>::new());
1090                assert_eq!(
1091                    log.insert_record(Record {
1092                        id: RecordId {
1093                            replica: 0,
1094                            sequence: 1
1095                        },
1096                        delta: OrSetDelta::Add {
1097                            element: "foreign".into(),
1098                            token: 3
1099                        },
1100                    }),
1101                    Admission::Accepted
1102                );
1103                DurableReplica::<OrSet<String, u64>>::commit(&path, config(), &log, 1).unwrap();
1104            }
1105            let bytes = fs::read(&path).unwrap();
1106            let result = restart_and_write(kind, &root, config());
1107            std::println!(
1108                "control=5 {kind} result={result:?} writable={}",
1109                result.is_ok()
1110            );
1111            assert!(matches!(
1112                result,
1113                Err(LocalError::Refused) | Err(LocalError::History(_))
1114            ));
1115            assert_eq!(fs::read(&path).unwrap(), bytes);
1116            let root = initialized(kind);
1117            let path = transaction_path(&root, config());
1118            let original = fs::read(&path).unwrap();
1119            for last in [0u64, 2, u64::MAX] {
1120                let mut bytes = original.clone();
1121                bytes[16..24].copy_from_slice(&last.to_le_bytes());
1122                fs::write(&path, &bytes).unwrap();
1123                assert!(matches!(
1124                    restart_and_write(kind, &root, config()),
1125                    Err(LocalError::InvalidHistory)
1126                ));
1127                assert_eq!(fs::read(&path).unwrap(), bytes);
1128            }
1129        }
1130    }
1131    #[test]
1132    fn restart_unavailable_ownership() {
1133        let root = root();
1134        let r = DurableReplica::counter(&root, config()).unwrap();
1135        assert!(matches!(
1136            DurableReplica::restart_counter(&root, config()),
1137            Err(LocalError::Refused)
1138        ));
1139        drop(r);
1140        restart_and_write("counter", &root, config()).unwrap();
1141        let root = self::root();
1142        let r = DurableReplica::utf8_set(&root, config()).unwrap();
1143        assert!(matches!(
1144            DurableReplica::restart_utf8_set(&root, config()),
1145            Err(LocalError::Refused)
1146        ));
1147        drop(r);
1148        restart_and_write("set", &root, config()).unwrap();
1149        std::println!("control=6 both primitives: held fence Refused, no replica; released fence and empty histories: writable");
1150    }
1151
1152    #[test]
1153    fn durable_child() {
1154        if let Ok(root) = std::env::var("SAFEMESH_JOINED_ROOT") {
1155            joined_start(
1156                Path::new(&root),
1157                std::env::var_os("SAFEMESH_JOINED_LOSS").is_some(),
1158            );
1159        }
1160        if let Ok(root) = std::env::var("SAFEMESH_DURABLE_ROOT") {
1161            let kind = std::env::var("SAFEMESH_DURABLE_KIND").unwrap();
1162            if std::env::var_os("SAFEMESH_DURABLE_READ").is_some() {
1163                replay(&kind, &read(Path::new(&root)));
1164                return;
1165            }
1166            let boundary = std::env::var("SAFEMESH_DURABLE_BOUNDARY")
1167                .unwrap()
1168                .parse()
1169                .unwrap();
1170            exercise(&kind, Path::new(&root), boundary);
1171        }
1172    }
1173    fn crash(kind: &str, boundary: u8, root: &Path) -> std::process::Output {
1174        let output = Command::new(std::env::current_exe().unwrap())
1175            .args([
1176                "--exact",
1177                "local::durable_tests::durable_child",
1178                "--nocapture",
1179            ])
1180            .env("SAFEMESH_DURABLE_ROOT", root)
1181            .env("SAFEMESH_DURABLE_KIND", kind)
1182            .env("SAFEMESH_DURABLE_BOUNDARY", boundary.to_string())
1183            .output()
1184            .unwrap();
1185        // Save and read child exit codes, too; never derive one through a pipe.
1186        let status = root.join("child.exit");
1187        fs::write(&status, output.status.code().unwrap().to_string()).unwrap();
1188        assert_eq!(fs::read_to_string(status).unwrap(), "77", "{:?}", output);
1189        output
1190    }
1191    #[test]
1192    fn durable_still_works() {
1193        for loss in [false, true] {
1194            let root = root();
1195            let mut command = Command::new(std::env::current_exe().unwrap());
1196            command
1197                .args([
1198                    "--exact",
1199                    "local::durable_tests::durable_child",
1200                    "--nocapture",
1201                ])
1202                .env("SAFEMESH_JOINED_ROOT", &root);
1203            if loss {
1204                command.env("SAFEMESH_JOINED_LOSS", "1");
1205            }
1206            let output = command.output().unwrap();
1207            fs::write(
1208                root.join("joined.exit"),
1209                output.status.code().unwrap().to_string(),
1210            )
1211            .unwrap();
1212            assert_eq!(fs::read_to_string(root.join("joined.exit")).unwrap(), "77");
1213            assert!(String::from_utf8_lossy(&output.stdout).contains("joined ACK"));
1214            joined_finish(&root, loss);
1215        }
1216
1217        for kind in ["counter", "set"] {
1218            let root = root();
1219            let (_, expected) = exercise(kind, &root, 0);
1220            // All writer handles are dropped. Reopen the committed transaction
1221            // and replay fresh state; enabling writes on restart is packet C.
1222            assert_eq!(read(&root), expected);
1223            replay(kind, &read(&root));
1224            let restarted = Command::new(std::env::current_exe().unwrap())
1225                .args([
1226                    "--exact",
1227                    "local::durable_tests::durable_child",
1228                    "--nocapture",
1229                ])
1230                .env("SAFEMESH_DURABLE_ROOT", &root)
1231                .env("SAFEMESH_DURABLE_KIND", kind)
1232                .env("SAFEMESH_DURABLE_READ", "1")
1233                .output()
1234                .unwrap();
1235            let status = root.join("restart.exit");
1236            fs::write(&status, restarted.status.code().unwrap().to_string()).unwrap();
1237            assert_eq!(fs::read_to_string(status).unwrap(), "0", "{restarted:?}");
1238            std::println!("control=6 {kind} ordinary-write/read/restart=PASS");
1239        }
1240    }
1241    #[test]
1242    fn durable_crash_boundaries() {
1243        for kind in ["counter", "set"] {
1244            let (old, new) = exercise(kind, &root(), 0);
1245            for boundary in 1..=8 {
1246                let root = root();
1247                let output = crash(kind, boundary, &root);
1248                let recovered = read(&root);
1249                assert_eq!(&recovered, if boundary < 5 { &old } else { &new });
1250                assert_eq!(
1251                    String::from_utf8_lossy(&output.stdout).contains("ACK"),
1252                    boundary == 8
1253                );
1254                std::println!(
1255                    "control=2 {kind} boundary={boundary}/8 complete={}",
1256                    if boundary < 5 { "old" } else { "new" }
1257                );
1258            }
1259        }
1260    }
1261    fn acknowledged_survives(kind: &str) {
1262        let root = root();
1263        let output = crash(kind, 8, &root);
1264        assert!(String::from_utf8_lossy(&output.stdout).contains("ACK"));
1265        replay(kind, &read(&root));
1266        std::println!("control=3 {kind} acknowledged-crash-recovery=PASS");
1267    }
1268    #[test]
1269    fn durable_acknowledged_survives_counter() {
1270        acknowledged_survives("counter");
1271    }
1272    #[test]
1273    fn durable_acknowledged_survives_set() {
1274        acknowledged_survives("set");
1275    }
1276    fn failure<C: Crdt + std::fmt::Debug>(mut r: DurableReplica<C>, delta: C::Delta, boundary: u8)
1277    where
1278        C::Delta: OwnedDelta + Clone + PartialEq + WireEncode + WireSchema,
1279    {
1280        let state = format!("{:?}", r.state());
1281        let log = r.log().to_wire_bytes().unwrap();
1282        let allocation = r.allocation_bytes();
1283        let ticket = r.ticket();
1284        fault(boundary, false);
1285        assert!(matches!(
1286            r.append(ticket, delta.clone()),
1287            Err(LocalError::Io(_))
1288        ));
1289        fault(0, false);
1290        assert_eq!(format!("{:?}", r.state()), state);
1291        assert_eq!(r.log().to_wire_bytes().unwrap(), log);
1292        assert_eq!(r.allocation_bytes(), allocation);
1293        assert!(matches!(
1294            r.append(ticket, delta.clone()),
1295            Err(LocalError::Refused)
1296        ));
1297        assert!(matches!(
1298            r.receive(
1299                ticket,
1300                Record {
1301                    id: RecordId {
1302                        replica: 0,
1303                        sequence: 1
1304                    },
1305                    delta
1306                }
1307            ),
1308            Err(LocalError::Refused)
1309        ));
1310        assert!(matches!(r.renew(ticket), Err(LocalError::Refused)));
1311    }
1312    #[test]
1313    fn durable_io_failure_disables_writes() {
1314        for boundary in 2..=6 {
1315            failure(
1316                DurableReplica::counter(&root(), config()).unwrap(),
1317                GCounterDelta {
1318                    replica: 0,
1319                    tally: 9,
1320                },
1321                boundary,
1322            );
1323            failure(
1324                DurableReplica::utf8_set(&root(), config()).unwrap(),
1325                OrSetDelta::Add {
1326                    element: "東京".into(),
1327                    token: 2,
1328                },
1329                boundary,
1330            );
1331            std::println!(
1332                "controls=4,5 write/sync/uncertain-boundary={boundary} error+disabled=PASS"
1333            );
1334        }
1335    }
1336    #[test]
1337    fn durable_receive_and_refusal() {
1338        let root = root();
1339        let mut r = DurableReplica::counter(&root, config()).unwrap();
1340        let record = Record {
1341            id: RecordId {
1342                replica: 1,
1343                sequence: 1,
1344            },
1345            delta: GCounterDelta {
1346                replica: 1,
1347                tally: 7,
1348            },
1349        };
1350        assert_eq!(
1351            r.receive(r.ticket(), record.clone()).unwrap(),
1352            Admission::Accepted
1353        );
1354        let transaction = read(&root);
1355        assert_eq!(transaction.last_sequence, 0);
1356        assert_eq!(transaction.log_bytes, r.log().to_wire_bytes().unwrap());
1357        assert_eq!(
1358            r.receive(r.ticket(), record.clone()).unwrap(),
1359            Admission::Duplicate
1360        );
1361        let mut collision = record;
1362        collision.delta.tally = 8;
1363        assert_eq!(
1364            r.receive(r.ticket(), collision).unwrap(),
1365            Admission::Collision
1366        );
1367        assert!(matches!(
1368            r.append(
1369                r.ticket(),
1370                GCounterDelta {
1371                    replica: 1,
1372                    tally: 10
1373                }
1374            ),
1375            Err(LocalError::Refused)
1376        ));
1377        assert_eq!(read(&root), transaction);
1378        let old = r.ticket();
1379        let ticket = r.renew(old).unwrap();
1380        assert!(matches!(r.bump(old, 9), Err(LocalError::Refused)));
1381        r.bump(ticket, 9).unwrap();
1382        assert_eq!(read(&root).last_sequence, 1);
1383        drop(r);
1384        assert!(matches!(
1385            DurableReplica::counter(&root, config()),
1386            Err(LocalError::RecoveryRequired)
1387        ));
1388    }
1389}
safemesh_crdt/local.rs

safemesh_crdt/
local.rs
