1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
//! Last-Writer-Wins key/value CRDT for SessionRoom shared state (GitHub #22).
//!
//! Pure — no I/O, no chain, no crypto. A SessionRoom is an append-only on-chain
//! log of *encrypted* ops ([`crate::kv_room`] seals/opens them); this module is
//! the deterministic fold that turns the decrypted op set into a converged map.
//! Because the merge is a total order, every replica that has seen the same set
//! of ops — in any order, with any duplicates — computes the SAME map. That
//! convergence is the load-bearing correctness property and is what the tests
//! below pin.
use std::collections::BTreeMap;
/// A single key/value write (or delete) appended to a room.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct KvOp {
/// The key being written.
pub key: String,
/// The value, or `None` for a tombstone (delete).
pub value: Option<Vec<u8>>,
/// Logical clock — higher wins. A writer stamps `next_lamport(seen)`.
pub lamport: u64,
/// Writer address (20 bytes) — the first deterministic tiebreak when two ops
/// for the same key share a `lamport` (the `value` breaks a further tie).
pub writer: [u8; 20],
/// Wall-clock seconds when written. Used ONLY for optional TTL filtering,
/// never for ordering (clocks are not trustworthy across writers).
pub ts: u64,
}
/// Does `candidate` beat `current` for the same key? Higher `lamport` wins; on a
/// tie the lexicographically greater `writer` wins; on a FURTHER tie the greater
/// `value` wins (a tombstone `None` sorts below any `Some`, so a concurrent write
/// beats a delete at an equal clock — "add-wins"). The `value` tiebreak is what
/// makes this a genuine TOTAL order over op CONTENT, not just over
/// `(lamport, writer)`: two distinct ops sharing a `(lamport, writer)` — a normal
/// event, since one identity's devices share a `writer` and `next_lamport` can
/// re-stamp the same clock — would otherwise be incomparable, so the FIRST in the
/// log won and the converged map depended on log order (a divergence bug). With
/// the value tiebreak every replica picks the same winner in any order.
fn op_wins(candidate: &KvOp, current: &KvOp) -> bool {
(candidate.lamport, candidate.writer, &candidate.value)
> (current.lamport, current.writer, ¤t.value)
}
/// Fold `ops` into the converged map. A tombstone that wins suppresses its key.
///
/// `ttl_secs == 0` disables TTL filtering; otherwise an op is ignored once
/// `op.ts + ttl_secs <= now` (ephemeral rooms age their state out). TTL is
/// applied BEFORE the LWW merge, so an expired winner does not mask a still-live
/// older write for the same key.
pub fn reduce(ops: &[KvOp], now: u64, ttl_secs: u64) -> BTreeMap<String, Vec<u8>> {
let mut winners: BTreeMap<&str, &KvOp> = BTreeMap::new();
for op in ops {
if ttl_secs != 0 && op.ts.saturating_add(ttl_secs) <= now {
continue;
}
match winners.get(op.key.as_str()) {
Some(current) if !op_wins(op, current) => {}
_ => {
winners.insert(op.key.as_str(), op);
}
}
}
winners
.into_iter()
.filter_map(|(k, op)| op.value.as_ref().map(|v| (k.to_string(), v.clone())))
.collect()
}
/// The lamport a writer should stamp on its next op, given every op it has
/// observed: `max(lamport) + 1`, or `0` for an empty log.
pub fn next_lamport(ops: &[KvOp]) -> u64 {
ops.iter().map(|o| o.lamport).max().map_or(0, |m| m + 1)
}
#[cfg(test)]
mod tests {
use super::*;
fn op(key: &str, val: Option<&[u8]>, lamport: u64, writer: u8, ts: u64) -> KvOp {
KvOp {
key: key.to_string(),
value: val.map(|v| v.to_vec()),
lamport,
writer: [writer; 20],
ts,
}
}
fn map_of(pairs: &[(&str, &[u8])]) -> BTreeMap<String, Vec<u8>> {
pairs
.iter()
.map(|(k, v)| (k.to_string(), v.to_vec()))
.collect()
}
#[test]
fn empty_log_is_empty_map() {
assert!(reduce(&[], 0, 0).is_empty());
assert_eq!(next_lamport(&[]), 0);
}
#[test]
fn higher_lamport_wins() {
let ops = [
op("a", Some(b"old"), 1, 1, 100),
op("a", Some(b"new"), 2, 1, 100),
];
assert_eq!(reduce(&ops, 0, 0), map_of(&[("a", b"new")]));
}
#[test]
fn writer_tiebreak_on_equal_lamport() {
// Same lamport — the greater writer address wins, regardless of order.
let lo = op("a", Some(b"lo"), 5, 1, 100);
let hi = op("a", Some(b"hi"), 5, 9, 100);
assert_eq!(reduce(&[lo.clone(), hi.clone()], 0, 0), map_of(&[("a", b"hi")]));
assert_eq!(reduce(&[hi, lo], 0, 0), map_of(&[("a", b"hi")]));
}
#[test]
fn tombstone_wins_then_loses_by_clock() {
// A newer tombstone deletes the key.
let del = [
op("a", Some(b"v"), 1, 1, 100),
op("a", None, 2, 1, 100),
];
assert!(reduce(&del, 0, 0).is_empty());
// A newer write resurrects it.
let revive = [
op("a", None, 2, 1, 100),
op("a", Some(b"v2"), 3, 1, 100),
];
assert_eq!(reduce(&revive, 0, 0), map_of(&[("a", b"v2")]));
}
#[test]
fn value_tiebreak_on_equal_lamport_and_writer() {
// Two distinct writes sharing (lamport, writer) — a NORMAL event (one
// identity's devices share a writer; `next_lamport` can re-stamp a clock).
// Without the value tiebreak the FIRST in the log won → replicas diverged.
// Now the greater value wins in BOTH orders.
let a = op("k", Some(b"AAA"), 5, 1, 100);
let b = op("k", Some(b"BBB"), 5, 1, 100);
assert_eq!(reduce(&[a.clone(), b.clone()], 0, 0), map_of(&[("k", b"BBB")]));
assert_eq!(reduce(&[b, a], 0, 0), map_of(&[("k", b"BBB")]));
}
#[test]
fn write_beats_tombstone_at_equal_clock_deterministically() {
// write vs delete at the SAME (lamport, writer): add-wins, order-
// independent (previously the key was present or absent by log order).
let write = op("k", Some(b"v"), 9, 7, 100);
let tomb = op("k", None, 9, 7, 100);
assert_eq!(reduce(&[write.clone(), tomb.clone()], 0, 0), map_of(&[("k", b"v")]));
assert_eq!(reduce(&[tomb, write], 0, 0), map_of(&[("k", b"v")]));
}
#[test]
fn converges_under_any_permutation() {
let base = [
op("x", Some(b"1"), 1, 1, 10),
op("y", Some(b"2"), 1, 2, 10),
op("x", Some(b"3"), 3, 2, 20),
op("y", None, 2, 1, 30),
op("z", Some(b"9"), 7, 3, 40),
op("x", Some(b"3"), 3, 2, 20), // exact duplicate (idempotence)
];
let expected = reduce(&base, 0, 0);
// Every rotation must produce the identical map.
for shift in 0..base.len() {
let mut perm = base.to_vec();
perm.rotate_left(shift);
assert_eq!(reduce(&perm, 0, 0), expected, "rotation {shift} diverged");
}
// y was tombstoned (lamport 2) after its only write (lamport 1) → gone.
assert_eq!(expected, map_of(&[("x", b"3"), ("z", b"9")]));
}
#[test]
fn idempotent_apply_twice_equals_once() {
let ops = [
op("a", Some(b"1"), 1, 1, 10),
op("b", Some(b"2"), 2, 2, 10),
];
let once = reduce(&ops, 0, 0);
let doubled: Vec<KvOp> = ops.iter().chain(ops.iter()).cloned().collect();
assert_eq!(reduce(&doubled, 0, 0), once);
}
#[test]
fn ttl_filters_expired_ops() {
let ops = [
op("a", Some(b"stale"), 1, 1, 100),
op("b", Some(b"fresh"), 1, 1, 900),
];
// now=1000, ttl=300 → 'a' expired (100+300<=1000), 'b' kept (900+300>1000).
assert_eq!(reduce(&ops, 1000, 300), map_of(&[("b", b"fresh")]));
// ttl=0 disables the filter — both survive.
assert_eq!(
reduce(&ops, 1000, 0),
map_of(&[("a", b"stale"), ("b", b"fresh")])
);
}
#[test]
fn ttl_expired_winner_does_not_mask_live_older_write() {
// An expired high-lamport write must not delete a still-live lower one.
let ops = [
op("a", Some(b"live"), 1, 1, 950),
op("a", Some(b"expired"), 5, 1, 100),
];
assert_eq!(reduce(&ops, 1000, 300), map_of(&[("a", b"live")]));
}
#[test]
fn next_lamport_is_max_plus_one() {
let ops = [op("a", Some(b"v"), 4, 1, 0), op("b", Some(b"v"), 9, 1, 0)];
assert_eq!(next_lamport(&ops), 10);
}
}