weavatrix_memory/snapshot/compact/
mod.rs

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

weavatrix_memory/snapshot/compact/
mod.rs
