linkrs_core/wal/
sync.rs

1use serde::{Deserialize, Serialize};
2
3use crate::types::{
4    IdempotencyKey, IndexGeneration, LabelId, OrderingKey, TargetId, TransactionId, VertexId,
5};
6
7pub const WAL_SYNC_WIRE_VERSION: u16 = 1;
8
9#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
10pub enum EntityRef {
11    Vertex(VertexId),
12    Edge {
13        src: VertexId,
14        dst: VertexId,
15        edge_type: LabelId,
16        ranking: i64,
17    },
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
21pub enum IndexOperation {
22    Upsert,
23    Delete,
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
27pub enum LifecycleMutation {
28    Create,
29    BeginBackfill,
30    BeginCatchUp,
31    Activate,
32    BeginDrain,
33    Drop,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37pub struct IndexMutation {
38    pub wire_version: u16,
39    pub target: TargetId,
40    pub index_id: u64,
41    pub index_generation: IndexGeneration,
42    pub entity_ref: EntityRef,
43    pub operation: IndexOperation,
44    pub document_or_vector: Vec<u8>,
45    pub idempotency_key: IdempotencyKey,
46    pub ordering_key: OrderingKey,
47}
48
49impl IndexMutation {
50    pub fn validate(&self) -> Result<(), String> {
51        validate_wire_version(self.wire_version)
52    }
53}
54
55#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
56pub struct OutboxIntent {
57    pub wire_version: u16,
58    pub transaction_id: TransactionId,
59    pub intent_sequence: u32,
60    pub mutation: IndexMutation,
61}
62
63impl OutboxIntent {
64    pub fn validate(&self) -> Result<(), String> {
65        validate_wire_version(self.wire_version)?;
66        self.mutation.validate()
67    }
68}
69
70#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
71pub struct TransactionCommit {
72    pub wire_version: u16,
73    pub transaction_id: TransactionId,
74    pub intent_count: u32,
75    pub batch_checksum: u32,
76    #[serde(default)]
77    pub first_sequence: u64,
78    #[serde(default)]
79    pub entry_count: u32,
80}
81
82impl TransactionCommit {
83    pub fn validate(&self) -> Result<(), String> {
84        validate_wire_version(self.wire_version)
85    }
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
89pub struct TransactionAbort {
90    pub wire_version: u16,
91    pub transaction_id: TransactionId,
92}
93
94impl TransactionAbort {
95    pub fn validate(&self) -> Result<(), String> {
96        validate_wire_version(self.wire_version)
97    }
98}
99
100fn validate_wire_version(version: u16) -> Result<(), String> {
101    if version == WAL_SYNC_WIRE_VERSION {
102        Ok(())
103    } else {
104        Err(format!(
105            "Unsupported WAL sync wire version: expected {}, got {}",
106            WAL_SYNC_WIRE_VERSION, version
107        ))
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use postcard::{from_bytes, to_allocvec};
114
115    use super::{EntityRef, IndexMutation, IndexOperation, OutboxIntent, WAL_SYNC_WIRE_VERSION};
116    use crate::types::{
117        IdempotencyKey, IndexGeneration, OrderingKey, TargetId, TransactionId, VertexId,
118    };
119
120    fn mutation() -> IndexMutation {
121        IndexMutation {
122            wire_version: WAL_SYNC_WIRE_VERSION,
123            target: TargetId::new("fulltext").expect("target should be valid"),
124            index_id: 7,
125            index_generation: IndexGeneration::new(3),
126            entity_ref: EntityRef::Vertex(VertexId::try_from_int64(42).expect("test vertex id")),
127            operation: IndexOperation::Upsert,
128            document_or_vector: b"document".to_vec(),
129            idempotency_key: IdempotencyKey::new("txn-1:0")
130                .expect("idempotency key should be valid"),
131            ordering_key: OrderingKey::new("index-7:vertex-42")
132                .expect("ordering key should be valid"),
133        }
134    }
135
136    #[test]
137    fn intent_roundtrips_without_transport_types() {
138        let intent = OutboxIntent {
139            wire_version: WAL_SYNC_WIRE_VERSION,
140            transaction_id: TransactionId::new(1),
141            intent_sequence: 0,
142            mutation: mutation(),
143        };
144        let bytes = to_allocvec(&intent).expect("intent should serialize");
145        let decoded: OutboxIntent = from_bytes(&bytes).expect("intent should deserialize");
146        assert_eq!(decoded, intent);
147        assert!(decoded.validate().is_ok());
148    }
149
150    #[test]
151    fn unknown_wire_version_is_rejected() {
152        let mut mutation = mutation();
153        mutation.wire_version = WAL_SYNC_WIRE_VERSION + 1;
154        assert!(mutation.validate().is_err());
155    }
156}
linkrs_core/wal/sync.rs

linkrs_core/wal/
sync.rs
