weavatrix_memory/snapshot/compact/
mod.rs

1mod domain;
2mod io;
3
4use crate::{
5    codec::Codec,
6    error::Result,
7    id::{FactId, StreamId},
8    projection::{
9        ProjectionSnapshot, ReplayCursor,
10        memory::{
11            index::IdIndex,
12            state::{MemoryProjection, NodeHistory, NodeRevision, Retraction, Supersession},
13        },
14    },
15    time::Timestamp,
16};
17use domain::{read_fact, read_node, write_fact, write_node};
18use io::{Reader, Writer, codec};
19use std::collections::{BTreeMap, HashMap};
20
21const HEADER: &[u8; 8] = b"WMEMCB01";
22
23#[derive(Debug, Clone, Copy, Default)]
24pub struct CompactSnapshotCodec;
25
26impl Codec<ProjectionSnapshot<MemoryProjection>> for CompactSnapshotCodec {
27    fn encode(&self, value: &ProjectionSnapshot<MemoryProjection>) -> Result<Vec<u8>> {
28        let mut writer = Writer::new(HEADER);
29        write_cursor(&mut writer, &value.cursor)?;
30        write_projection(&mut writer, &value.projection)?;
31        Ok(writer.finish())
32    }
33
34    fn decode(&self, bytes: &[u8]) -> Result<ProjectionSnapshot<MemoryProjection>> {
35        let mut reader = Reader::new(bytes, HEADER)?;
36        let cursor = read_cursor(&mut reader)?;
37        let projection = read_projection(&mut reader)?;
38        reader.finish()?;
39        if cursor.global_position != projection.last_global_position {
40            return Err(codec("cursor and projection positions disagree"));
41        }
42        Ok(ProjectionSnapshot { cursor, projection })
43    }
44}
45
46fn write_cursor(writer: &mut Writer, cursor: &ReplayCursor) -> Result<()> {
47    writer.optional_u64(cursor.global_position);
48    writer.usize(cursor.stream_versions.len())?;
49    for (stream, version) in &cursor.stream_versions {
50        writer.string(stream.as_str())?;
51        writer.varint(*version);
52    }
53    Ok(())
54}
55
56fn read_cursor(reader: &mut Reader<'_>) -> Result<ReplayCursor> {
57    let global_position = reader.optional_u64()?;
58    let count = reader.count()?;
59    let mut stream_versions = BTreeMap::new();
60    for _ in 0..count {
61        let stream = StreamId::new(reader.string()?)?;
62        let version = reader.varint()?;
63        if stream_versions.insert(stream, version).is_some() {
64            return Err(codec("duplicate stream in replay cursor"));
65        }
66    }
67    Ok(ReplayCursor {
68        global_position,
69        stream_versions,
70    })
71}
72
73fn write_projection(writer: &mut Writer, projection: &MemoryProjection) -> Result<()> {
74    writer.usize(projection.nodes.len())?;
75    for history in &projection.nodes {
76        writer.usize(history.later.len() + 1)?;
77        write_revision(writer, &history.first)?;
78        for revision in &history.later {
79            write_revision(writer, revision)?;
80        }
81    }
82    writer.usize(projection.facts.len())?;
83    for fact in &projection.facts {
84        write_fact(writer, fact)?;
85    }
86    writer.usize(projection.supersessions.len())?;
87    for (prior, change) in &projection.supersessions {
88        writer.string(prior.as_str())?;
89        writer.string(change.replacement.as_str())?;
90        writer.signed(change.valid_from.as_unix_micros());
91        writer.signed(change.recorded_at.as_unix_micros());
92    }
93    writer.usize(projection.retractions.len())?;
94    for (fact, change) in &projection.retractions {
95        writer.string(fact.as_str())?;
96        writer.signed(change.valid_until.as_unix_micros());
97        writer.signed(change.recorded_at.as_unix_micros());
98    }
99    writer.optional_u64(projection.last_global_position);
100    Ok(())
101}
102
103fn read_projection(reader: &mut Reader<'_>) -> Result<MemoryProjection> {
104    let node_count = reader.count()?;
105    let mut nodes = Vec::with_capacity(node_count);
106    for _ in 0..node_count {
107        nodes.push(read_history(reader)?);
108    }
109    let fact_count = reader.count()?;
110    let mut facts = Vec::with_capacity(fact_count);
111    for _ in 0..fact_count {
112        facts.push(read_fact(reader)?);
113    }
114    let supersessions = read_supersessions(reader)?;
115    let retractions = read_retractions(reader)?;
116    let last_global_position = reader.optional_u64()?;
117    let mut projection = MemoryProjection {
118        nodes,
119        facts,
120        node_lookup: IdIndex::with_capacity(node_count),
121        fact_lookup: IdIndex::with_capacity(fact_count),
122        incident_offsets: Vec::new(),
123        incident_facts: Vec::new(),
124        incident_delta: HashMap::new(),
125        supersessions,
126        retractions,
127        last_global_position,
128    };
129    projection.rebuild_indexes()?;
130    Ok(projection)
131}
132
133fn write_revision(writer: &mut Writer, revision: &NodeRevision) -> Result<()> {
134    write_node(writer, &revision.node)?;
135    writer.signed(revision.recorded_at.as_unix_micros());
136    writer.varint(revision.position);
137    Ok(())
138}
139
140fn read_history(reader: &mut Reader<'_>) -> Result<NodeHistory> {
141    let count = reader.count()?;
142    if count == 0 {
143        return Err(codec("node revision history must not be empty"));
144    }
145    let first = read_revision(reader)?;
146    let mut later = Vec::with_capacity(count - 1);
147    for _ in 1..count {
148        later.push(read_revision(reader)?);
149    }
150    Ok(NodeHistory { first, later })
151}
152
153fn read_revision(reader: &mut Reader<'_>) -> Result<NodeRevision> {
154    Ok(NodeRevision {
155        node: read_node(reader)?,
156        recorded_at: Timestamp::from_unix_micros(reader.signed()?),
157        position: reader.varint()?,
158    })
159}
160
161fn read_supersessions(reader: &mut Reader<'_>) -> Result<BTreeMap<FactId, Supersession>> {
162    let count = reader.count()?;
163    let mut changes = BTreeMap::new();
164    for _ in 0..count {
165        let prior = FactId::new(reader.string()?)?;
166        let change = Supersession {
167            replacement: FactId::new(reader.string()?)?,
168            valid_from: Timestamp::from_unix_micros(reader.signed()?),
169            recorded_at: Timestamp::from_unix_micros(reader.signed()?),
170        };
171        if changes.insert(prior, change).is_some() {
172            return Err(codec("duplicate supersession"));
173        }
174    }
175    Ok(changes)
176}
177
178fn read_retractions(reader: &mut Reader<'_>) -> Result<BTreeMap<FactId, Retraction>> {
179    let count = reader.count()?;
180    let mut changes = BTreeMap::new();
181    for _ in 0..count {
182        let fact = FactId::new(reader.string()?)?;
183        let change = Retraction {
184            valid_until: Timestamp::from_unix_micros(reader.signed()?),
185            recorded_at: Timestamp::from_unix_micros(reader.signed()?),
186        };
187        if changes.insert(fact, change).is_some() {
188            return Err(codec("duplicate retraction"));
189        }
190    }
191    Ok(changes)
192}
weavatrix_memory/snapshot/compact/mod.rs

weavatrix_memory/snapshot/compact/
mod.rs
