reddb_wire/replication/
change_record.rs

1use serde_json::Value as JsonValue;
2
3use super::util::{hex_decode, hex_encode};
4
5pub const DEFAULT_REPLICATION_TERM: u64 = 1;
6pub type ChangeRecordJsonValue = JsonValue;
7
8pub fn parse_change_record_json_value(text: &str) -> Result<ChangeRecordJsonValue, String> {
9    serde_json::from_str(text).map_err(|err| err.to_string())
10}
11
12pub fn change_record_json_value_to_string(value: &ChangeRecordJsonValue) -> String {
13    serde_json::to_string(value).unwrap_or_else(|_| "null".to_string())
14}
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum ChangeOperation {
18    Insert,
19    Update,
20    Delete,
21    Refresh,
22    /// One committed transaction; payload contains encoded member records.
23    Transaction,
24}
25
26impl ChangeOperation {
27    #[allow(clippy::should_implement_trait)]
28    pub fn from_str(value: &str) -> Option<Self> {
29        Self::from_wire_str(value)
30    }
31
32    pub fn from_wire_str(value: &str) -> Option<Self> {
33        match value {
34            "insert" => Some(Self::Insert),
35            "update" => Some(Self::Update),
36            "delete" => Some(Self::Delete),
37            "refresh" => Some(Self::Refresh),
38            "transaction" => Some(Self::Transaction),
39            _ => None,
40        }
41    }
42
43    pub fn as_wire_str(&self) -> &'static str {
44        match self {
45            Self::Insert => "insert",
46            Self::Update => "update",
47            Self::Delete => "delete",
48            Self::Refresh => "refresh",
49            Self::Transaction => "transaction",
50        }
51    }
52
53    pub fn as_str(&self) -> &'static str {
54        self.as_wire_str()
55    }
56}
57
58#[derive(Debug, Clone)]
59pub struct ChangeRecord {
60    pub term: u64,
61    pub lsn: u64,
62    pub timestamp: u64,
63    pub operation: ChangeOperation,
64    pub collection: String,
65    pub entity_id: u64,
66    pub entity_kind: String,
67    pub entity_bytes: Option<Vec<u8>>,
68    pub metadata: Option<JsonValue>,
69    pub refresh_records: Option<Vec<Vec<u8>>>,
70    /// Issue #991 — stable identity of the range this user-data change
71    /// belongs to (`RangeId`, from the #989 ownership catalog), or `None`
72    /// for records that predate range replication. Carried so replicas and
73    /// recovery can route a change to its range and gate it against that
74    /// range's authority watermark.
75    pub range_id: Option<u64>,
76    /// Issue #991 — the owning range's `OwnershipEpoch` at the moment this
77    /// change was produced. The epoch bumps only when write authority moves
78    /// to a new owner, so a record whose epoch is behind the target range's
79    /// accepted epoch is a write from a deposed owner and is fenced. `None`
80    /// for legacy records and non-range-replicated changes.
81    pub ownership_epoch: Option<u64>,
82}
83
84impl ChangeRecord {
85    /// Older peers reject the new operation before applying any member.
86    pub fn for_transaction(lsn: u64, timestamp: u64, records: Vec<ChangeRecord>) -> Self {
87        let mut record = Self::for_refresh(
88            lsn,
89            timestamp,
90            "",
91            records.iter().map(Self::encode).collect(),
92        );
93        record.operation = ChangeOperation::Transaction;
94        record.entity_kind = "transaction".into();
95        record
96    }
97
98    pub fn for_refresh(
99        lsn: u64,
100        timestamp: u64,
101        collection: impl Into<String>,
102        records: Vec<Vec<u8>>,
103    ) -> Self {
104        Self {
105            term: DEFAULT_REPLICATION_TERM,
106            lsn,
107            timestamp,
108            operation: ChangeOperation::Refresh,
109            collection: collection.into(),
110            entity_id: 0,
111            entity_kind: "refresh".to_string(),
112            entity_bytes: None,
113            metadata: None,
114            refresh_records: Some(records),
115            range_id: None,
116            ownership_epoch: None,
117        }
118    }
119
120    /// Stamp this record with the authority metadata of the range that owns
121    /// it (issue #991): the stable range identity and the owner's current
122    /// ownership epoch. The term is set independently via [`Self::with_term`].
123    pub fn with_range_authority(mut self, range_id: u64, ownership_epoch: u64) -> Self {
124        self.range_id = Some(range_id);
125        self.ownership_epoch = Some(ownership_epoch);
126        self
127    }
128
129    pub fn to_json_value(&self) -> JsonValue {
130        let mut object = serde_json::Map::new();
131        object.insert("term".to_string(), JsonValue::Number(self.term.into()));
132        object.insert("lsn".to_string(), JsonValue::Number(self.lsn.into()));
133        object.insert(
134            "timestamp".to_string(),
135            JsonValue::Number(self.timestamp.into()),
136        );
137        object.insert(
138            "operation".to_string(),
139            JsonValue::String(self.operation.as_wire_str().to_string()),
140        );
141        object.insert(
142            "collection".to_string(),
143            JsonValue::String(self.collection.clone()),
144        );
145        object.insert("rid".to_string(), JsonValue::Number(self.entity_id.into()));
146        object.insert(
147            "kind".to_string(),
148            JsonValue::String(public_item_kind(&self.entity_kind).to_string()),
149        );
150        if let Some(bytes) = &self.entity_bytes {
151            object.insert(
152                "entity_bytes_hex".to_string(),
153                JsonValue::String(hex_encode(bytes)),
154            );
155        }
156        if let Some(metadata) = &self.metadata {
157            object.insert("metadata".to_string(), metadata.clone());
158        }
159        if let Some(records) = &self.refresh_records {
160            let arr = records
161                .iter()
162                .map(|bytes| JsonValue::String(hex_encode(bytes)))
163                .collect();
164            object.insert("refresh_records_hex".to_string(), JsonValue::Array(arr));
165        }
166        // Issue #991 — range authority is omitted entirely when absent so a
167        // non-range-replicated record stays byte-for-byte the legacy shape.
168        if let Some(range_id) = self.range_id {
169            object.insert("range_id".to_string(), JsonValue::Number(range_id.into()));
170        }
171        if let Some(epoch) = self.ownership_epoch {
172            object.insert(
173                "ownership_epoch".to_string(),
174                JsonValue::Number(epoch.into()),
175            );
176        }
177        JsonValue::Object(object)
178    }
179
180    pub fn encode(&self) -> Vec<u8> {
181        serde_json::to_string(&self.to_json_value())
182            .unwrap_or_else(|_| "{}".to_string())
183            .into_bytes()
184    }
185
186    pub fn with_term(mut self, term: u64) -> Self {
187        self.term = term;
188        self
189    }
190
191    pub fn decode(bytes: &[u8]) -> Result<Self, String> {
192        let text = std::str::from_utf8(bytes).map_err(|err| err.to_string())?;
193        let value = serde_json::from_str::<JsonValue>(text).map_err(|err| err.to_string())?;
194        let operation = value
195            .get("operation")
196            .and_then(JsonValue::as_str)
197            .and_then(ChangeOperation::from_wire_str)
198            .ok_or_else(|| "invalid replication operation".to_string())?;
199        let entity_bytes = value
200            .get("entity_bytes_hex")
201            .and_then(JsonValue::as_str)
202            .map(|value| hex_decode_string("entity_bytes_hex", value))
203            .transpose()?;
204
205        Ok(Self {
206            term: value
207                .get("term")
208                .and_then(JsonValue::as_u64)
209                .unwrap_or(DEFAULT_REPLICATION_TERM),
210            lsn: value.get("lsn").and_then(JsonValue::as_u64).unwrap_or(0),
211            timestamp: value
212                .get("timestamp")
213                .and_then(JsonValue::as_u64)
214                .unwrap_or(0),
215            operation,
216            collection: value
217                .get("collection")
218                .and_then(JsonValue::as_str)
219                .unwrap_or_default()
220                .to_string(),
221            entity_id: value
222                .get("rid")
223                .or_else(|| value.get("entity_id"))
224                .and_then(JsonValue::as_u64)
225                .unwrap_or(0),
226            entity_kind: value
227                .get("kind")
228                .or_else(|| value.get("entity_kind"))
229                .and_then(JsonValue::as_str)
230                .unwrap_or("entity")
231                .to_string(),
232            entity_bytes,
233            metadata: value.get("metadata").cloned(),
234            refresh_records: match value.get("refresh_records_hex") {
235                Some(JsonValue::Array(items)) => {
236                    let mut out = Vec::with_capacity(items.len());
237                    for item in items {
238                        let hex_str = item
239                            .as_str()
240                            .ok_or_else(|| "refresh_records_hex entry not a string".to_string())?;
241                        out.push(hex_decode_string("refresh_records_hex", hex_str)?);
242                    }
243                    Some(out)
244                }
245                None | Some(JsonValue::Null) => None,
246                _ => return Err("refresh_records_hex is not an array".to_string()),
247            },
248            // Issue #991 — absent on legacy records (decodes to `None`), so
249            // old payloads keep round-tripping unchanged.
250            range_id: value.get("range_id").and_then(JsonValue::as_u64),
251            ownership_epoch: value.get("ownership_epoch").and_then(JsonValue::as_u64),
252        })
253    }
254}
255
256/// Issue #991 — the range-authority watermark a replica or recovery target
257/// holds for a single range: the minimum term and ownership epoch it will
258/// accept for that range's user-data changes.
259///
260/// The physical WAL stays the source of truth; this fence operates purely on
261/// the derived metadata each [`ChangeRecord`] carries. A record stamped for
262/// this range whose term or ownership epoch is behind the watermark is a write
263/// from a stale (deposed) owner or a superseded timeline and must be rejected
264/// before it is applied. Records for a *different* range, or records with no
265/// range identity at all (legacy / non-range-replicated), are not this fence's
266/// concern and pass through untouched.
267#[derive(Debug, Clone, Copy, PartialEq, Eq)]
268pub struct RangeAuthority {
269    pub range_id: u64,
270    pub min_term: u64,
271    pub min_ownership_epoch: u64,
272}
273
274/// Why a [`RangeAuthority`] rejected a record (issue #991).
275#[derive(Debug, Clone, Copy, PartialEq, Eq)]
276pub enum RangeAdmitError {
277    /// The record's term is behind the term this range has accepted — a
278    /// returning ex-owner streaming on a superseded timeline.
279    StaleTerm {
280        record_term: u64,
281        accepted_term: u64,
282    },
283    /// The record's ownership epoch is behind the range's accepted epoch — a
284    /// write produced by an owner that has since lost write authority.
285    StaleOwnershipEpoch {
286        record_epoch: u64,
287        accepted_epoch: u64,
288    },
289}
290
291impl RangeAuthority {
292    /// Decide whether `record` may be applied to this range. Only records
293    /// explicitly stamped for `self.range_id` are gated; everything else is
294    /// admitted. Term is checked before epoch so a stale timeline is reported
295    /// as such even when its epoch also lags.
296    pub fn admit(&self, record: &ChangeRecord) -> Result<(), RangeAdmitError> {
297        if record.range_id != Some(self.range_id) {
298            return Ok(());
299        }
300        if record.term < self.min_term {
301            return Err(RangeAdmitError::StaleTerm {
302                record_term: record.term,
303                accepted_term: self.min_term,
304            });
305        }
306        // A record stamped for this range but missing an epoch predates epoch
307        // fencing; admit it on term alone rather than fail closed on absence.
308        if let Some(epoch) = record.ownership_epoch {
309            if epoch < self.min_ownership_epoch {
310                return Err(RangeAdmitError::StaleOwnershipEpoch {
311                    record_epoch: epoch,
312                    accepted_epoch: self.min_ownership_epoch,
313                });
314            }
315        }
316        Ok(())
317    }
318}
319
320pub fn public_item_kind(entity_kind: &str) -> &'static str {
321    match entity_kind {
322        "table" | "entity" | "row" => "row",
323        "graph_node" | "node" => "node",
324        "graph_edge" | "edge" => "edge",
325        "kv" => "kv",
326        "document" => "document",
327        "vector" => "vector",
328        other if other.contains("kv") => "kv",
329        other if other.contains("document") => "document",
330        other if other.contains("vector") => "vector",
331        _ => "item",
332    }
333}
334
335fn hex_decode_string(field: &'static str, value: &str) -> Result<Vec<u8>, String> {
336    hex_decode(field, value).map_err(|err| err.to_string())
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342
343    #[test]
344    fn change_record_round_trips_json_wire_payload() {
345        let record = ChangeRecord {
346            term: 3,
347            lsn: 7,
348            timestamp: 1234,
349            operation: ChangeOperation::Update,
350            collection: "users".to_string(),
351            entity_id: 42,
352            entity_kind: "row".to_string(),
353            entity_bytes: Some(vec![1, 2, 3]),
354            metadata: Some(serde_json::json!({"role": "admin"})),
355            refresh_records: None,
356            range_id: None,
357            ownership_epoch: None,
358        };
359
360        let decoded = ChangeRecord::decode(&record.encode()).expect("decode");
361
362        assert_eq!(decoded.term, record.term);
363        assert_eq!(decoded.lsn, record.lsn);
364        assert_eq!(decoded.collection, record.collection);
365        assert_eq!(decoded.entity_id, record.entity_id);
366        assert_eq!(decoded.entity_bytes, record.entity_bytes);
367        assert_eq!(decoded.metadata, record.metadata);
368    }
369
370    #[test]
371    fn range_authority_round_trips_on_the_json_wire() {
372        let record = ChangeRecord {
373            term: 5,
374            lsn: 12,
375            timestamp: 999,
376            operation: ChangeOperation::Insert,
377            collection: "orders".to_string(),
378            entity_id: 8,
379            entity_kind: "row".to_string(),
380            entity_bytes: Some(vec![9, 9]),
381            metadata: None,
382            refresh_records: None,
383            range_id: None,
384            ownership_epoch: None,
385        }
386        .with_range_authority(7, 3);
387
388        let decoded = ChangeRecord::decode(&record.encode()).expect("decode");
389
390        assert_eq!(decoded.range_id, Some(7));
391        assert_eq!(decoded.ownership_epoch, Some(3));
392        assert_eq!(decoded.term, 5);
393    }
394
395    #[test]
396    fn legacy_record_without_range_authority_decodes_to_none() {
397        // A payload from before #991 carries neither field; decoding must
398        // leave both absent rather than fabricate a default that would later
399        // collide with a real range fence.
400        let legacy =
401            br#"{"term":2,"lsn":4,"timestamp":1,"operation":"insert","collection":"users","rid":1,"kind":"row"}"#;
402
403        let decoded = ChangeRecord::decode(legacy).expect("decode legacy");
404
405        assert_eq!(decoded.range_id, None);
406        assert_eq!(decoded.ownership_epoch, None);
407    }
408
409    #[test]
410    fn unstamped_record_omits_range_keys_from_the_wire() {
411        let record = ChangeRecord::for_refresh(1, 1, "mv", Vec::new());
412        let text = String::from_utf8(record.encode()).expect("utf8");
413        assert!(!text.contains("range_id"), "got {text}");
414        assert!(!text.contains("ownership_epoch"), "got {text}");
415    }
416
417    fn stamped(term: u64, range_id: u64, epoch: u64) -> ChangeRecord {
418        ChangeRecord {
419            term,
420            lsn: 1,
421            timestamp: 1,
422            operation: ChangeOperation::Insert,
423            collection: "c".to_string(),
424            entity_id: 1,
425            entity_kind: "row".to_string(),
426            entity_bytes: Some(vec![1]),
427            metadata: None,
428            refresh_records: None,
429            range_id: None,
430            ownership_epoch: None,
431        }
432        .with_range_authority(range_id, epoch)
433    }
434
435    #[test]
436    fn range_authority_admits_current_term_and_epoch() {
437        let fence = RangeAuthority {
438            range_id: 7,
439            min_term: 3,
440            min_ownership_epoch: 4,
441        };
442        assert_eq!(fence.admit(&stamped(3, 7, 4)), Ok(()));
443        assert_eq!(fence.admit(&stamped(9, 7, 9)), Ok(()));
444    }
445
446    #[test]
447    fn range_authority_rejects_stale_epoch_and_term() {
448        let fence = RangeAuthority {
449            range_id: 7,
450            min_term: 3,
451            min_ownership_epoch: 4,
452        };
453        assert_eq!(
454            fence.admit(&stamped(3, 7, 2)),
455            Err(RangeAdmitError::StaleOwnershipEpoch {
456                record_epoch: 2,
457                accepted_epoch: 4,
458            })
459        );
460        // Term is checked first, so a doubly-stale record reports the term.
461        assert_eq!(
462            fence.admit(&stamped(1, 7, 1)),
463            Err(RangeAdmitError::StaleTerm {
464                record_term: 1,
465                accepted_term: 3,
466            })
467        );
468    }
469
470    #[test]
471    fn range_authority_ignores_other_ranges_and_unstamped_records() {
472        let fence = RangeAuthority {
473            range_id: 7,
474            min_term: 5,
475            min_ownership_epoch: 5,
476        };
477        // Stale, but for a different range — not this fence's business.
478        assert_eq!(fence.admit(&stamped(1, 99, 1)), Ok(()));
479        // No range identity at all — legacy record passes through.
480        let mut legacy = stamped(1, 7, 1);
481        legacy.range_id = None;
482        legacy.ownership_epoch = None;
483        assert_eq!(fence.admit(&legacy), Ok(()));
484    }
485
486    #[test]
487    fn refresh_records_round_trip_without_reordering() {
488        let records = vec![vec![0x10, 0x20, 0x30], vec![0xAA, 0xBB], Vec::new()];
489        let record =
490            ChangeRecord::for_refresh(11, 99, "mv_orders_summary", records.clone()).with_term(4);
491
492        let decoded = ChangeRecord::decode(&record.encode()).expect("decode");
493
494        assert_eq!(decoded.term, 4);
495        assert_eq!(decoded.operation, ChangeOperation::Refresh);
496        assert_eq!(decoded.collection, "mv_orders_summary");
497        assert_eq!(decoded.refresh_records.as_deref(), Some(&records[..]));
498    }
499
500    #[test]
501    fn legacy_change_record_defaults_term() {
502        let legacy =
503            br#"{"lsn":9,"timestamp":1,"operation":"delete","collection":"users","rid":5,"kind":"row"}"#;
504
505        let decoded = ChangeRecord::decode(legacy).expect("decode legacy record");
506
507        assert_eq!(decoded.term, DEFAULT_REPLICATION_TERM);
508        assert_eq!(decoded.lsn, 9);
509    }
510}
reddb_wire/replication/change_record.rs

reddb_wire/replication/
change_record.rs
