heddle_object_model/op_record/
codec.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Versioned codec for current OpRecord payloads stored in the packed oplog.
3//!
4//! Repository format v3 changed state identity from 16-byte logical ChangeIds
5//! to 32-byte content-addressed StateIds. That boundary is intentionally not
6//! migratable: repository open refuses older formats, and the oplog codec
7//! refuses record schema versions 1 through 3 without rewriting their bytes.
8//! Future incompatible record changes must allocate a new schema version.
9
10use serde::Deserialize;
11
12use super::OpRecord;
13use crate::error::{HeddleError, Result};
14
15pub const CURRENT_OP_RECORD_SCHEMA_VERSION: u32 = 4;
16const CURRENT_OP_RECORD_SCHEMA_NAME: &str = "state-id-v4";
17const OP_RECORD_STORAGE: &str = "oplog record schema";
18
19pub fn validate_op_record_schema_version(version: u32) -> Result<()> {
20    if version < CURRENT_OP_RECORD_SCHEMA_VERSION {
21        return Err(HeddleError::StorageFormatTooOld {
22            storage: OP_RECORD_STORAGE.to_string(),
23            found: version,
24            required: CURRENT_OP_RECORD_SCHEMA_VERSION,
25        });
26    }
27    if version > CURRENT_OP_RECORD_SCHEMA_VERSION {
28        return Err(HeddleError::StorageFormatTooNew {
29            storage: OP_RECORD_STORAGE.to_string(),
30            found: version,
31            supported: CURRENT_OP_RECORD_SCHEMA_VERSION,
32        });
33    }
34    Ok(())
35}
36
37pub fn decode_current_record(bytes: &[u8]) -> Result<OpRecord> {
38    decode_rmp(bytes, CURRENT_OP_RECORD_SCHEMA_NAME)
39}
40
41pub fn encode_current_record(record: &OpRecord) -> Result<Vec<u8>> {
42    rmp_serde::to_vec(record).map_err(|e| HeddleError::Serialization(e.to_string()))
43}
44
45fn decode_rmp<T>(bytes: &[u8], schema_name: &str) -> Result<T>
46where
47    T: for<'de> Deserialize<'de>,
48{
49    rmp_serde::from_slice(bytes).map_err(|e| {
50        HeddleError::Serialization(format!(
51            "failed to decode OpRecord payload as {schema_name}: {e}"
52        ))
53    })
54}
55
56#[cfg(test)]
57mod tests {
58    use super::super::{ConflictResolutionMode, RecordedHead, ThreadUpdateSnapshots};
59    use super::*;
60    use crate::object::{Agent, Attribution, ContentHash, Principal, StateId, VisibilityTier};
61
62    fn state(byte: u8) -> StateId {
63        StateId::from_bytes([byte; 32])
64    }
65
66    fn hash(byte: u8) -> ContentHash {
67        ContentHash::from_bytes([byte; 32])
68    }
69
70    fn assert_round_trip(record: OpRecord) {
71        let bytes = encode_current_record(&record).unwrap();
72        let decoded = decode_current_record(&bytes).unwrap();
73        assert_eq!(format!("{decoded:?}"), format!("{record:?}"));
74    }
75
76    fn canonical_current_records() -> Vec<OpRecord> {
77        vec![
78            OpRecord::Snapshot {
79                new_state: state(1),
80                prev_head: Some(state(2)),
81                head: None,
82                thread: Some("main".into()),
83            },
84            OpRecord::Goto {
85                target: state(3),
86                prev_head: Some(state(2)),
87                head: state(3),
88            },
89            OpRecord::ThreadCreate {
90                name: "topic".into(),
91                state: state(4),
92                manager_snapshot: Some(vec![1, 2, 3]),
93            },
94            OpRecord::ThreadDelete {
95                name: "old".into(),
96                state: state(5),
97            },
98            OpRecord::ThreadUpdate {
99                name: "main".into(),
100                old_state: state(6),
101                new_state: state(7),
102                manager_snapshots: ThreadUpdateSnapshots::from_record_sets(
103                    Some(vec![6]),
104                    Some(vec![7]),
105                    vec![vec![60], vec![61]],
106                    vec![vec![70]],
107                    true,
108                ),
109            },
110            OpRecord::Fork {
111                from: state(8),
112                new_state: state(9),
113                thread: Some("topic".into()),
114                head: None,
115            },
116            OpRecord::Collapse {
117                sources: vec![state(8), state(9)],
118                result: state(10),
119                thread: Some("main".into()),
120                pre_thread_state: Some(state(7)),
121            },
122            OpRecord::MarkerCreate {
123                name: "release".into(),
124                state: state(11),
125            },
126            OpRecord::MarkerDelete {
127                name: "draft".into(),
128                state: state(12),
129            },
130            OpRecord::Checkpoint {
131                parent: Some(state(12)),
132                state: state(13),
133                thread: Some("main".into()),
134            },
135            OpRecord::TransactionAbort {
136                transaction_id: "abort".into(),
137                reason: "reason".into(),
138            },
139            OpRecord::EphemeralThreadCollapse {
140                thread: "ephemeral".into(),
141                final_state: state(14),
142            },
143            OpRecord::ConflictResolved {
144                conflict_id: "conflict".into(),
145                resolution: "ours".into(),
146                resolver: Attribution::with_agent(
147                    Principal::new("Resolver", "resolver@example.com"),
148                    Agent::new("openai", "gpt-5-codex"),
149                ),
150                mode: ConflictResolutionMode::Ours,
151            },
152            OpRecord::TransactionCommit {
153                transaction_id: "tx".into(),
154                op_count: 2,
155            },
156            OpRecord::Redact {
157                redaction_id: hash(1),
158                blob: hash(2),
159                state: state(15),
160                path: "secret.txt".into(),
161            },
162            OpRecord::Purge {
163                redaction_id: hash(3),
164                blob: hash(4),
165            },
166            OpRecord::FastForward {
167                source_thread: "feature".into(),
168                target_thread: "main".into(),
169                pre_target_id: state(17),
170                post_target_id: state(18),
171            },
172            OpRecord::GitCheckpoint {
173                branch: "main".into(),
174                state: state(20),
175                previous_git_oid: Some("abc".into()),
176                new_git_oid: "def".into(),
177            },
178            OpRecord::RemoteThreadUpdate {
179                remote: "origin".into(),
180                thread: "main".into(),
181                state: state(21),
182            },
183            OpRecord::RemoteThreadDelete {
184                remote: "origin".into(),
185                thread: "old".into(),
186                state: state(22),
187            },
188            OpRecord::UndoRecoveryUpdate { state: state(23) },
189            OpRecord::StateVisibilitySet {
190                state: state(24),
191                record_id: hash(5),
192                tier: VisibilityTier::Internal,
193                prior_sidecar: None,
194                new_sidecar: Some(vec![1, 2, 3]),
195            },
196            OpRecord::StateVisibilityPromote {
197                state: state(25),
198                superseded: hash(6),
199                record_id: hash(7),
200                tier: VisibilityTier::Restricted {
201                    scope_label: "embargo".into(),
202                },
203                prior_sidecar: Some(vec![4]),
204                new_sidecar: Some(vec![5]),
205            },
206            OpRecord::HeadUpdate {
207                previous: RecordedHead::Detached { state: state(26) },
208                new: RecordedHead::Attached {
209                    thread: "main".into(),
210                },
211            },
212            OpRecord::EntryVisibilitySet {
213                change_id: crate::object::ChangeId::from_bytes([7u8; 16]),
214                record_id: hash(8),
215                prior_sidecar: None,
216                new_sidecar: Some(vec![9, 9, 9]),
217            },
218        ]
219    }
220
221    fn variant_name(record: &OpRecord) -> &'static str {
222        match record {
223            OpRecord::Snapshot { .. } => "Snapshot",
224            OpRecord::Goto { .. } => "Goto",
225            OpRecord::ThreadCreate { .. } => "ThreadCreate",
226            OpRecord::ThreadDelete { .. } => "ThreadDelete",
227            OpRecord::ThreadUpdate { .. } => "ThreadUpdate",
228            OpRecord::Fork { .. } => "Fork",
229            OpRecord::Collapse { .. } => "Collapse",
230            OpRecord::MarkerCreate { .. } => "MarkerCreate",
231            OpRecord::MarkerDelete { .. } => "MarkerDelete",
232            OpRecord::Checkpoint { .. } => "Checkpoint",
233            OpRecord::TransactionAbort { .. } => "TransactionAbort",
234            OpRecord::EphemeralThreadCollapse { .. } => "EphemeralThreadCollapse",
235            OpRecord::ConflictResolved { .. } => "ConflictResolved",
236            OpRecord::TransactionCommit { .. } => "TransactionCommit",
237            OpRecord::Redact { .. } => "Redact",
238            OpRecord::Purge { .. } => "Purge",
239            OpRecord::FastForward { .. } => "FastForward",
240            OpRecord::GitCheckpoint { .. } => "GitCheckpoint",
241            OpRecord::RemoteThreadUpdate { .. } => "RemoteThreadUpdate",
242            OpRecord::RemoteThreadDelete { .. } => "RemoteThreadDelete",
243            OpRecord::UndoRecoveryUpdate { .. } => "UndoRecoveryUpdate",
244            OpRecord::StateVisibilitySet { .. } => "StateVisibilitySet",
245            OpRecord::StateVisibilityPromote { .. } => "StateVisibilityPromote",
246            OpRecord::HeadUpdate { .. } => "HeadUpdate",
247            OpRecord::EntryVisibilitySet { .. } => "EntryVisibilitySet",
248        }
249    }
250
251    #[test]
252    fn schema_four_is_current_and_legacy_versions_are_refused() {
253        assert_eq!(CURRENT_OP_RECORD_SCHEMA_VERSION, 4);
254        validate_op_record_schema_version(4).unwrap();
255        for legacy in 1..=3 {
256            let error = validate_op_record_schema_version(legacy).unwrap_err();
257            assert!(matches!(
258                error,
259                HeddleError::StorageFormatTooOld {
260                    found,
261                    required: 4,
262                    ..
263                } if found == legacy
264            ));
265        }
266        assert!(matches!(
267            validate_op_record_schema_version(5).unwrap_err(),
268            HeddleError::StorageFormatTooNew {
269                found: 5,
270                supported: 4,
271                ..
272            }
273        ));
274    }
275
276    #[test]
277    fn every_current_variant_round_trips() {
278        let records = canonical_current_records();
279        assert_eq!(
280            records.iter().map(variant_name).collect::<Vec<_>>(),
281            [
282                "Snapshot",
283                "Goto",
284                "ThreadCreate",
285                "ThreadDelete",
286                "ThreadUpdate",
287                "Fork",
288                "Collapse",
289                "MarkerCreate",
290                "MarkerDelete",
291                "Checkpoint",
292                "TransactionAbort",
293                "EphemeralThreadCollapse",
294                "ConflictResolved",
295                "TransactionCommit",
296                "Redact",
297                "Purge",
298                "FastForward",
299                "GitCheckpoint",
300                "RemoteThreadUpdate",
301                "RemoteThreadDelete",
302                "UndoRecoveryUpdate",
303                "StateVisibilitySet",
304                "StateVisibilityPromote",
305                "HeadUpdate",
306                "EntryVisibilitySet",
307            ]
308        );
309        for record in records {
310            assert_round_trip(record);
311        }
312    }
313
314    #[test]
315    fn state_id_v4_visibility_tail_bytes_are_frozen() {
316        let record = OpRecord::StateVisibilityPromote {
317            state: state(1),
318            superseded: hash(2),
319            record_id: hash(3),
320            tier: VisibilityTier::Internal,
321            prior_sidecar: Some(vec![4]),
322            new_sidecar: Some(vec![5]),
323        };
324
325        let expected = [
326            &[
327                129, 182, 83, 116, 97, 116, 101, 86, 105, 115, 105, 98, 105, 108, 105, 116, 121,
328                80, 114, 111, 109, 111, 116, 101, 150, 220, 0, 32,
329            ][..],
330            &[1; 32],
331            &[220, 0, 32],
332            &[2; 32],
333            &[220, 0, 32],
334            &[3; 32],
335            &[168, 73, 110, 116, 101, 114, 110, 97, 108, 145, 4, 145, 5],
336        ]
337        .concat();
338
339        assert_eq!(encode_current_record(&record).unwrap(), expected);
340    }
341
342    #[test]
343    fn historical_sixteen_byte_payload_is_not_a_state_id_record() {
344        let historical = [
345            129, 168, 67, 111, 108, 108, 97, 112, 115, 101, 147, 146, 220, 0, 16, 10, 10, 10, 10,
346            10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 220, 0, 16, 11, 11, 11, 11, 11, 11, 11,
347            11, 11, 11, 11, 11, 11, 11, 11, 11, 220, 0, 16, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12,
348            12, 12, 12, 12, 12, 12, 164, 109, 97, 105, 110,
349        ];
350        let error = decode_current_record(&historical)
351            .expect_err("16-byte ChangeIds must not decode as StateIds");
352        assert!(error.to_string().contains("expected an array of length 32"));
353    }
354}
heddle_object_model/op_record/codec.rs

heddle_object_model/op_record/
codec.rs
