Skip to main content

wire/
object_graph.rs

1// SPDX-License-Identifier: Apache-2.0
2use std::collections::{HashSet, VecDeque};
3
4use objects::{
5    object::{
6        ContentHash, SemanticEntryKind, SemanticIndexRoot, SemanticTreeNode, State, StateAttachment,
7        StateAttachmentBody, StateAttachmentId, StateId, TreeEntryTarget,
8    },
9    store::{ObjectStore, pack::ObjectType as PackObjectType},
10};
11use serde::{Deserialize, Serialize};
12
13use crate::{ProtocolError, Result};
14
15#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
16pub enum ObjectId {
17    Hash(ContentHash),
18    StateId(StateId),
19    StateAttachment {
20        state: StateId,
21        id: StateAttachmentId,
22    },
23}
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct ObjectInfo {
27    pub id: ObjectId,
28    pub obj_type: ObjectType,
29    pub size: u64,
30    pub delta_base: Option<ContentHash>,
31}
32
33#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
34pub struct PlannedObject {
35    pub id: ObjectId,
36    pub obj_type: ObjectType,
37}
38
39#[derive(Debug, Clone)]
40pub struct StateClosureTransferObjects {
41    pub planned_objects: Vec<PlannedObject>,
42    pub full_objects: Option<Vec<ObjectInfo>>,
43}
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
46pub enum ObjectType {
47    Blob,
48    Tree,
49    State,
50    Action,
51    /// A `RedactionsBlob` sidecar — the rmp-encoded record(s) declaring
52    /// that a specific blob has been redacted (and possibly purged) by
53    /// an authorized operator. Keyed on the wire by `ObjectId::Hash` of
54    /// the *redacted blob*, since `Repository`'s sidecar store is
55    /// indexed that way.
56    Redaction,
57    /// A `StateVisibilityBlob` sidecar — the rmp-encoded record(s)
58    /// declaring a non-public audience tier for a specific state. Keyed
59    /// on the wire by `ObjectId::StateId` of the state, since the
60    /// per-state sidecar store is indexed that way. Like `Redaction`, it
61    /// is a sidecar record that lives outside the content-addressed pack
62    /// and ships via the per-object transfer path, not the pack.
63    StateVisibility,
64    StateAttachment,
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
68pub enum ObjectTypeBucket {
69    Blob,
70    Tree,
71    State,
72    Action,
73    Redaction,
74    StateVisibility,
75    StateAttachment,
76}
77
78impl ObjectType {
79    pub fn wire_name(self) -> &'static str {
80        match self {
81            ObjectType::Blob => "blob",
82            ObjectType::Tree => "tree",
83            ObjectType::State => "state",
84            ObjectType::Action => "action",
85            ObjectType::Redaction => "redaction",
86            ObjectType::StateVisibility => "state_visibility",
87            ObjectType::StateAttachment => "state_attachment",
88        }
89    }
90
91    pub fn from_wire(value: &str) -> Result<Self> {
92        match value {
93            "blob" => Ok(ObjectType::Blob),
94            "tree" => Ok(ObjectType::Tree),
95            "state" => Ok(ObjectType::State),
96            "action" => Ok(ObjectType::Action),
97            "redaction" => Ok(ObjectType::Redaction),
98            "state_visibility" => Ok(ObjectType::StateVisibility),
99            "state_attachment" => Ok(ObjectType::StateAttachment),
100            _ => Err(ProtocolError::InvalidState(format!(
101                "unknown object type: {value}"
102            ))),
103        }
104    }
105
106    pub fn packable(self) -> bool {
107        !matches!(self, ObjectType::Redaction | ObjectType::StateVisibility)
108    }
109
110    pub fn pack_object_type(self) -> Result<PackObjectType> {
111        match self {
112            ObjectType::Blob => Ok(PackObjectType::Blob),
113            ObjectType::Tree => Ok(PackObjectType::Tree),
114            ObjectType::State => Ok(PackObjectType::State),
115            ObjectType::Action => Ok(PackObjectType::Action),
116            ObjectType::StateAttachment => Ok(PackObjectType::StateAttachment),
117            ObjectType::Redaction => Err(ProtocolError::InvalidState(
118                "Redaction sidecar records cannot be packed into the content-addressed object pack"
119                    .to_string(),
120            )),
121            ObjectType::StateVisibility => Err(ProtocolError::InvalidState(
122                "StateVisibility sidecar records cannot be packed into the content-addressed object pack"
123                    .to_string(),
124            )),
125        }
126    }
127
128    pub fn bucket(self) -> ObjectTypeBucket {
129        match self {
130            ObjectType::Blob => ObjectTypeBucket::Blob,
131            ObjectType::Tree => ObjectTypeBucket::Tree,
132            ObjectType::State => ObjectTypeBucket::State,
133            ObjectType::Action => ObjectTypeBucket::Action,
134            ObjectType::Redaction => ObjectTypeBucket::Redaction,
135            ObjectType::StateVisibility => ObjectTypeBucket::StateVisibility,
136            ObjectType::StateAttachment => ObjectTypeBucket::StateAttachment,
137        }
138    }
139}
140
141#[derive(Debug, Clone, Default)]
142pub struct StateClosureOptions {
143    pub depth: Option<u32>,
144    pub exclude_states: Vec<StateId>,
145}
146
147pub fn enumerate_state_closure(
148    store: &impl ObjectStore,
149    state_id: StateId,
150) -> Result<Vec<ObjectInfo>> {
151    enumerate_state_closure_with_options(store, state_id, StateClosureOptions::default())
152}
153
154pub fn enumerate_state_closure_with_options(
155    store: &impl ObjectStore,
156    state_id: StateId,
157    options: StateClosureOptions,
158) -> Result<Vec<ObjectInfo>> {
159    let mut out = Vec::new();
160    walk_state_closure(store, state_id, options, |event| {
161        if let Some(info) = object_info_from_event(store, event)? {
162            out.push(info);
163        }
164        Ok(())
165    })?;
166
167    Ok(out)
168}
169
170pub fn enumerate_state_closure_plan(
171    store: &impl ObjectStore,
172    state_id: StateId,
173) -> Result<Vec<PlannedObject>> {
174    enumerate_state_closure_plan_with_options(store, state_id, StateClosureOptions::default())
175}
176
177pub fn enumerate_state_closure_plan_with_options(
178    store: &impl ObjectStore,
179    state_id: StateId,
180    options: StateClosureOptions,
181) -> Result<Vec<PlannedObject>> {
182    let mut out = Vec::new();
183    walk_state_closure(store, state_id, options, |event| {
184        if let Some(object) = planned_object_from_event(store, event)? {
185            out.push(object);
186        }
187        Ok(())
188    })?;
189
190    Ok(out)
191}
192
193pub fn enumerate_state_closure_transfer_with_options(
194    store: &impl ObjectStore,
195    state_id: StateId,
196    options: StateClosureOptions,
197    full_descriptor_object_threshold: usize,
198) -> Result<StateClosureTransferObjects> {
199    let mut planned_objects = Vec::new();
200    let mut full_objects = Some(Vec::new());
201
202    walk_state_closure(store, state_id, options, |event| {
203        if let Some(object) = planned_object_from_event(store, event)? {
204            planned_objects.push(object);
205        }
206
207        if full_objects.is_some() && planned_objects.len() > full_descriptor_object_threshold {
208            full_objects = None;
209        }
210        if let Some(objects) = full_objects.as_mut()
211            && let Some(info) = object_info_from_event(store, event)?
212        {
213            objects.push(info);
214        }
215
216        Ok(())
217    })?;
218
219    Ok(StateClosureTransferObjects {
220        planned_objects,
221        full_objects,
222    })
223}
224
225#[derive(Debug, Clone, Copy, PartialEq, Eq)]
226enum BlobSource {
227    Tree,
228    StateMetadata,
229}
230
231#[derive(Debug, Clone, Copy)]
232enum StateClosureEvent<'a> {
233    State {
234        id: StateId,
235        state: &'a State,
236    },
237    Tree {
238        hash: ContentHash,
239        tree: &'a objects::object::Tree,
240    },
241    Blob {
242        hash: ContentHash,
243        source: BlobSource,
244    },
245    Redaction {
246        blob: ContentHash,
247    },
248    StateVisibility {
249        state: StateId,
250    },
251    StateAttachment {
252        state: StateId,
253        attachment: &'a StateAttachment,
254    },
255    ExcludedState {
256        id: StateId,
257    },
258    ExcludedHash {
259        hash: ContentHash,
260    },
261}
262
263fn walk_state_closure(
264    store: &impl ObjectStore,
265    state_id: StateId,
266    options: StateClosureOptions,
267    mut visit: impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
268) -> Result<()> {
269    let (excluded_states, excluded_hashes) = collect_excluded(store, &options.exclude_states)?;
270
271    let mut seen_states: HashSet<StateId> = HashSet::new();
272    let mut seen_hashes: HashSet<ContentHash> = HashSet::new();
273    let mut queue: VecDeque<(StateId, u32)> = VecDeque::new();
274    queue.push_back((state_id, 0));
275
276    while let Some((id, depth)) = queue.pop_front() {
277        if excluded_states.contains(&id) {
278            visit(StateClosureEvent::ExcludedState { id })?;
279            continue;
280        }
281        if !seen_states.insert(id) {
282            continue;
283        }
284
285        let state = store
286            .get_state(&id)?
287            .ok_or_else(|| ProtocolError::ObjectNotFound(id.to_string()))?;
288
289        visit(StateClosureEvent::State { id, state: &state })?;
290        if store.has_state_visibility_for_state(&id)? {
291            visit(StateClosureEvent::StateVisibility { state: id })?;
292        }
293        for attachment in store.list_state_attachments(&id)? {
294            visit(StateClosureEvent::StateAttachment {
295                state: id,
296                attachment: &attachment,
297            })?;
298            match attachment.body {
299                StateAttachmentBody::Context(root) => walk_tree_closure_filtered(
300                    store,
301                    root,
302                    &excluded_hashes,
303                    &mut seen_hashes,
304                    &mut visit,
305                )?,
306                StateAttachmentBody::RiskSignals(hash)
307                | StateAttachmentBody::ReviewSignatures(hash)
308                | StateAttachmentBody::Discussions(hash)
309                | StateAttachmentBody::StructuredConflicts(hash) => walk_blob_filtered(
310                    store,
311                    hash,
312                    BlobSource::StateMetadata,
313                    &excluded_hashes,
314                    &mut seen_hashes,
315                    &mut visit,
316                )?,
317                StateAttachmentBody::SemanticIndex(root) => walk_semantic_index_closure(
318                    store,
319                    root,
320                    &excluded_hashes,
321                    &mut seen_hashes,
322                    &mut visit,
323                )?,
324                StateAttachmentBody::Signature(_) => {}
325            }
326        }
327
328        if options.depth.map(|max| depth < max).unwrap_or(true) {
329            for parent in &state.parents {
330                queue.push_back((*parent, depth + 1));
331            }
332        }
333
334        walk_tree_closure_filtered(
335            store,
336            state.tree,
337            &excluded_hashes,
338            &mut seen_hashes,
339            &mut visit,
340        )?;
341        if let Some(provenance_root) = state.provenance {
342            walk_tree_closure_filtered(
343                store,
344                provenance_root,
345                &excluded_hashes,
346                &mut seen_hashes,
347                &mut visit,
348            )?;
349        }
350    }
351
352    Ok(())
353}
354
355fn walk_tree_closure_filtered(
356    store: &impl ObjectStore,
357    tree_hash: ContentHash,
358    excluded: &HashSet<ContentHash>,
359    seen: &mut HashSet<ContentHash>,
360    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
361) -> Result<()> {
362    if excluded.contains(&tree_hash) {
363        visit(StateClosureEvent::ExcludedHash { hash: tree_hash })?;
364        return Ok(());
365    }
366    if !seen.insert(tree_hash) {
367        return Ok(());
368    }
369
370    let tree = store
371        .get_tree(&tree_hash)?
372        .ok_or_else(|| ProtocolError::ObjectNotFound(tree_hash.to_hex()))?;
373
374    visit(StateClosureEvent::Tree {
375        hash: tree_hash,
376        tree: &tree,
377    })?;
378
379    for entry in tree.entries() {
380        match entry.target() {
381            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
382                walk_blob_filtered(store, *hash, BlobSource::Tree, excluded, seen, visit)?;
383            }
384            TreeEntryTarget::Tree { hash } => {
385                walk_tree_closure_filtered(store, *hash, excluded, seen, visit)?;
386            }
387            TreeEntryTarget::Gitlink { .. } => {}
388            // Native child-spool edge: its target lives in a separate spool
389            // object graph, not this store, so it is not walked here.
390            TreeEntryTarget::Spoollink { .. } => {}
391        }
392    }
393
394    Ok(())
395}
396
397fn walk_blob_filtered(
398    store: &impl ObjectStore,
399    blob_hash: ContentHash,
400    source: BlobSource,
401    excluded: &HashSet<ContentHash>,
402    seen: &mut HashSet<ContentHash>,
403    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
404) -> Result<()> {
405    if excluded.contains(&blob_hash) {
406        visit(StateClosureEvent::ExcludedHash { hash: blob_hash })?;
407        return Ok(());
408    }
409    if !seen.insert(blob_hash) {
410        return Ok(());
411    }
412    visit(StateClosureEvent::Blob {
413        hash: blob_hash,
414        source,
415    })?;
416    if store.has_redactions_for_blob(&blob_hash)? {
417        visit(StateClosureEvent::Redaction { blob: blob_hash })?;
418    }
419    Ok(())
420}
421
422/// Walk the merkle semantic-index closure rooted at `root_hash` (a
423/// `SemanticIndexRoot` blob), emitting every reachable semantic node blob.
424///
425/// All semantic-index nodes (root, tree nodes, file nodes) are stored as
426/// ordinary content-addressed blobs, so replication just enumerates them.
427/// Opaque entries point back at raw source blobs already covered by the state's
428/// tree closure, so they are not re-walked here.
429///
430/// Iterative (explicit stack) so a crafted deep `SemanticTreeNode` chain in a
431/// pushed state can't overflow the stack. A missing or undecodable node in the
432/// closure is a HARD failure (`ObjectNotFound`/`Serialization`) — a partial or
433/// corrupt semantic closure must never be shipped silently.
434fn walk_semantic_index_closure(
435    store: &impl ObjectStore,
436    root_hash: ContentHash,
437    excluded: &HashSet<ContentHash>,
438    seen: &mut HashSet<ContentHash>,
439    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
440) -> Result<()> {
441    // Stack of (node_hash, is_tree_node): the root and dir nodes must be decoded
442    // to enumerate their children; file/opaque leaves are emitted only.
443    let mut stack: Vec<ContentHash> = vec![root_hash];
444    while let Some(node_hash) = stack.pop() {
445        if !emit_semantic_blob(store, node_hash, excluded, seen, visit)? {
446            continue; // excluded (present on the far side) or already seen.
447        }
448        let blob = store
449            .get_blob(&node_hash)?
450            .ok_or_else(|| ProtocolError::ObjectNotFound(node_hash.to_hex()))?;
451        // The root and tree nodes decode to the same `SemanticTreeNode.entries`
452        // shape after the root's one indirection; walk uniformly by decoding
453        // every interior node as a tree node, tolerating the root shape.
454        let node = decode_semantic_container(&blob, node_hash)?;
455        for child in node {
456            match child {
457                SemanticChild::Interior(hash) => stack.push(hash),
458                SemanticChild::Leaf(hash) => {
459                    // Emit the leaf (file node) blob; it has no children.
460                    emit_semantic_blob(store, hash, excluded, seen, visit)?;
461                }
462            }
463        }
464    }
465    Ok(())
466}
467
468/// The two child kinds encountered walking the semantic closure: an interior
469/// node to descend into, or a leaf (file node) blob to emit.
470enum SemanticChild {
471    Interior(ContentHash),
472    Leaf(ContentHash),
473}
474
475/// Decode a semantic-closure node — the root (which points at a tree node) or a
476/// tree node — into the child hashes to walk. Missing/corrupt is a hard error.
477fn decode_semantic_container(
478    blob: &objects::object::Blob,
479    node_hash: ContentHash,
480) -> Result<Vec<SemanticChild>> {
481    // Try the root shape first (it has an extra `tree` indirection), then the
482    // tree-node shape. Content-addressed hashes make this unambiguous in
483    // practice; a blob that decodes as neither is corrupt.
484    if let Ok(root) = SemanticIndexRoot::decode(blob.content()) {
485        return Ok(vec![SemanticChild::Interior(root.tree)]);
486    }
487    let node = SemanticTreeNode::decode(blob.content())
488        .map_err(|err| ProtocolError::Serialization(format!("semantic node {node_hash}: {err}")))?;
489    Ok(node
490        .entries
491        .iter()
492        .filter_map(|entry| match entry.kind {
493            SemanticEntryKind::Dir => Some(SemanticChild::Interior(entry.node)),
494            SemanticEntryKind::File => Some(SemanticChild::Leaf(entry.node)),
495            // Opaque `node` is the raw source blob, already in the tree closure.
496            SemanticEntryKind::Opaque => None,
497        })
498        .collect())
499}
500
501/// Emit a semantic node blob as state metadata. Returns `true` when the blob
502/// was newly visited (so the caller may descend into it), `false` when it was
503/// excluded (present on the far side) or already seen. A blob that is neither
504/// excluded nor present is a HARD failure — the closure must be complete.
505fn emit_semantic_blob(
506    store: &impl ObjectStore,
507    hash: ContentHash,
508    excluded: &HashSet<ContentHash>,
509    seen: &mut HashSet<ContentHash>,
510    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
511) -> Result<bool> {
512    if excluded.contains(&hash) {
513        visit(StateClosureEvent::ExcludedHash { hash })?;
514        return Ok(false);
515    }
516    if !seen.insert(hash) {
517        return Ok(false);
518    }
519    if store.get_blob(&hash)?.is_none() {
520        return Err(ProtocolError::ObjectNotFound(hash.to_hex()));
521    }
522    visit(StateClosureEvent::Blob {
523        hash,
524        source: BlobSource::StateMetadata,
525    })?;
526    Ok(true)
527}
528
529/// Collect every semantic-index node blob hash reachable from `root_hash` into
530/// `excluded`, for the have-set computation. Iterative + tolerant (a broken
531/// have-set index is not fatal — it just means fewer objects are marked
532/// already-present, which is safe over-fetching, not corruption).
533fn collect_semantic_hashes(
534    store: &impl ObjectStore,
535    root_hash: ContentHash,
536    excluded: &mut HashSet<ContentHash>,
537) -> Result<()> {
538    let mut stack: Vec<ContentHash> = vec![root_hash];
539    while let Some(node_hash) = stack.pop() {
540        if !excluded.insert(node_hash) {
541            continue;
542        }
543        let Some(blob) = store.get_blob(&node_hash)? else {
544            continue;
545        };
546        let children = match decode_semantic_container(&blob, node_hash) {
547            Ok(children) => children,
548            Err(_) => continue,
549        };
550        for child in children {
551            match child {
552                SemanticChild::Interior(hash) => stack.push(hash),
553                SemanticChild::Leaf(hash) => {
554                    excluded.insert(hash);
555                }
556            }
557        }
558    }
559    Ok(())
560}
561
562fn object_info_from_event(
563    store: &impl ObjectStore,
564    event: StateClosureEvent<'_>,
565) -> Result<Option<ObjectInfo>> {
566    match event {
567        StateClosureEvent::State { id, state } => {
568            let state_bytes = rmp_serde::to_vec_named(state)?;
569            Ok(Some(ObjectInfo {
570                id: ObjectId::StateId(id),
571                obj_type: ObjectType::State,
572                size: state_bytes.len() as u64,
573                delta_base: None,
574            }))
575        }
576        StateClosureEvent::Tree { hash, tree } => {
577            let tree_bytes = rmp_serde::to_vec_named(tree)?;
578            Ok(Some(ObjectInfo {
579                id: ObjectId::Hash(hash),
580                obj_type: ObjectType::Tree,
581                size: tree_bytes.len() as u64,
582                delta_base: None,
583            }))
584        }
585        StateClosureEvent::Blob { hash, .. } => {
586            let blob = store
587                .get_blob(&hash)?
588                .ok_or_else(|| ProtocolError::ObjectNotFound(hash.to_hex()))?;
589            Ok(Some(ObjectInfo {
590                id: ObjectId::Hash(hash),
591                obj_type: ObjectType::Blob,
592                size: blob.size() as u64,
593                delta_base: None,
594            }))
595        }
596        StateClosureEvent::Redaction { blob } => Ok(store
597            .get_redactions_bytes_for_blob(&blob)?
598            .map(|bytes| ObjectInfo {
599                id: ObjectId::Hash(blob),
600                obj_type: ObjectType::Redaction,
601                size: bytes.len() as u64,
602                delta_base: None,
603            })),
604        StateClosureEvent::StateVisibility { state } => Ok(store
605            .get_state_visibility_bytes_for_state(&state)?
606            .map(|bytes| ObjectInfo {
607                id: ObjectId::StateId(state),
608                obj_type: ObjectType::StateVisibility,
609                size: bytes.len() as u64,
610                delta_base: None,
611            })),
612        StateClosureEvent::StateAttachment { state, attachment } => {
613            let bytes = rmp_serde::to_vec_named(attachment)?;
614            Ok(Some(ObjectInfo {
615                id: ObjectId::StateAttachment {
616                    state,
617                    id: attachment.id(),
618                },
619                obj_type: ObjectType::StateAttachment,
620                size: bytes.len() as u64,
621                delta_base: None,
622            }))
623        }
624        StateClosureEvent::ExcludedState { id } => {
625            let _ = id;
626            Ok(None)
627        }
628        StateClosureEvent::ExcludedHash { hash } => {
629            let _ = hash;
630            Ok(None)
631        }
632    }
633}
634
635fn planned_object_from_event(
636    store: &impl ObjectStore,
637    event: StateClosureEvent<'_>,
638) -> Result<Option<PlannedObject>> {
639    match event {
640        StateClosureEvent::State { id, .. } => Ok(Some(PlannedObject {
641            id: ObjectId::StateId(id),
642            obj_type: ObjectType::State,
643        })),
644        StateClosureEvent::Tree { hash, .. } => Ok(Some(PlannedObject {
645            id: ObjectId::Hash(hash),
646            obj_type: ObjectType::Tree,
647        })),
648        StateClosureEvent::Blob { hash, source } => {
649            if source == BlobSource::StateMetadata && store.get_blob(&hash)?.is_none() {
650                return Err(ProtocolError::ObjectNotFound(hash.to_hex()));
651            }
652            Ok(Some(PlannedObject {
653                id: ObjectId::Hash(hash),
654                obj_type: ObjectType::Blob,
655            }))
656        }
657        StateClosureEvent::Redaction { blob } => Ok(Some(PlannedObject {
658            id: ObjectId::Hash(blob),
659            obj_type: ObjectType::Redaction,
660        })),
661        StateClosureEvent::StateVisibility { state } => Ok(Some(PlannedObject {
662            id: ObjectId::StateId(state),
663            obj_type: ObjectType::StateVisibility,
664        })),
665        StateClosureEvent::StateAttachment { state, attachment } => Ok(Some(PlannedObject {
666            id: ObjectId::StateAttachment {
667                state,
668                id: attachment.id(),
669            },
670            obj_type: ObjectType::StateAttachment,
671        })),
672        StateClosureEvent::ExcludedState { id } => {
673            let _ = id;
674            Ok(None)
675        }
676        StateClosureEvent::ExcludedHash { hash } => {
677            let _ = hash;
678            Ok(None)
679        }
680    }
681}
682
683pub fn missing_blobs_in_tree(
684    store: &impl ObjectStore,
685    tree_hash: ContentHash,
686) -> Result<Vec<ContentHash>> {
687    let mut missing = Vec::new();
688    collect_missing_blobs_recursive(store, &tree_hash, &mut missing)?;
689    Ok(missing)
690}
691
692fn collect_missing_blobs_recursive(
693    store: &impl ObjectStore,
694    tree_hash: &ContentHash,
695    missing: &mut Vec<ContentHash>,
696) -> Result<()> {
697    let Some(tree) = store.get_tree(tree_hash).map_err(|err| {
698        ProtocolError::InvalidState(format!(
699            "load tree {} while collecting lazy hydration missing blobs: {err}",
700            tree_hash.to_hex()
701        ))
702    })?
703    else {
704        return Ok(());
705    };
706
707    for entry in tree.entries() {
708        match entry.target() {
709            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
710                if !store.has_blob(hash).map_err(|err| {
711                    ProtocolError::InvalidState(format!(
712                        "check blob {} while collecting lazy hydration missing blobs: {err}",
713                        hash.to_hex()
714                    ))
715                })? {
716                    missing.push(*hash);
717                }
718            }
719            TreeEntryTarget::Tree { hash } => {
720                collect_missing_blobs_recursive(store, hash, missing)?;
721            }
722            TreeEntryTarget::Gitlink { .. } => {}
723            // Native child-spool edge: its target lives in a separate spool
724            // object graph, not this store, so it is not walked here.
725            TreeEntryTarget::Spoollink { .. } => {}
726        }
727    }
728    Ok(())
729}
730
731fn collect_excluded(
732    store: &impl ObjectStore,
733    roots: &[StateId],
734) -> Result<(HashSet<StateId>, HashSet<ContentHash>)> {
735    if roots.is_empty() {
736        return Ok((HashSet::new(), HashSet::new()));
737    }
738
739    let mut excluded_states: HashSet<StateId> = HashSet::new();
740    let mut excluded_hashes: HashSet<ContentHash> = HashSet::new();
741    let mut queue: VecDeque<StateId> = VecDeque::new();
742
743    for id in roots {
744        queue.push_back(*id);
745    }
746
747    while let Some(id) = queue.pop_front() {
748        if !excluded_states.insert(id) {
749            continue;
750        }
751
752        let state = match store.get_state(&id)? {
753            Some(state) => state,
754            None => continue,
755        };
756
757        for parent in &state.parents {
758            queue.push_back(*parent);
759        }
760
761        collect_tree_hashes(store, state.tree, &mut excluded_hashes)?;
762        if let Some(provenance_root) = state.provenance {
763            collect_tree_hashes(store, provenance_root, &mut excluded_hashes)?;
764        }
765        for attachment in store.list_state_attachments(&id)? {
766            match attachment.body {
767                StateAttachmentBody::Context(root) => {
768                    collect_tree_hashes(store, root, &mut excluded_hashes)?
769                }
770                StateAttachmentBody::RiskSignals(hash)
771                | StateAttachmentBody::ReviewSignatures(hash)
772                | StateAttachmentBody::Discussions(hash)
773                | StateAttachmentBody::StructuredConflicts(hash) => {
774                    excluded_hashes.insert(hash);
775                }
776                StateAttachmentBody::SemanticIndex(root) => {
777                    collect_semantic_hashes(store, root, &mut excluded_hashes)?;
778                }
779                StateAttachmentBody::Signature(_) => {}
780            }
781        }
782    }
783
784    Ok((excluded_states, excluded_hashes))
785}
786
787fn collect_tree_hashes(
788    store: &impl ObjectStore,
789    tree_hash: ContentHash,
790    excluded: &mut HashSet<ContentHash>,
791) -> Result<()> {
792    if !excluded.insert(tree_hash) {
793        return Ok(());
794    }
795
796    let tree = match store.get_tree(&tree_hash)? {
797        Some(tree) => tree,
798        None => return Ok(()),
799    };
800
801    for entry in tree.entries() {
802        match entry.target() {
803            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
804                excluded.insert(*hash);
805            }
806            TreeEntryTarget::Tree { hash } => {
807                collect_tree_hashes(store, *hash, excluded)?;
808            }
809            TreeEntryTarget::Gitlink { .. } => {}
810            // Native child-spool edge: its target lives in a separate spool
811            // object graph, not this store, so it is not walked here.
812            TreeEntryTarget::Spoollink { .. } => {}
813        }
814    }
815
816    Ok(())
817}
818
819pub fn is_ancestor(
820    store: &impl ObjectStore,
821    ancestor: StateId,
822    descendant: StateId,
823) -> Result<bool> {
824    if ancestor == descendant {
825        return Ok(true);
826    }
827
828    let mut seen: HashSet<StateId> = HashSet::new();
829    let mut queue: VecDeque<StateId> = VecDeque::new();
830    queue.push_back(descendant);
831
832    while let Some(id) = queue.pop_front() {
833        if !seen.insert(id) {
834            continue;
835        }
836        let state = match store.get_state(&id)? {
837            Some(s) => s,
838            None => return Ok(false),
839        };
840        for parent in state.parents {
841            if parent == ancestor {
842                return Ok(true);
843            }
844            queue.push_back(parent);
845        }
846    }
847
848    Ok(false)
849}
850
851#[cfg(test)]
852mod tests {
853    use std::{
854        collections::HashSet,
855        sync::atomic::{AtomicUsize, Ordering},
856    };
857
858    use chrono::Utc;
859    use objects::{
860        object::{
861            Action, ActionId, Attribution, Blob, ContentHash, Discussion, DiscussionResolution,
862            DiscussionTurn, DiscussionsBlob, Principal, Redaction, State, StateAttachment,
863            StateAttachmentBody, StateId, StateVisibility, SymbolAnchor, Tree, TreeEntry,
864            VisibilityTier,
865        },
866        store::{ObjectStore, Result as StoreResult},
867    };
868    use repo::Repository;
869    use sley::ObjectId as GitObjectId;
870    use tempfile::TempDir;
871
872    use super::{
873        ObjectId, ObjectInfo, ObjectType, PlannedObject, StateClosureOptions,
874        enumerate_state_closure_plan_with_options, enumerate_state_closure_transfer_with_options,
875        enumerate_state_closure_with_options, missing_blobs_in_tree,
876    };
877
878    fn pairs_from_full(objects: &[ObjectInfo]) -> HashSet<(ObjectId, ObjectType)> {
879        objects
880            .iter()
881            .map(|info| (info.id.clone(), info.obj_type))
882            .collect()
883    }
884
885    fn pairs_from_plan(objects: &[PlannedObject]) -> HashSet<(ObjectId, ObjectType)> {
886        objects
887            .iter()
888            .map(|info| (info.id.clone(), info.obj_type))
889            .collect()
890    }
891
892    fn object_info_fingerprint(
893        objects: &[ObjectInfo],
894    ) -> Vec<(ObjectId, ObjectType, u64, Option<ContentHash>)> {
895        objects
896            .iter()
897            .map(|info| (info.id.clone(), info.obj_type, info.size, info.delta_base))
898            .collect()
899    }
900
901    fn assert_plan_parity(
902        repo: &Repository,
903        state_id: StateId,
904        options: StateClosureOptions,
905    ) -> HashSet<(ObjectId, ObjectType)> {
906        let full =
907            enumerate_state_closure_with_options(repo.store(), state_id, options.clone()).unwrap();
908        let plan =
909            enumerate_state_closure_plan_with_options(repo.store(), state_id, options).unwrap();
910
911        let full_pairs = pairs_from_full(&full);
912        let plan_pairs = pairs_from_plan(&plan);
913        assert_eq!(full_pairs, plan_pairs);
914        full_pairs
915    }
916
917    fn assert_contains_object(
918        objects: &HashSet<(ObjectId, ObjectType)>,
919        id: ObjectId,
920        obj_type: ObjectType,
921    ) {
922        assert!(
923            objects.contains(&(id.clone(), obj_type)),
924            "expected closure to contain {id:?} as {obj_type:?}: {objects:?}"
925        );
926    }
927
928    struct CountingStore<'a, S> {
929        inner: &'a S,
930        state_reads: AtomicUsize,
931    }
932
933    impl<'a, S> CountingStore<'a, S> {
934        fn new(inner: &'a S) -> Self {
935            Self {
936                inner,
937                state_reads: AtomicUsize::new(0),
938            }
939        }
940
941        fn state_reads(&self) -> usize {
942            self.state_reads.load(Ordering::SeqCst)
943        }
944    }
945
946    impl<S: ObjectStore> ObjectStore for CountingStore<'_, S> {
947        fn get_blob(&self, hash: &ContentHash) -> StoreResult<Option<Blob>> {
948            self.inner.get_blob(hash)
949        }
950
951        fn put_blob(&self, blob: &Blob) -> StoreResult<ContentHash> {
952            self.inner.put_blob(blob)
953        }
954
955        fn has_blob(&self, hash: &ContentHash) -> StoreResult<bool> {
956            self.inner.has_blob(hash)
957        }
958
959        fn get_tree(&self, hash: &ContentHash) -> StoreResult<Option<Tree>> {
960            self.inner.get_tree(hash)
961        }
962
963        fn put_tree(&self, tree: &Tree) -> StoreResult<ContentHash> {
964            self.inner.put_tree(tree)
965        }
966
967        fn has_tree(&self, hash: &ContentHash) -> StoreResult<bool> {
968            self.inner.has_tree(hash)
969        }
970
971        fn get_state(&self, id: &StateId) -> StoreResult<Option<State>> {
972            self.state_reads.fetch_add(1, Ordering::SeqCst);
973            self.inner.get_state(id)
974        }
975
976        fn put_state(&self, state: &State) -> StoreResult<()> {
977            self.inner.put_state(state)
978        }
979
980        fn has_state(&self, id: &StateId) -> StoreResult<bool> {
981            self.inner.has_state(id)
982        }
983
984        fn list_states(&self) -> StoreResult<Vec<StateId>> {
985            self.inner.list_states()
986        }
987
988        fn get_action(&self, id: &ActionId) -> StoreResult<Option<Action>> {
989            self.inner.get_action(id)
990        }
991
992        fn put_action(&self, action: &mut Action) -> StoreResult<ActionId> {
993            self.inner.put_action(action)
994        }
995
996        fn list_actions(&self) -> StoreResult<Vec<ActionId>> {
997            self.inner.list_actions()
998        }
999
1000        fn list_blobs(&self) -> StoreResult<Vec<ContentHash>> {
1001            self.inner.list_blobs()
1002        }
1003
1004        fn list_trees(&self) -> StoreResult<Vec<ContentHash>> {
1005            self.inner.list_trees()
1006        }
1007    }
1008
1009    fn test_attribution() -> Attribution {
1010        Attribution::human(Principal::new("Graph Tester", "graph@example.com"))
1011    }
1012
1013    #[test]
1014    fn lean_closure_planner_matches_object_info_ids_and_types() {
1015        let temp = TempDir::new().unwrap();
1016        let repo = Repository::init_default(temp.path()).unwrap();
1017        std::fs::create_dir_all(temp.path().join("src")).unwrap();
1018        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1019        std::fs::write(temp.path().join("src/lib.rs"), "pub fn hi() {}\n").unwrap();
1020        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1021
1022        let full = enumerate_state_closure_with_options(
1023            repo.store(),
1024            state.state_id,
1025            StateClosureOptions::default(),
1026        )
1027        .unwrap();
1028        let lean = enumerate_state_closure_plan_with_options(
1029            repo.store(),
1030            state.state_id,
1031            StateClosureOptions::default(),
1032        )
1033        .unwrap();
1034
1035        let full_pairs = full
1036            .into_iter()
1037            .map(|info| (info.id, info.obj_type))
1038            .collect::<std::collections::HashSet<_>>();
1039        let lean_pairs = lean
1040            .into_iter()
1041            .map(|info| (info.id, info.obj_type))
1042            .collect::<std::collections::HashSet<_>>();
1043
1044        assert_eq!(full_pairs, lean_pairs);
1045        assert!(
1046            full_pairs
1047                .iter()
1048                .any(|(id, _)| matches!(id, ObjectId::StateId(_)))
1049        );
1050    }
1051
1052    #[test]
1053    fn transfer_projection_matches_full_and_plan_on_mixed_state_closure_fixture() {
1054        let temp = TempDir::new().unwrap();
1055        let repo = Repository::init_default(temp.path()).unwrap();
1056
1057        let excluded_blob = repo
1058            .store()
1059            .put_blob(&Blob::from("excluded"))
1060            .expect("put excluded blob");
1061        let excluded_tree_hash = repo
1062            .store()
1063            .put_tree(&Tree::from_entries(vec![
1064                TreeEntry::file("excluded.txt", excluded_blob, false).unwrap(),
1065            ]))
1066            .expect("put excluded tree");
1067        let excluded_parent = State::new(excluded_tree_hash, Vec::new(), test_attribution());
1068        repo.store()
1069            .put_state(&excluded_parent)
1070            .expect("put excluded parent");
1071
1072        let redacted_blob = repo
1073            .store()
1074            .put_blob(&Blob::from("secret"))
1075            .expect("put redacted blob");
1076        let nested_blob = repo
1077            .store()
1078            .put_blob(&Blob::from("nested"))
1079            .expect("put nested blob");
1080        let symlink_blob = repo
1081            .store()
1082            .put_blob(&Blob::from("target"))
1083            .expect("put symlink blob");
1084        let context_blob = repo
1085            .store()
1086            .put_blob(&Blob::from("context"))
1087            .expect("put context blob");
1088        let provenance_blob = repo
1089            .store()
1090            .put_blob(&Blob::from("provenance"))
1091            .expect("put provenance blob");
1092        let risk_blob = repo
1093            .store()
1094            .put_blob(&Blob::from("risk"))
1095            .expect("put risk blob");
1096        let review_blob = repo
1097            .store()
1098            .put_blob(&Blob::from("review"))
1099            .expect("put review blob");
1100        let discussions_blob = repo
1101            .store()
1102            .put_blob(&Blob::from("discussion"))
1103            .expect("put discussion blob");
1104        let conflicts_blob = repo
1105            .store()
1106            .put_blob(&Blob::from("conflicts"))
1107            .expect("put conflicts blob");
1108
1109        let nested_tree_hash = repo
1110            .store()
1111            .put_tree(&Tree::from_entries(vec![
1112                TreeEntry::file("nested.txt", nested_blob, false).unwrap(),
1113                TreeEntry::symlink("latest", symlink_blob).unwrap(),
1114            ]))
1115            .expect("put nested tree");
1116        let context_tree_hash = repo
1117            .store()
1118            .put_tree(&Tree::from_entries(vec![
1119                TreeEntry::file("context.txt", context_blob, false).unwrap(),
1120            ]))
1121            .expect("put context tree");
1122        let provenance_tree_hash = repo
1123            .store()
1124            .put_tree(&Tree::from_entries(vec![
1125                TreeEntry::file("lineage.txt", provenance_blob, false).unwrap(),
1126            ]))
1127            .expect("put provenance tree");
1128        let gitlink_target: GitObjectId = "0303030303030303030303030303030303030303"
1129            .parse()
1130            .expect("git oid");
1131        let root_tree_hash = repo
1132            .store()
1133            .put_tree(&Tree::from_entries(vec![
1134                TreeEntry::file("secret.txt", redacted_blob, false).unwrap(),
1135                TreeEntry::directory("nested", nested_tree_hash).unwrap(),
1136                TreeEntry::gitlink("vendor", gitlink_target).unwrap(),
1137            ]))
1138            .expect("put root tree");
1139        let state = State::new(
1140            root_tree_hash,
1141            vec![excluded_parent.state_id],
1142            test_attribution(),
1143        )
1144        .with_provenance(provenance_tree_hash);
1145        repo.store().put_state(&state).expect("put state");
1146        for body in [
1147            StateAttachmentBody::Context(context_tree_hash),
1148            StateAttachmentBody::RiskSignals(risk_blob),
1149            StateAttachmentBody::ReviewSignatures(review_blob),
1150            StateAttachmentBody::Discussions(discussions_blob),
1151            StateAttachmentBody::StructuredConflicts(conflicts_blob),
1152        ] {
1153            repo.put_state_attachment(&StateAttachment {
1154                state_id: state.id(),
1155                body,
1156                attribution: state.attribution.clone(),
1157                created_at: Utc::now(),
1158                supersedes: None,
1159            })
1160            .unwrap();
1161        }
1162
1163        repo.put_redaction(Redaction {
1164            redacted_blob,
1165            state: state.state_id,
1166            path: "secret.txt".to_string(),
1167            reason: "test leak".to_string(),
1168            redactor: Principal::new("Tester", "tester@example.test"),
1169            redacted_at: Utc::now(),
1170            signature: None,
1171            purged_at: None,
1172            supersedes: None,
1173        })
1174        .expect("put redaction");
1175        repo.put_state_visibility(StateVisibility {
1176            state: state.state_id,
1177            tier: VisibilityTier::Restricted {
1178                scope_label: "security".to_string(),
1179            },
1180            embargo_until: None,
1181            declarer: Principal::new("Tester", "tester@example.test"),
1182            declared_at: Utc::now(),
1183            signature: None,
1184            supersedes: None,
1185        })
1186        .expect("put visibility");
1187
1188        let options = StateClosureOptions {
1189            depth: None,
1190            exclude_states: vec![excluded_parent.state_id],
1191        };
1192        let transfer = enumerate_state_closure_transfer_with_options(
1193            repo.store(),
1194            state.state_id,
1195            options.clone(),
1196            512,
1197        )
1198        .expect("transfer projection");
1199
1200        let full =
1201            enumerate_state_closure_with_options(repo.store(), state.state_id, options.clone())
1202                .expect("full closure");
1203        let plan = enumerate_state_closure_plan_with_options(repo.store(), state.state_id, options)
1204            .expect("plan closure");
1205        assert_eq!(
1206            transfer
1207                .full_objects
1208                .as_deref()
1209                .map(object_info_fingerprint),
1210            Some(object_info_fingerprint(&full))
1211        );
1212        assert_eq!(transfer.planned_objects, plan);
1213
1214        let full_pairs = pairs_from_full(&full);
1215        assert_eq!(full_pairs, pairs_from_plan(&plan));
1216        assert_contains_object(
1217            &full_pairs,
1218            ObjectId::StateId(state.state_id),
1219            ObjectType::State,
1220        );
1221        assert_contains_object(
1222            &full_pairs,
1223            ObjectId::StateId(state.state_id),
1224            ObjectType::StateVisibility,
1225        );
1226        assert_contains_object(&full_pairs, ObjectId::Hash(redacted_blob), ObjectType::Blob);
1227        assert_contains_object(
1228            &full_pairs,
1229            ObjectId::Hash(redacted_blob),
1230            ObjectType::Redaction,
1231        );
1232        for hash in [
1233            root_tree_hash,
1234            nested_tree_hash,
1235            context_tree_hash,
1236            provenance_tree_hash,
1237        ] {
1238            assert_contains_object(&full_pairs, ObjectId::Hash(hash), ObjectType::Tree);
1239        }
1240        for hash in [
1241            nested_blob,
1242            symlink_blob,
1243            context_blob,
1244            provenance_blob,
1245            risk_blob,
1246            review_blob,
1247            discussions_blob,
1248            conflicts_blob,
1249        ] {
1250            assert_contains_object(&full_pairs, ObjectId::Hash(hash), ObjectType::Blob);
1251        }
1252        assert!(!full_pairs.contains(&(
1253            ObjectId::StateId(excluded_parent.state_id),
1254            ObjectType::State
1255        )));
1256        assert!(!full_pairs.contains(&(ObjectId::Hash(excluded_tree_hash), ObjectType::Tree)));
1257        assert!(!full_pairs.contains(&(ObjectId::Hash(excluded_blob), ObjectType::Blob)));
1258    }
1259
1260    #[test]
1261    fn transfer_projection_reads_root_state_once_on_small_transfer() {
1262        let temp = TempDir::new().unwrap();
1263        let repo = Repository::init_default(temp.path()).unwrap();
1264        let blob = repo
1265            .store()
1266            .put_blob(&Blob::from("hello\n"))
1267            .expect("put blob");
1268        let tree_hash = repo
1269            .store()
1270            .put_tree(&Tree::from_entries(vec![
1271                TreeEntry::file("README.md", blob, false).unwrap(),
1272            ]))
1273            .expect("put tree");
1274        let state = State::new(tree_hash, Vec::new(), test_attribution());
1275        repo.store().put_state(&state).expect("put state");
1276        let store = CountingStore::new(repo.store());
1277
1278        let transfer = enumerate_state_closure_transfer_with_options(
1279            &store,
1280            state.state_id,
1281            StateClosureOptions::default(),
1282            512,
1283        )
1284        .expect("transfer projection");
1285
1286        assert!(
1287            !transfer.planned_objects.is_empty(),
1288            "lean projection should be available"
1289        );
1290        assert!(transfer.full_objects.is_some());
1291        assert_eq!(
1292            store.state_reads(),
1293            1,
1294            "small transfer projection must not read the root state through a second closure walk"
1295        );
1296    }
1297
1298    #[test]
1299    fn transfer_projection_drops_full_descriptors_after_threshold() {
1300        let temp = TempDir::new().unwrap();
1301        let repo = Repository::init_default(temp.path()).unwrap();
1302        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1303        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1304
1305        let transfer = enumerate_state_closure_transfer_with_options(
1306            repo.store(),
1307            state.state_id,
1308            StateClosureOptions::default(),
1309            0,
1310        )
1311        .expect("transfer projection");
1312
1313        assert!(
1314            !transfer.planned_objects.is_empty(),
1315            "lean projection should still be available over the threshold"
1316        );
1317        assert!(transfer.full_objects.is_none());
1318    }
1319
1320    #[test]
1321    fn depth_and_exclude_options_match_between_full_and_plan() {
1322        let temp = TempDir::new().unwrap();
1323        let repo = Repository::init_default(temp.path()).unwrap();
1324        let path = temp.path().join("story.txt");
1325
1326        std::fs::write(&path, "base\n").unwrap();
1327        let base = repo.snapshot(Some("base".to_string()), None).unwrap();
1328        std::fs::write(&path, "middle\n").unwrap();
1329        let middle = repo.snapshot(Some("middle".to_string()), None).unwrap();
1330        std::fs::write(&path, "tip\n").unwrap();
1331        let tip = repo.snapshot(Some("tip".to_string()), None).unwrap();
1332
1333        let depth_zero = assert_plan_parity(
1334            &repo,
1335            tip.state_id,
1336            StateClosureOptions {
1337                depth: Some(0),
1338                exclude_states: Vec::new(),
1339            },
1340        );
1341        assert!(depth_zero.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
1342        assert!(!depth_zero.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
1343        assert!(!depth_zero.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
1344
1345        let depth_one = assert_plan_parity(
1346            &repo,
1347            tip.state_id,
1348            StateClosureOptions {
1349                depth: Some(1),
1350                exclude_states: Vec::new(),
1351            },
1352        );
1353        assert!(depth_one.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
1354        assert!(depth_one.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
1355        assert!(!depth_one.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
1356
1357        let exclude_middle = assert_plan_parity(
1358            &repo,
1359            tip.state_id,
1360            StateClosureOptions {
1361                depth: None,
1362                exclude_states: vec![middle.state_id],
1363            },
1364        );
1365        assert!(exclude_middle.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
1366        assert!(!exclude_middle.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
1367        assert!(!exclude_middle.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
1368    }
1369
1370    #[test]
1371    fn shared_tree_and_blob_references_are_emitted_once() {
1372        let temp = TempDir::new().unwrap();
1373        let repo = Repository::init_default(temp.path()).unwrap();
1374
1375        let shared_blob = Blob::from("shared contents\n");
1376        let shared_blob_hash = repo.store().put_blob(&shared_blob).unwrap();
1377        let shared_tree = Tree::from_entries(vec![
1378            TreeEntry::file("shared.txt", shared_blob_hash, false).unwrap(),
1379        ]);
1380        let shared_tree_hash = repo.store().put_tree(&shared_tree).unwrap();
1381        let root = Tree::from_entries(vec![
1382            TreeEntry::directory("left", shared_tree_hash).unwrap(),
1383            TreeEntry::directory("right", shared_tree_hash).unwrap(),
1384        ]);
1385        let root_hash = repo.store().put_tree(&root).unwrap();
1386        let state = State::new(root_hash, Vec::new(), test_attribution());
1387        repo.store().put_state(&state).unwrap();
1388
1389        let full = enumerate_state_closure_with_options(
1390            repo.store(),
1391            state.state_id,
1392            StateClosureOptions::default(),
1393        )
1394        .unwrap();
1395        let plan = enumerate_state_closure_plan_with_options(
1396            repo.store(),
1397            state.state_id,
1398            StateClosureOptions::default(),
1399        )
1400        .unwrap();
1401
1402        assert_eq!(
1403            pairs_from_full(&full),
1404            pairs_from_plan(&plan),
1405            "full and lean closure enumerators must dedup the same objects"
1406        );
1407
1408        assert_eq!(
1409            full.iter()
1410                .filter(|info| info.id == ObjectId::Hash(root_hash)
1411                    && info.obj_type == ObjectType::Tree)
1412                .count(),
1413            1
1414        );
1415        assert_eq!(
1416            full.iter()
1417                .filter(|info| info.id == ObjectId::Hash(shared_tree_hash)
1418                    && info.obj_type == ObjectType::Tree)
1419                .count(),
1420            1
1421        );
1422        assert_eq!(
1423            full.iter()
1424                .filter(|info| info.id == ObjectId::Hash(shared_blob_hash)
1425                    && info.obj_type == ObjectType::Blob)
1426                .count(),
1427            1
1428        );
1429    }
1430
1431    #[test]
1432    fn state_closure_skips_gitlink_targets() {
1433        let temp = TempDir::new().unwrap();
1434        let repo = Repository::init_default(temp.path()).unwrap();
1435        let target: GitObjectId = "0303030303030303030303030303030303030303"
1436            .parse()
1437            .expect("git oid");
1438        let root = Tree::from_entries(vec![
1439            TreeEntry::gitlink("vendor", target).expect("gitlink entry"),
1440        ]);
1441        let root_hash = repo.store().put_tree(&root).unwrap();
1442        let state = State::new(root_hash, Vec::new(), test_attribution());
1443        repo.store().put_state(&state).unwrap();
1444
1445        let full = enumerate_state_closure_with_options(
1446            repo.store(),
1447            state.state_id,
1448            StateClosureOptions::default(),
1449        )
1450        .unwrap();
1451        let plan = enumerate_state_closure_plan_with_options(
1452            repo.store(),
1453            state.state_id,
1454            StateClosureOptions::default(),
1455        )
1456        .unwrap();
1457
1458        assert_eq!(pairs_from_full(&full), pairs_from_plan(&plan));
1459        assert!(
1460            !full.iter().any(|info| info.obj_type == ObjectType::Blob),
1461            "gitlinks carry foreign Git commit ids, not Heddle blob dependencies: {full:?}"
1462        );
1463        assert!(full.iter().any(|info| {
1464            info.id == ObjectId::Hash(root_hash) && info.obj_type == ObjectType::Tree
1465        }));
1466    }
1467
1468    #[test]
1469    fn missing_blobs_in_tree_skips_gitlinks_and_walks_nested_side_paths() {
1470        let temp = TempDir::new().unwrap();
1471        let repo = Repository::init_default(temp.path()).unwrap();
1472        let present_blob = repo
1473            .store()
1474            .put_blob(&Blob::from("already local"))
1475            .expect("put present blob");
1476        let missing_nested = ContentHash::from_bytes([7; 32]);
1477        let missing_symlink = ContentHash::from_bytes([8; 32]);
1478        let nested_tree = Tree::from_entries(vec![
1479            TreeEntry::file("remote.txt", missing_nested, false).unwrap(),
1480            TreeEntry::symlink("remote-link", missing_symlink).unwrap(),
1481        ]);
1482        let nested_tree_hash = repo
1483            .store()
1484            .put_tree(&nested_tree)
1485            .expect("put nested tree");
1486        let gitlink_target: GitObjectId = "0404040404040404040404040404040404040404"
1487            .parse()
1488            .expect("git oid");
1489        let root = Tree::from_entries(vec![
1490            TreeEntry::file("local.txt", present_blob, false).unwrap(),
1491            TreeEntry::directory("nested", nested_tree_hash).unwrap(),
1492            TreeEntry::gitlink("vendor", gitlink_target).unwrap(),
1493        ]);
1494        let root_hash = repo.store().put_tree(&root).expect("put root tree");
1495
1496        let missing = missing_blobs_in_tree(repo.store(), root_hash).expect("missing blobs");
1497
1498        assert_eq!(
1499            missing.into_iter().collect::<HashSet<_>>(),
1500            HashSet::from([missing_nested, missing_symlink])
1501        );
1502    }
1503
1504    /// Once a redaction is declared for a blob in a snapshot, the
1505    /// state closure must include an `ObjectType::Redaction` entry
1506    /// keyed on that blob's hash — that's the wire-side signal the
1507    /// receiver replays.
1508    #[test]
1509    fn enumerate_state_closure_emits_redaction_for_redacted_blob() {
1510        let temp = TempDir::new().unwrap();
1511        let repo = Repository::init_default(temp.path()).unwrap();
1512        std::fs::write(temp.path().join("secret.toml"), "api_token = \"x\"\n").unwrap();
1513        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1514
1515        // Find the blob hash for secret.toml by walking the snapshot's tree.
1516        let tree = repo
1517            .store()
1518            .get_tree(&state.tree)
1519            .unwrap()
1520            .expect("tree present");
1521        let blob_hash = tree
1522            .iter()
1523            .find(|e| e.name() == "secret.toml")
1524            .expect("entry present")
1525            .blob_hash()
1526            .expect("secret.toml is a blob");
1527
1528        let redaction = Redaction {
1529            redacted_blob: blob_hash,
1530            state: state.state_id,
1531            path: "secret.toml".to_string(),
1532            reason: "test leak".to_string(),
1533            redactor: Principal {
1534                name: "Tester".into(),
1535                email: "tester@heddle.sh".into(),
1536            },
1537            redacted_at: Utc::now(),
1538            signature: None,
1539            purged_at: None,
1540            supersedes: None,
1541        };
1542        repo.put_redaction(redaction).unwrap();
1543
1544        let full = enumerate_state_closure_with_options(
1545            repo.store(),
1546            state.state_id,
1547            StateClosureOptions::default(),
1548        )
1549        .unwrap();
1550        let plan = enumerate_state_closure_plan_with_options(
1551            repo.store(),
1552            state.state_id,
1553            StateClosureOptions::default(),
1554        )
1555        .unwrap();
1556
1557        assert!(
1558            full.iter()
1559                .any(|info| info.obj_type == ObjectType::Redaction
1560                    && info.id == ObjectId::Hash(blob_hash)),
1561            "full closure must include a Redaction entry for the redacted blob"
1562        );
1563        assert!(
1564            plan.iter()
1565                .any(|p| p.obj_type == ObjectType::Redaction && p.id == ObjectId::Hash(blob_hash)),
1566            "plan closure must include a Redaction entry for the redacted blob"
1567        );
1568    }
1569
1570    #[test]
1571    fn enumerate_state_closure_emits_state_visibility_for_visible_state() {
1572        let temp = TempDir::new().unwrap();
1573        let repo = Repository::init_default(temp.path()).unwrap();
1574        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1575        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1576
1577        repo.put_state_visibility(StateVisibility {
1578            state: state.state_id,
1579            tier: VisibilityTier::Restricted {
1580                scope_label: "security-embargo".into(),
1581            },
1582            embargo_until: None,
1583            declarer: Principal {
1584                name: "Tester".into(),
1585                email: "tester@heddle.sh".into(),
1586            },
1587            declared_at: Utc::now(),
1588            signature: None,
1589            supersedes: None,
1590        })
1591        .unwrap();
1592
1593        let full = enumerate_state_closure_with_options(
1594            repo.store(),
1595            state.state_id,
1596            StateClosureOptions::default(),
1597        )
1598        .unwrap();
1599        let plan = enumerate_state_closure_plan_with_options(
1600            repo.store(),
1601            state.state_id,
1602            StateClosureOptions::default(),
1603        )
1604        .unwrap();
1605
1606        assert!(
1607            full.iter()
1608                .any(|info| info.obj_type == ObjectType::StateVisibility
1609                    && info.id == ObjectId::StateId(state.state_id)),
1610            "full closure must include a StateVisibility entry for the visible state"
1611        );
1612        assert!(
1613            plan.iter()
1614                .any(|p| p.obj_type == ObjectType::StateVisibility
1615                    && p.id == ObjectId::StateId(state.state_id)),
1616            "plan closure must include a StateVisibility entry for the visible state"
1617        );
1618    }
1619
1620    #[test]
1621    fn enumerate_state_closure_emits_state_metadata_blobs() {
1622        let temp = TempDir::new().unwrap();
1623        let repo = Repository::init_default(temp.path()).unwrap();
1624        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1625        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1626
1627        let principal = Principal::new("Tester", "tester@example.test");
1628        let discussion_bytes = DiscussionsBlob::new(vec![Discussion {
1629            id: "disc-1".to_string(),
1630            anchor: SymbolAnchor::new("src/lib.rs", "answer"),
1631            opened_against_state: state.state_id,
1632            opened_at: 1_782_400_000,
1633            thread_ref: None,
1634            turns: vec![DiscussionTurn {
1635                author: principal,
1636                body: "Should this sync?".to_string(),
1637                posted_at: 1_782_400_000,
1638            }],
1639            resolution: DiscussionResolution::Open,
1640            body_changed_since_open: false,
1641            orphaned: false,
1642            visibility: VisibilityTier::default(),
1643            resolved_annotation_id: None,
1644        }])
1645        .encode()
1646        .expect("encode discussions");
1647        let discussion_hash = repo
1648            .store()
1649            .put_blob(&Blob::new(discussion_bytes))
1650            .expect("put discussions blob");
1651        let risk_hash = repo
1652            .store()
1653            .put_blob(&Blob::from_slice(b"risk signals"))
1654            .expect("put risk blob");
1655        let review_hash = repo
1656            .store()
1657            .put_blob(&Blob::from_slice(b"review signatures"))
1658            .expect("put review blob");
1659        let conflicts_hash = repo
1660            .store()
1661            .put_blob(&Blob::from_slice(b"structured conflicts"))
1662            .expect("put conflicts blob");
1663        for body in [
1664            StateAttachmentBody::RiskSignals(risk_hash),
1665            StateAttachmentBody::ReviewSignatures(review_hash),
1666            StateAttachmentBody::Discussions(discussion_hash),
1667            StateAttachmentBody::StructuredConflicts(conflicts_hash),
1668        ] {
1669            repo.put_state_attachment(&StateAttachment {
1670                state_id: state.id(),
1671                body,
1672                attribution: state.attribution.clone(),
1673                created_at: Utc::now(),
1674                supersedes: None,
1675            })
1676            .unwrap();
1677        }
1678
1679        let full = enumerate_state_closure_with_options(
1680            repo.store(),
1681            state.state_id,
1682            StateClosureOptions::default(),
1683        )
1684        .unwrap();
1685        let plan = enumerate_state_closure_plan_with_options(
1686            repo.store(),
1687            state.state_id,
1688            StateClosureOptions::default(),
1689        )
1690        .unwrap();
1691
1692        for metadata_hash in [risk_hash, review_hash, discussion_hash, conflicts_hash] {
1693            assert!(
1694                full.iter().any(|info| info.obj_type == ObjectType::Blob
1695                    && info.id == ObjectId::Hash(metadata_hash)),
1696                "full closure must include state metadata blob {metadata_hash}"
1697            );
1698            assert!(
1699                plan.iter().any(
1700                    |p| p.obj_type == ObjectType::Blob && p.id == ObjectId::Hash(metadata_hash)
1701                ),
1702                "plan closure must include state metadata blob {metadata_hash}"
1703            );
1704        }
1705    }
1706}