OXYGENHUB
# afaba5f1306d oxygen / crates / oxy-core / src / state.rs
41.6 KiB · Rust · last changed Checkpoint payload version 2: large file graphs as shared blobs named from an index, read in parallel (loose and packed) and decoded in parallel; version 1 still read; gc, fsck and shallow-clone bundles carry the graph blobs (spec 11, Draft 0.8j) 1 month ago source blame history raw
1
//! A state: a dependency-closed set of changes, applied to a tree graph and
2
//! per-node file graphs (spec §8.3), with materialization (spec §9.5, §11).
3
4
use crate::cbor::{self, Value};
5
use crate::graph::{AtomKey, DepOracle, FileGraph, Materialized, Vertex};
6
use crate::hash::KeyId;
7
use crate::hash::{ChangeHash, ObjectRef};
8
use crate::ids::NodeId;
9
use crate::objects::{
10
    self as o, AtomRef, Change, ChangeKind, Edit, Hlc, Mode, NodeKind, ObjectError, TreeOp,
11
};
12
use crate::tree::{TreeGraph, TreeView};
13
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
14
use std::sync::{Arc, Mutex};
15
16
/// Checkpoint payload version 2 (spec §11): file graphs by reference.
17
pub const CHECKPOINT_INDEX_VERSION: u64 = 2;
18
19
/// A file graph encoded in fewer bytes stays inline in a version 2 index.
20
pub const INLINE_GRAPH_LIMIT: usize = 2048;
21
22
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
23
pub enum StateError {
24
    #[error("change {1} depends on {0}, which is not in the state")]
25
    MissingDep(ChangeHash, ChangeHash),
26
}
27
28
/// One `Replace` edit: `(change, old blob, new blob)`.
29
type ReplaceRecord = (ChangeHash, Option<ObjectRef>, Option<ObjectRef>);
30
31
/// What a state remembers about a change: everything but its operations,
32
/// which live in the graphs once applied (and in the store).
33
#[derive(Clone, Debug, PartialEq, Eq)]
34
pub struct ChangeMeta {
35
    pub change_id: crate::ids::ChangeId,
36
    pub deps: Vec<ChangeHash>,
37
    pub author: KeyId,
38
    pub hlc: Hlc,
39
    pub message: String,
40
    pub kind: ChangeKind,
41
}
42
43
impl ChangeMeta {
44
    pub fn of(c: &Change) -> ChangeMeta {
45
        ChangeMeta {
46
            change_id: c.change_id,
47
            deps: c.deps.clone(),
48
            author: c.author,
49
            hlc: c.hlc,
50
            message: c.message.clone(),
51
            kind: c.kind,
52
        }
53
    }
54
}
55
56
#[derive(Default)]
57
pub struct State {
58
    changes: HashMap<ChangeHash, ChangeMeta>,
59
    tree: TreeGraph,
60
    files: HashMap<NodeId, FileGraph>,
61
    replaces: HashMap<NodeId, Vec<ReplaceRecord>>,
62
    /// Memoised transitive dependencies. A mutex, not a `RefCell`, so a
63
    /// loaded state can be shared across threads by a host that caches it.
64
    closure: Mutex<HashMap<ChangeHash, Arc<HashSet<ChangeHash>>>>,
65
}
66
67
impl Clone for State {
68
    fn clone(&self) -> State {
69
        State {
70
            changes: self.changes.clone(),
71
            tree: self.tree.clone(),
72
            files: self.files.clone(),
73
            replaces: self.replaces.clone(),
74
            closure: Mutex::new(self.closure.lock().unwrap().clone()),
75
        }
76
    }
77
}
78
79
#[derive(Clone, Debug, PartialEq, Eq)]
80
pub enum FileContent {
81
    Text {
82
        bytes: Vec<u8>,
83
        materialized: Materialized,
84
    },
85
    /// Live blobs; more than one is a replace conflict (spec §9.3).
86
    Opaque {
87
        blobs: Vec<ObjectRef>,
88
    },
89
    Dir,
90
}
91
92
#[derive(Clone, Debug, PartialEq, Eq)]
93
pub struct FileState {
94
    pub node: NodeId,
95
    pub kind: NodeKind,
96
    pub mode: Mode,
97
    pub content: FileContent,
98
}
99
100
impl FileState {
101
    pub fn is_conflicted(&self) -> bool {
102
        match &self.content {
103
            FileContent::Text { materialized, .. } => materialized.is_conflicted(),
104
            FileContent::Opaque { blobs } => blobs.len() > 1,
105
            FileContent::Dir => false,
106
        }
107
    }
108
}
109
110
/// See `State::conflict_regions`.
111
#[derive(Clone, Debug, Default, PartialEq, Eq)]
112
pub struct ConflictRegions {
113
    pub plain: Vec<(AtomKey, Vec<u8>)>,
114
    pub region_atoms: Vec<AtomKey>,
115
    pub contributors: Vec<ChangeHash>,
116
}
117
118
/// The whole tree of a state, materialized.
119
#[derive(Clone, Debug, Default)]
120
pub struct Snapshot {
121
    pub tree: TreeView,
122
    pub files: BTreeMap<String, FileState>,
123
}
124
125
impl Snapshot {
126
    pub fn is_conflicted(&self) -> bool {
127
        self.tree.is_conflicted() || self.files.values().any(|f| f.is_conflicted())
128
    }
129
}
130
131
fn vertex(r: AtomRef) -> Vertex {
132
    match r {
133
        AtomRef::Start => Vertex::Start,
134
        AtomRef::End => Vertex::End,
135
        AtomRef::Atom { change, index } => Vertex::Atom((change, index)),
136
    }
137
}
138
139
/// Replace self references in a tree op with the concrete change hash.
140
pub fn resolve_tree_op(op: &TreeOp, me: ChangeHash) -> TreeOp {
141
    match op {
142
        TreeOp::RemoveNode { node, placement } => TreeOp::RemoveNode {
143
            node: *node,
144
            placement: placement.resolve(me),
145
        },
146
        TreeOp::Move {
147
            node,
148
            old,
149
            parent,
150
            name,
151
        } => TreeOp::Move {
152
            node: *node,
153
            old: old.resolve(me),
154
            parent: *parent,
155
            name: name.clone(),
156
        },
157
        TreeOp::SetMode { node, old, mode } => TreeOp::SetMode {
158
            node: *node,
159
            old: old.resolve(me),
160
            mode: *mode,
161
        },
162
        add @ TreeOp::AddNode { .. } => add.clone(),
163
    }
164
}
165
166
impl State {
167
    pub fn new() -> Self {
168
        Self::default()
169
    }
170
171
    pub fn len(&self) -> usize {
172
        self.changes.len()
173
    }
174
    pub fn is_empty(&self) -> bool {
175
        self.changes.is_empty()
176
    }
177
    pub fn contains(&self, h: &ChangeHash) -> bool {
178
        self.changes.contains_key(h)
179
    }
180
    pub fn get(&self, h: &ChangeHash) -> Option<&ChangeMeta> {
181
        self.changes.get(h)
182
    }
183
    pub fn changes(&self) -> impl Iterator<Item = (&ChangeHash, &ChangeMeta)> {
184
        self.changes.iter()
185
    }
186
    pub fn tree(&self) -> &TreeGraph {
187
        &self.tree
188
    }
189
    pub fn file(&self, node: &NodeId) -> Option<&FileGraph> {
190
        self.files.get(node)
191
    }
192
193
    /// Add a change whose dependencies are already present.
194
    pub fn add(&mut self, change: Change) -> Result<ChangeHash, StateError> {
195
        let h = change.hash();
196
        if self.changes.contains_key(&h) {
197
            return Ok(h);
198
        }
199
        for d in &change.deps {
200
            if !self.changes.contains_key(d) {
201
                return Err(StateError::MissingDep(*d, h));
202
            }
203
        }
204
        Ok(self.add_unchecked(change))
205
    }
206
207
    /// Add a change in any order; call `validate` once all are in.
208
    pub fn add_unchecked(&mut self, change: Change) -> ChangeHash {
209
        let h = change.hash();
210
        if self.changes.contains_key(&h) {
211
            return h;
212
        }
213
        let meta = ChangeMeta::of(&change);
214
        self.apply(h, change);
215
        self.changes.insert(h, meta);
216
        self.closure.lock().unwrap().clear();
217
        h
218
    }
219
220
    // ------------------------------------------------------------ checkpoints (spec §11)
221
222
    /// Serialize the whole state in payload version 1: change metadata,
223
    /// tree, file graphs, replaces, all in one value (spec §11).
224
    pub fn checkpoint_payload(&self) -> Vec<u8> {
225
        let mut nodes: Vec<&NodeId> = self.files.keys().collect();
226
        nodes.sort();
227
        let files = Value::Array(
228
            nodes
229
                .into_iter()
230
                .map(|n| Value::Array(vec![Value::Bytes(n.0.to_vec()), self.files[n].to_value()]))
231
                .collect(),
232
        );
233
        cbor::encode(&self.payload_value(o::FORMAT_VERSION, files))
234
    }
235
236
    /// Each file graph encoded on its own, by node, for a version 2 payload
237
    /// (spec §11): the graphs of large files become blobs of their own.
238
    pub fn checkpoint_file_graphs(&self) -> Vec<(NodeId, Vec<u8>)> {
239
        use rayon::prelude::*;
240
        let mut nodes: Vec<(&NodeId, &FileGraph)> = self.files.iter().collect();
241
        nodes.sort_by_key(|(n, _)| **n);
242
        nodes
243
            .into_par_iter()
244
            .map(|(n, g)| (*n, cbor::encode(&g.to_value())))
245
            .collect()
246
    }
247
248
    /// The payload version 2 index for the graphs of
249
    /// [`State::checkpoint_file_graphs`]: a graph encoded in fewer than
250
    /// [`INLINE_GRAPH_LIMIT`] bytes is kept inline, a larger one is named by
251
    /// `blob_of` (the ObjectRef of the blob holding its encoding).
252
    pub fn checkpoint_index(
253
        &self,
254
        graphs: &[(NodeId, Vec<u8>)],
255
        mut blob_of: impl FnMut(&[u8]) -> ObjectRef,
256
    ) -> Vec<u8> {
257
        let files = Value::Array(
258
            graphs
259
                .iter()
260
                .map(|(n, enc)| {
261
                    let entry = if enc.len() < INLINE_GRAPH_LIMIT {
262
                        cbor::decode(enc).expect("an encoded graph decodes")
263
                    } else {
264
                        Value::Bytes(blob_of(enc).0.to_vec())
265
                    };
266
                    Value::Array(vec![Value::Bytes(n.0.to_vec()), entry])
267
                })
268
                .collect(),
269
        );
270
        cbor::encode(&self.payload_value(CHECKPOINT_INDEX_VERSION, files))
271
    }
272
273
    /// The payload map around `files`: changes, tree and replaces as in version 1.
274
    fn payload_value(&self, version: u64, files: Value) -> Value {
275
        let mut hashes: Vec<&ChangeHash> = self.changes.keys().collect();
276
        hashes.sort();
277
        let changes = Value::Array(
278
            hashes
279
                .into_iter()
280
                .map(|h| {
281
                    let m = &self.changes[h];
282
                    Value::Array(vec![
283
                        Value::Bytes(h.0.to_vec()),
284
                        Value::Bytes(m.change_id.0.to_vec()),
285
                        Value::Array(m.deps.iter().map(|d| Value::Bytes(d.0.to_vec())).collect()),
286
                        Value::Text(m.message.clone()),
287
                        Value::Uint(m.hlc.wall_ms),
288
                        Value::Uint(m.hlc.logical as u64),
289
                        Value::Uint(m.kind.tag()),
290
                        Value::Bytes(m.author.0.to_vec()),
291
                    ])
292
                })
293
                .collect(),
294
        );
295
        let mut rnodes: Vec<&NodeId> = self.replaces.keys().collect();
296
        rnodes.sort();
297
        let opt =
298
            |o: &Option<ObjectRef>| o.map(|r| Value::Bytes(r.0.to_vec())).unwrap_or(Value::Null);
299
        let replaces = Value::Array(
300
            rnodes
301
                .into_iter()
302
                .map(|n| {
303
                    Value::Array(vec![
304
                        Value::Bytes(n.0.to_vec()),
305
                        Value::Array(
306
                            self.replaces[n]
307
                                .iter()
308
                                .map(|(c, o, nw)| {
309
                                    Value::Array(vec![Value::Bytes(c.0.to_vec()), opt(o), opt(nw)])
310
                                })
311
                                .collect(),
312
                        ),
313
                    ])
314
                })
315
                .collect(),
316
        );
317
        Value::map([
318
            (0, Value::Uint(version)),
319
            (1, changes),
320
            (2, self.tree.to_value()),
321
            (3, files),
322
            (4, replaces),
323
        ])
324
    }
325
326
    /// Decode a version 1 payload (the whole state in one value). A
327
    /// version 2 payload needs its graph blobs: use [`State::from_checkpoint`].
328
    pub fn from_checkpoint_payload(bytes: &[u8]) -> Result<State, ObjectError> {
329
        State::from_checkpoint(bytes, |_| {
330
            Err(ObjectError::Schema(
331
                "a version 2 checkpoint needs its graph blobs".into(),
332
            ))
333
        })
334
    }
335
336
    /// Decode a checkpoint payload of either version (spec §11). For
337
    /// version 2, `load` returns the contents of the named graph blobs, in
338
    /// order; the graphs are then decoded in parallel.
339
    pub fn from_checkpoint(
340
        bytes: &[u8],
341
        load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
342
    ) -> Result<State, ObjectError> {
343
        use rayon::prelude::*;
344
        let v = cbor::decode(bytes)?;
345
        let m = o::map(&v, "checkpoint payload")?;
346
        let version = o::uint(o::field(m, 0, "version")?, "version")?;
347
        if version != o::FORMAT_VERSION && version != CHECKPOINT_INDEX_VERSION {
348
            return Err(ObjectError::Schema(format!(
349
                "checkpoint payload version {version} is not supported"
350
            )));
351
        }
352
        let mut st = State::payload_common(m)?;
353
        let entries = o::array(o::field(m, 3, "files")?, "files")?;
354
        let mut parsed: Vec<(NodeId, &Value)> = Vec::with_capacity(entries.len());
355
        for f in entries {
356
            let e = o::array(f, "file")?;
357
            if e.len() != 2 {
358
                return Err(ObjectError::Schema("file: expected [node, graph]".into()));
359
            }
360
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
361
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
362
            parsed.push((node, &e[1]));
363
        }
364
        // Version 2 names large graphs by blob: fetch those in one batch.
365
        let mut fetched: HashMap<NodeId, Vec<u8>> = HashMap::new();
366
        if version == CHECKPOINT_INDEX_VERSION {
367
            let wanted: Vec<(NodeId, ObjectRef)> = parsed
368
                .iter()
369
                .filter(|(_, e)| matches!(e, Value::Bytes(_)))
370
                .map(|(n, e)| Ok((*n, o::oref(e, "graph blob")?)))
371
                .collect::<Result<_, ObjectError>>()?;
372
            if !wanted.is_empty() {
373
                let refs: Vec<ObjectRef> = wanted.iter().map(|(_, r)| *r).collect();
374
                let contents = {
375
                    let _t = crate::timing::phase("load:   graph blobs read");
376
                    load(&refs)?
377
                };
378
                if contents.len() != refs.len() {
379
                    return Err(ObjectError::Schema("graph blobs: wrong count".into()));
380
                }
381
                fetched.extend(wanted.iter().map(|(n, _)| *n).zip(contents));
382
            }
383
        } else if parsed.iter().any(|(_, e)| matches!(e, Value::Bytes(_))) {
384
            return Err(ObjectError::Schema(
385
                "a version 1 payload holds graphs inline".into(),
386
            ));
387
        }
388
        // File graphs are independent; decode and build them in parallel.
389
        let graphs = parsed
390
            .par_iter()
391
            .map(|(node, entry)| {
392
                let graph = match fetched.get(node) {
393
                    Some(bytes) => FileGraph::from_value(&cbor::decode(bytes)?)?,
394
                    None => FileGraph::from_value(entry)?,
395
                };
396
                Ok((*node, graph))
397
            })
398
            .collect::<Result<Vec<_>, ObjectError>>()?;
399
        st.files.extend(graphs);
400
        Ok(st)
401
    }
402
403
    /// The graph blobs a payload names: empty for version 1.
404
    pub fn checkpoint_graph_refs(bytes: &[u8]) -> Result<Vec<ObjectRef>, ObjectError> {
405
        let v = cbor::decode(bytes)?;
406
        let m = o::map(&v, "checkpoint payload")?;
407
        let mut out = Vec::new();
408
        for f in o::array(o::field(m, 3, "files")?, "files")? {
409
            if let Some(entry @ Value::Bytes(_)) = o::array(f, "file")?.get(1) {
410
                out.push(o::oref(entry, "graph blob")?);
411
            }
412
        }
413
        Ok(out)
414
    }
415
416
    /// Changes, tree and replaces of a payload of either version.
417
    fn payload_common(m: &BTreeMap<u64, Value>) -> Result<State, ObjectError> {
418
        let mut st = State::new();
419
        for c in o::array(o::field(m, 1, "changes")?, "changes")? {
420
            let e = o::array(c, "change meta")?;
421
            if e.len() != 8 {
422
                return Err(ObjectError::Schema("change meta: expected 8 fields".into()));
423
            }
424
            let hash = o::oref(&e[0], "hash")?;
425
            let meta = ChangeMeta {
426
                change_id: crate::ids::ChangeId::from_slice(o::bytes(&e[1], "change id")?)
427
                    .ok_or_else(|| ObjectError::Schema("change id: expected 16 bytes".into()))?,
428
                deps: o::array(&e[2], "deps")?
429
                    .iter()
430
                    .map(|d| o::oref(d, "dep"))
431
                    .collect::<Result<Vec<_>, _>>()?,
432
                message: o::text(&e[3], "message")?.to_owned(),
433
                hlc: Hlc {
434
                    wall_ms: o::uint(&e[4], "wall")?,
435
                    logical: o::u32_(&e[5], "logical")?,
436
                },
437
                kind: ChangeKind::from_tag(o::uint(&e[6], "kind")?)?,
438
                author: o::oref(&e[7], "author")?,
439
            };
440
            st.changes.insert(hash, meta);
441
        }
442
        st.tree = TreeGraph::from_value(o::field(m, 2, "tree")?)?;
443
        let opt = |v: &Value| -> Result<Option<ObjectRef>, ObjectError> {
444
            if v.is_null() {
445
                Ok(None)
446
            } else {
447
                o::oref(v, "blob").map(Some)
448
            }
449
        };
450
        for r in o::array(o::field(m, 4, "replaces")?, "replaces")? {
451
            let e = o::array(r, "replace")?;
452
            if e.len() != 2 {
453
                return Err(ObjectError::Schema(
454
                    "replace: expected [node, records]".into(),
455
                ));
456
            }
457
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
458
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
459
            let mut recs = Vec::new();
460
            for rec in o::array(&e[1], "records")? {
461
                let x = o::array(rec, "record")?;
462
                if x.len() != 3 {
463
                    return Err(ObjectError::Schema("record: expected 3 fields".into()));
464
                }
465
                recs.push((o::oref(&x[0], "change")?, opt(&x[1])?, opt(&x[2])?));
466
            }
467
            st.replaces.insert(node, recs);
468
        }
469
        Ok(st)
470
    }
471
472
    pub fn validate(&self) -> Result<(), StateError> {
473
        for (h, c) in &self.changes {
474
            for d in &c.deps {
475
                if !self.changes.contains_key(d) {
476
                    return Err(StateError::MissingDep(*d, *h));
477
                }
478
            }
479
        }
480
        Ok(())
481
    }
482
483
    fn apply(&mut self, h: ChangeHash, c: Change) {
484
        for (i, op) in c.tree_ops.iter().enumerate() {
485
            self.tree.apply(h, i as u32, &resolve_tree_op(op, h));
486
        }
487
        let mut counters: HashMap<NodeId, u32> = HashMap::new();
488
        for e in c.edits {
489
            match e {
490
                Edit::Insert {
491
                    node,
492
                    after,
493
                    before,
494
                    lines,
495
                } => {
496
                    let idx = counters.entry(node).or_default();
497
                    let first = *idx;
498
                    *idx += lines.len() as u32;
499
                    self.files.entry(node).or_default().insert_owned(
500
                        h,
501
                        first,
502
                        vertex(after.resolve(h)),
503
                        vertex(before.resolve(h)),
504
                        lines,
505
                    );
506
                }
507
                Edit::Delete { node, atoms } => {
508
                    let g = self.files.entry(node).or_default();
509
                    for r in atoms {
510
                        g.delete(h, &r.resolve(h));
511
                    }
512
                }
513
                Edit::Replace { node, old, new } => {
514
                    self.replaces.entry(node).or_default().push((h, old, new));
515
                }
516
            }
517
        }
518
    }
519
520
    /// Transitive dependencies of `h`, excluding `h` itself.
521
    pub fn closure(&self, h: &ChangeHash) -> Arc<HashSet<ChangeHash>> {
522
        if let Some(c) = self.closure.lock().unwrap().get(h) {
523
            return c.clone();
524
        }
525
        let mut out = HashSet::new();
526
        let mut stack: Vec<ChangeHash> = self
527
            .changes
528
            .get(h)
529
            .map(|c| c.deps.clone())
530
            .unwrap_or_default();
531
        while let Some(d) = stack.pop() {
532
            if out.insert(d) {
533
                if let Some(c) = self.changes.get(&d) {
534
                    stack.extend(c.deps.iter().copied());
535
                }
536
            }
537
        }
538
        let rc = Arc::new(out);
539
        self.closure.lock().unwrap().insert(*h, rc.clone());
540
        rc
541
    }
542
543
    /// Changes that no other change depends on directly, sorted.
544
    pub fn heads(&self) -> Vec<ChangeHash> {
545
        let depended: HashSet<&ChangeHash> =
546
            self.changes.values().flat_map(|c| c.deps.iter()).collect();
547
        let mut heads: Vec<ChangeHash> = self
548
            .changes
549
            .keys()
550
            .filter(|h| !depended.contains(h))
551
            .copied()
552
            .collect();
553
        heads.sort();
554
        heads
555
    }
556
557
    /// Maximal elements among `hashes` (those not in the closure of another),
558
    /// sorted. One walk down from all of them at once: whatever the walk
559
    /// reaches is below some other element. Linear in the changes reached,
560
    /// where comparing every pair's closures was quadratic in the number
561
    /// of heads, which a bookmark of independent changes has many of.
562
    pub fn maximal(&self, hashes: &[ChangeHash]) -> Vec<ChangeHash> {
563
        let set: BTreeSet<ChangeHash> = hashes.iter().copied().collect();
564
        let mut below: HashSet<ChangeHash> = HashSet::new();
565
        let mut stack: Vec<ChangeHash> = set
566
            .iter()
567
            .filter_map(|h| self.changes.get(h))
568
            .flat_map(|c| c.deps.iter().copied())
569
            .collect();
570
        while let Some(d) = stack.pop() {
571
            if below.insert(d) {
572
                if let Some(c) = self.changes.get(&d) {
573
                    stack.extend(c.deps.iter().copied());
574
                }
575
            }
576
        }
577
        set.into_iter().filter(|h| !below.contains(h)).collect()
578
    }
579
580
    pub fn materialize_file(&self, node: &NodeId) -> Materialized {
581
        self.files
582
            .get(node)
583
            .map(|g| g.materialize(self))
584
            .unwrap_or_default()
585
    }
586
587
    /// Live lines of a text node in materialized order.
588
    pub fn live_lines(&self, node: &NodeId) -> Vec<(AtomKey, Vec<u8>)> {
589
        let Some(g) = self.files.get(node) else {
590
            return vec![];
591
        };
592
        let m = g.materialize(self);
593
        m.live_atoms()
594
            .into_iter()
595
            .map(|a| (a, g.content(&a).unwrap_or_default().to_vec()))
596
            .collect()
597
    }
598
599
    /// Live blobs of an opaque node: every `new` that no later `Replace` retired.
600
    pub fn live_blobs(&self, node: &NodeId) -> Vec<ObjectRef> {
601
        let mut live: Vec<ObjectRef> = self
602
            .live_replaces(node)
603
            .into_iter()
604
            .map(|(_, n)| n)
605
            .collect();
606
        live.sort();
607
        live.dedup();
608
        live
609
    }
610
611
    /// The replace records whose blob is live, with the change that set it
612
    /// (spec §9.3). A record's blob is superseded by a later replacement of
613
    /// that blob: one whose change depends on the record's change, as every
614
    /// replacement depends on the change that made its old blob current
615
    /// (§8.1). Two replacements that do not know each other both stay live
616
    /// (a replace conflict), and a file reverted to an earlier blob keeps it.
617
    /// A replacement without that dependency, from before the rule, is read
618
    /// as before: it supersedes every record that did not come after it.
619
    pub fn live_replaces(&self, node: &NodeId) -> Vec<(ChangeHash, ObjectRef)> {
620
        let Some(rs) = self.replaces.get(node) else {
621
            return vec![];
622
        };
623
        let knows_its_old = |s: &ChangeHash, old: &ObjectRef| {
624
            rs.iter()
625
                .any(|(q, _, n)| n.as_ref() == Some(old) && q != s && self.closure(s).contains(q))
626
        };
627
        rs.iter()
628
            .filter_map(|(c, _, n)| n.map(|n| (*c, n)))
629
            .filter(|(c, n)| {
630
                !rs.iter().any(|(s, o, _)| {
631
                    s != c
632
                        && *o == Some(*n)
633
                        && (self.closure(s).contains(c)
634
                            || (!knows_its_old(s, n) && !self.closure(c).contains(s)))
635
                })
636
            })
637
            .collect()
638
    }
639
640
    /// Short label for conflict markers: `changeid-prefix "message"`.
641
    pub fn label(&self, h: &ChangeHash) -> String {
642
        match self.changes.get(h) {
643
            Some(c) => format!(
644
                "{} \"{}\"",
645
                c.change_id.short(8),
646
                c.message.lines().next().unwrap_or("")
647
            ),
648
            None => format!("#{}", h.short(12)),
649
        }
650
    }
651
652
    /// For resolving a conflicted text node (spec §9.6): the live atoms that
653
    /// are not part of any conflict region in order, every live atom that is
654
    /// part of a region, and every change that contributed to a region.
655
    /// The changes that introduced the live blobs of an opaque node, for a
656
    /// replace conflict (spec §9.3).
657
    pub fn blob_contributors(&self, node: &NodeId, blobs: &[ObjectRef]) -> Vec<ChangeHash> {
658
        let mut out: Vec<ChangeHash> = Vec::new();
659
        for (h, _, new) in self.replaces.get(node).map(|v| v.as_slice()).unwrap_or(&[]) {
660
            if new.is_some_and(|b| blobs.contains(&b)) && !out.contains(h) {
661
                out.push(*h);
662
            }
663
        }
664
        out.sort();
665
        out
666
    }
667
668
    pub fn conflict_regions(&self, node: &NodeId) -> ConflictRegions {
669
        let mut out = ConflictRegions::default();
670
        let Some(g) = self.files.get(node) else {
671
            return out;
672
        };
673
        let m = g.materialize(self);
674
        let mut contributors = BTreeSet::new();
675
        for seg in &m.segments {
676
            match seg {
677
                crate::graph::Segment::Lines(atoms) => {
678
                    for a in atoms {
679
                        out.plain
680
                            .push((*a, g.content(a).unwrap_or_default().to_vec()));
681
                    }
682
                }
683
                crate::graph::Segment::Conflict(sides) => {
684
                    for side in sides {
685
                        for a in side {
686
                            out.region_atoms.push(*a);
687
                            contributors.insert(a.0);
688
                        }
689
                    }
690
                }
691
                crate::graph::Segment::Zombie { deleted, inserted } => {
692
                    for a in deleted {
693
                        contributors.insert(a.0);
694
                        contributors.extend(g.deleters(a).copied());
695
                    }
696
                    for a in inserted {
697
                        out.region_atoms.push(*a);
698
                        contributors.insert(a.0);
699
                    }
700
                }
701
            }
702
        }
703
        out.contributors = contributors.into_iter().collect();
704
        out
705
    }
706
707
    pub fn render_file(&self, node: &NodeId) -> (Vec<u8>, Materialized) {
708
        let Some(g) = self.files.get(node) else {
709
            return (Vec::new(), Materialized::default());
710
        };
711
        let m = g.materialize(self);
712
        let bytes = g.render(&m, &|h| self.label(h));
713
        (bytes, m)
714
    }
715
716
    /// Every file rendered, in parallel: files are independent, and
717
    /// rendering them is most of a snapshot's cost in a large tree.
718
    pub fn materialize(&self) -> Snapshot {
719
        use rayon::prelude::*;
720
        let tree = self.tree.view();
721
        let entries: Vec<(&String, &crate::tree::TreeEntry)> = tree.entries.iter().collect();
722
        let rendered: Vec<(String, FileState)> = entries
723
            .par_iter()
724
            .map(|(path, e)| (path.to_string(), self.file_state(e)))
725
            .collect();
726
        let files: BTreeMap<String, FileState> = rendered.into_iter().collect();
727
        Snapshot { tree, files }
728
    }
729
730
    fn file_state(&self, e: &crate::tree::TreeEntry) -> FileState {
731
        {
732
            let content = match e.kind {
733
                NodeKind::Dir => FileContent::Dir,
734
                NodeKind::Symlink | NodeKind::Subrepo => FileContent::Opaque {
735
                    blobs: self.live_blobs(&e.node),
736
                },
737
                NodeKind::File if e.mode.opaque => FileContent::Opaque {
738
                    blobs: self.live_blobs(&e.node),
739
                },
740
                NodeKind::File => {
741
                    let (bytes, materialized) = self.render_file(&e.node);
742
                    FileContent::Text {
743
                        bytes,
744
                        materialized,
745
                    }
746
                }
747
            };
748
            FileState {
749
                node: e.node,
750
                kind: e.kind,
751
                mode: e.mode,
752
                content,
753
            }
754
        }
755
    }
756
}
757
758
impl DepOracle for State {
759
    fn depends_on(&self, a: &ChangeHash, b: &ChangeHash) -> bool {
760
        a == b || self.closure(a).contains(b)
761
    }
762
}
763
764
#[cfg(test)]
765
mod tests {
766
    use super::*;
767
    use crate::ids::ChangeId;
768
    use crate::objects::{AtomRange, ChangeKind, Hlc, OpRef};
769
770
    fn change(deps: Vec<ChangeHash>, tree_ops: Vec<TreeOp>, edits: Vec<Edit>) -> Change {
771
        let mut c = Change {
772
            change_id: ChangeId::random(),
773
            deps,
774
            author: ObjectRef::ZERO,
775
            hlc: Hlc {
776
                wall_ms: 1,
777
                logical: 0,
778
            },
779
            message: "m".into(),
780
            tree_ops,
781
            edits,
782
            kind: ChangeKind::Normal,
783
        };
784
        c.normalize();
785
        c
786
    }
787
788
    #[test]
789
    fn build_edit_and_materialize() {
790
        let mut st = State::new();
791
        let f = NodeId::random();
792
        let c1 = change(
793
            vec![],
794
            vec![TreeOp::AddNode {
795
                node: f,
796
                kind: NodeKind::File,
797
                mode: Mode::TEXT,
798
                parent: NodeId::ROOT,
799
                name: "a.txt".into(),
800
            }],
801
            vec![Edit::Insert {
802
                node: f,
803
                after: AtomRef::Start,
804
                before: AtomRef::End,
805
                lines: vec![b"one\n".to_vec(), b"two\n".to_vec()],
806
            }],
807
        );
808
        let h1 = st.add(c1).unwrap();
809
        // Depends on h1: append a line and delete "one".
810
        let c2 = change(
811
            vec![h1],
812
            vec![],
813
            vec![
814
                Edit::Delete {
815
                    node: f,
816
                    atoms: vec![AtomRange {
817
                        change: h1,
818
                        from: 0,
819
                        to: 1,
820
                    }],
821
                },
822
                Edit::Insert {
823
                    node: f,
824
                    after: AtomRef::Atom {
825
                        change: h1,
826
                        index: 1,
827
                    },
828
                    before: AtomRef::End,
829
                    lines: vec![b"three\n".to_vec()],
830
                },
831
            ],
832
        );
833
        let h2 = st.add(c2).unwrap();
834
        assert_eq!(st.heads(), vec![h2]);
835
        assert!(st.depends_on(&h2, &h1));
836
        assert!(!st.depends_on(&h1, &h2));
837
        let snap = st.materialize();
838
        assert!(!snap.is_conflicted());
839
        let a = &snap.files["a.txt"];
840
        assert!(matches!(&a.content, FileContent::Text { bytes, .. } if bytes == b"two\nthree\n"));
841
        assert_eq!(st.live_lines(&f).len(), 2);
842
    }
843
844
    #[test]
845
    fn missing_dependency_is_rejected() {
846
        let mut st = State::new();
847
        let c = change(vec![ObjectRef([1; 32])], vec![], vec![]);
848
        assert!(matches!(st.add(c), Err(StateError::MissingDep(..))));
849
    }
850
851
    #[test]
852
    fn self_references_and_multi_insert_indexing() {
853
        let mut st = State::new();
854
        let f = NodeId::random();
855
        let me = OpRef {
856
            change: ObjectRef::ZERO,
857
            index: 0,
858
        };
859
        let c1 = change(
860
            vec![],
861
            vec![
862
                TreeOp::AddNode {
863
                    node: f,
864
                    kind: NodeKind::File,
865
                    mode: Mode::TEXT,
866
                    parent: NodeId::ROOT,
867
                    name: "x".into(),
868
                },
869
                TreeOp::SetMode {
870
                    node: f,
871
                    old: me,
872
                    mode: Mode {
873
                        exec: true,
874
                        opaque: false,
875
                    },
876
                },
877
            ],
878
            vec![
879
                Edit::Insert {
880
                    node: f,
881
                    after: AtomRef::Start,
882
                    before: AtomRef::End,
883
                    lines: vec![b"a\n".to_vec(), b"c\n".to_vec()],
884
                },
885
                // Insert between the two lines just inserted, using self references.
886
                Edit::Insert {
887
                    node: f,
888
                    after: AtomRef::Atom {
889
                        change: ObjectRef::ZERO,
890
                        index: 0,
891
                    },
892
                    before: AtomRef::Atom {
893
                        change: ObjectRef::ZERO,
894
                        index: 1,
895
                    },
896
                    lines: vec![b"b\n".to_vec()],
897
                },
898
            ],
899
        );
900
        st.add(c1).unwrap();
901
        let snap = st.materialize();
902
        let x = &snap.files["x"];
903
        assert_eq!(
904
            x.mode,
905
            Mode {
906
                exec: true,
907
                opaque: false
908
            }
909
        );
910
        assert!(matches!(&x.content, FileContent::Text { bytes, .. } if bytes == b"a\nb\nc\n"));
911
    }
912
913
    #[test]
914
    fn opaque_replace_and_conflict() {
915
        let mut st = State::new();
916
        let f = NodeId::random();
917
        let b1 = ObjectRef([1; 32]);
918
        let b2 = ObjectRef([2; 32]);
919
        let b3 = ObjectRef([3; 32]);
920
        let h1 = st
921
            .add(change(
922
                vec![],
923
                vec![TreeOp::AddNode {
924
                    node: f,
925
                    kind: NodeKind::File,
926
                    mode: Mode::OPAQUE,
927
                    parent: NodeId::ROOT,
928
                    name: "img".into(),
929
                }],
930
                vec![Edit::Replace {
931
                    node: f,
932
                    old: None,
933
                    new: Some(b1),
934
                }],
935
            ))
936
            .unwrap();
937
        assert_eq!(st.live_blobs(&f), vec![b1]);
938
        st.add(change(
939
            vec![h1],
940
            vec![],
941
            vec![Edit::Replace {
942
                node: f,
943
                old: Some(b1),
944
                new: Some(b2),
945
            }],
946
        ))
947
        .unwrap();
948
        st.add(change(
949
            vec![h1],
950
            vec![],
951
            vec![Edit::Replace {
952
                node: f,
953
                old: Some(b1),
954
                new: Some(b3),
955
            }],
956
        ))
957
        .unwrap();
958
        let snap = st.materialize();
959
        assert!(snap.is_conflicted());
960
        assert!(
961
            matches!(&snap.files["img"].content, FileContent::Opaque { blobs } if blobs == &vec![b2, b3])
962
        );
963
    }
964
965
    #[test]
966
    fn checkpoint_payload_round_trip_and_replay() {
967
        let mut st = State::new();
968
        let f = NodeId::random();
969
        let h1 = st
970
            .add(change(
971
                vec![],
972
                vec![TreeOp::AddNode {
973
                    node: f,
974
                    kind: NodeKind::File,
975
                    mode: Mode::TEXT,
976
                    parent: NodeId::ROOT,
977
                    name: "a".into(),
978
                }],
979
                vec![Edit::Insert {
980
                    node: f,
981
                    after: AtomRef::Start,
982
                    before: AtomRef::End,
983
                    lines: vec![b"1\n".to_vec(), b"2\n".to_vec()],
984
                }],
985
            ))
986
            .unwrap();
987
        let h2 = st
988
            .add(change(
989
                vec![h1],
990
                vec![],
991
                vec![Edit::Delete {
992
                    node: f,
993
                    atoms: vec![AtomRange {
994
                        change: h1,
995
                        from: 0,
996
                        to: 1,
997
                    }],
998
                }],
999
            ))
1000
            .unwrap();
1001
        let payload = st.checkpoint_payload();
1002
        let back = State::from_checkpoint_payload(&payload).unwrap();
1003
        assert_eq!(back.len(), 2);
1004
        assert_eq!(back.heads(), vec![h2]);
1005
        assert_eq!(back.get(&h1).unwrap().message, "m");
1006
        assert_eq!(back.checkpoint_payload(), payload, "deterministic");
1007
        // Replaying a further change onto the restored state equals a full build.
1008
        let c3 = change(
1009
            vec![h2],
1010
            vec![],
1011
            vec![Edit::Insert {
1012
                node: f,
1013
                after: AtomRef::Atom {
1014
                    change: h1,
1015
                    index: 1,
1016
                },
1017
                before: AtomRef::End,
1018
                lines: vec![b"3\n".to_vec()],
1019
            }],
1020
        );
1021
        let mut full = st.clone();
1022
        full.add(c3.clone()).unwrap();
1023
        let mut restored = back;
1024
        restored.add(c3).unwrap();
1025
        let (a, b) = (full.materialize(), restored.materialize());
1026
        assert_eq!(
1027
            a.files.keys().collect::<Vec<_>>(),
1028
            b.files.keys().collect::<Vec<_>>()
1029
        );
1030
        assert_eq!(a.files["a"].content, b.files["a"].content);
1031
        assert!(
1032
            matches!(&b.files["a"].content, FileContent::Text { bytes, .. } if bytes == b"2\n3\n")
1033
        );
1034
        assert!(State::from_checkpoint_payload(b"nonsense").is_err());
1035
    }
1036
1037
    /// Version 2 payloads (spec §11): small graphs inline, large ones by
1038
    /// blob; the state read back is the same as from version 1.
1039
    #[test]
1040
    fn checkpoint_index_round_trip_matches_version_1() {
1041
        let mut st = State::new();
1042
        let small = NodeId::random();
1043
        let big = NodeId::random();
1044
        let add = |node: NodeId, name: &str| TreeOp::AddNode {
1045
            node,
1046
            kind: NodeKind::File,
1047
            mode: Mode::TEXT,
1048
            parent: NodeId::ROOT,
1049
            name: name.into(),
1050
        };
1051
        let big_lines: Vec<Vec<u8>> = (0..400)
1052
            .map(|i| format!("line {i} of a long file\n").into_bytes())
1053
            .collect();
1054
        let h1 = st
1055
            .add(change(
1056
                vec![],
1057
                vec![add(small, "small"), add(big, "big")],
1058
                vec![
1059
                    Edit::Insert {
1060
                        node: small,
1061
                        after: AtomRef::Start,
1062
                        before: AtomRef::End,
1063
                        lines: vec![b"x\n".to_vec()],
1064
                    },
1065
                    Edit::Insert {
1066
                        node: big,
1067
                        after: AtomRef::Start,
1068
                        before: AtomRef::End,
1069
                        lines: big_lines,
1070
                    },
1071
                ],
1072
            ))
1073
            .unwrap();
1074
        st.add(change(
1075
            vec![h1],
1076
            vec![],
1077
            vec![Edit::Delete {
1078
                node: big,
1079
                atoms: vec![AtomRange {
1080
                    change: h1,
1081
                    from: 3,
1082
                    to: 7,
1083
                }],
1084
            }],
1085
        ))
1086
        .unwrap();
1087
1088
        let graphs = st.checkpoint_file_graphs();
1089
        let mut blobs: HashMap<ObjectRef, Vec<u8>> = HashMap::new();
1090
        let index = st.checkpoint_index(&graphs, |enc| {
1091
            let r = crate::hash::hash_object(crate::hash::ObjectType::Blob, enc);
1092
            blobs.insert(r, enc.to_vec());
1093
            r
1094
        });
1095
        assert_eq!(blobs.len(), 1, "only the large graph is a blob of its own");
1096
        assert_eq!(
1097
            State::checkpoint_graph_refs(&index).unwrap(),
1098
            blobs.keys().copied().collect::<Vec<_>>()
1099
        );
1100
        assert!(
1101
            State::checkpoint_graph_refs(&st.checkpoint_payload())
1102
                .unwrap()
1103
                .is_empty(),
1104
            "version 1 names none"
1105
        );
1106
        assert!(index.len() < st.checkpoint_payload().len());
1107
1108
        let back = State::from_checkpoint(&index, |refs| {
1109
            refs.iter()
1110
                .map(|r| {
1111
                    blobs
1112
                        .get(r)
1113
                        .cloned()
1114
                        .ok_or_else(|| ObjectError::Schema("missing".into()))
1115
                })
1116
                .collect()
1117
        })
1118
        .unwrap();
1119
        assert_eq!(
1120
            back.checkpoint_payload(),
1121
            st.checkpoint_payload(),
1122
            "the same state as version 1"
1123
        );
1124
        assert_eq!(
1125
            back.materialize().files["big"].content,
1126
            st.materialize().files["big"].content
1127
        );
1128
        // Deterministic: the same state gives the same index.
1129
        assert_eq!(
1130
            back.checkpoint_index(&back.checkpoint_file_graphs(), |enc| {
1131
                crate::hash::hash_object(crate::hash::ObjectType::Blob, enc)
1132
            }),
1133
            index
1134
        );
1135
        // A missing graph blob is an error, not an empty file.
1136
        assert!(
1137
            State::from_checkpoint(&index, |_| Err(ObjectError::Schema("gone".into()))).is_err()
1138
        );
1139
        assert!(
1140
            State::from_checkpoint_payload(&index).is_err(),
1141
            "version 2 needs its blobs"
1142
        );
1143
        // Version 1 payloads still read, and unknown versions do not.
1144
        assert!(State::from_checkpoint(&st.checkpoint_payload(), |_| unreachable!()).is_ok());
1145
        let mut v3 = cbor::decode(&index).unwrap();
1146
        if let Value::Map(m) = &mut v3 {
1147
            m.insert(0, Value::Uint(3));
1148
        }
1149
        assert!(State::from_checkpoint(&cbor::encode(&v3), |_| unreachable!()).is_err());
1150
    }
1151
1152
    #[test]
1153
    fn maximal_elements() {
1154
        let mut st = State::new();
1155
        let h1 = st.add(change(vec![], vec![], vec![])).unwrap();
1156
        let h2 = st.add(change(vec![h1], vec![], vec![])).unwrap();
1157
        let h3 = st.add(change(vec![], vec![], vec![])).unwrap();
1158
        let mut m = st.maximal(&[h1, h2, h3]);
1159
        m.sort();
1160
        let mut want = vec![h2, h3];
1161
        want.sort();
1162
        assert_eq!(m, want);
1163
        assert_eq!(st.heads(), want);
1164
    }
1165
1166
    /// The one-walk `maximal` agrees with the definition (not in the
1167
    /// closure of another element) on random graphs, with repeated and
1168
    /// unknown hashes among the input.
1169
    #[test]
1170
    fn maximal_matches_its_definition() {
1171
        use rand::{Rng, SeedableRng};
1172
        let mut rng = rand::rngs::StdRng::seed_from_u64(7);
1173
        for _ in 0..20 {
1174
            let mut st = State::new();
1175
            let mut all: Vec<ChangeHash> = Vec::new();
1176
            for _ in 0..60 {
1177
                let deps: Vec<ChangeHash> =
1178
                    all.iter().copied().filter(|_| rng.gen_bool(0.05)).collect();
1179
                all.push(st.add(change(deps, vec![], vec![])).unwrap());
1180
            }
1181
            let mut pick: Vec<ChangeHash> =
1182
                all.iter().copied().filter(|_| rng.gen_bool(0.4)).collect();
1183
            pick.extend(pick.clone().into_iter().take(3));
1184
            pick.push(ObjectRef([0xee; 32]));
1185
            let set: BTreeSet<ChangeHash> = pick.iter().copied().collect();
1186
            let want: Vec<ChangeHash> = set
1187
                .iter()
1188
                .filter(|h| !set.iter().any(|o| o != *h && st.closure(o).contains(h)))
1189
                .copied()
1190
                .collect();
1191
            assert_eq!(st.maximal(&pick), want);
1192
            assert_eq!(st.maximal(&all), st.heads());
1193
        }
1194
    }
1195
}