heddle_thread_api/fetch/
staging.rs

1//! Source bytes remain temporary until the terminal receipt and exact closure
2//! have been checked. Staging never changes a repository or a checkout.
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    path::Path,
6};
7
8use api::v2::client::MessageReader;
9use crypto::thread_operation::SignedOperation;
10use heddle_object_model::object::{ContentHash, State, thread_replication::ThreadOperation};
11use heddle_pack::store::pack::PackReader;
12use prost::Message;
13use tokio::io::AsyncWriteExt;
14
15use super::{Download, Error, Item};
16use crate::{contract::*, transport};
17
18const METADATA_BYTES: usize = 16 * 1024 * 1024;
19const SOURCE_BYTES: u64 = 256 * 1024 * 1024;
20const SOURCE_OBJECTS: usize = 100_000;
21
22/// Verified source artifacts and their original proofs. Dropping this value
23/// removes its temporary files. Native callers can install it on a disk worker.
24pub struct StagedSource {
25    pub(super) directory: tempfile::TempDir,
26    pub(super) ready: TransferReady,
27    pub(super) operations: Vec<SignedOperation>,
28    pub(super) dependencies: Vec<ThreadGenesisRecord>,
29    pub(super) state: State,
30    pub(super) partial_trees: Vec<heddle_object_model::object::PartialTree>,
31    pub(super) authority_admissions:
32        BTreeMap<ContentHash, crypto::thread_authority_admission::SignedAuthorityAdmission>,
33}
34impl StagedSource {
35    pub fn artifact_paths(&self) -> [std::path::PathBuf; 2] {
36        [
37            self.directory.path().join("source.pack"),
38            self.directory.path().join("source.idx"),
39        ]
40    }
41    pub fn operations(&self) -> &[SignedOperation] {
42        &self.operations
43    }
44    pub fn dependency_geneses(&self) -> &[ThreadGenesisRecord] {
45        &self.dependencies
46    }
47    pub fn ready(&self) -> &TransferReady {
48        &self.ready
49    }
50    pub fn state(&self) -> &State {
51        &self.state
52    }
53    /// Partial source remains read-only until a verified full closure arrives.
54    pub fn is_complete(&self) -> bool {
55        self.ready.full_closure_available
56    }
57}
58impl<R: MessageReader<Error = transport::Error>> Download<R> {
59    /// Consume one complete source download into bounded temporary files and
60    /// validate its original causal ancestry and exact selected source closure.
61    pub async fn stage(mut self, scratch: &Path) -> Result<StagedSource, Error> {
62        if self.state.facets != [SharedFacet::Source as i32] {
63            return Err(Error::Invalid("staging requires the source facet alone"));
64        }
65        let total = self
66            .state
67            .ready
68            .packs
69            .iter()
70            .try_fold(0u64, |sum, extent| sum.checked_add(extent.length))
71            .ok_or(Error::Invalid("source artifact length overflow"))?;
72        if total > SOURCE_BYTES {
73            return Err(Error::Invalid("staged source exceeds 256 MiB"));
74        }
75        self.state.limits.max_operations = self.state.limits.max_operations.min(10_000);
76        let directory = tempfile::Builder::new()
77            .prefix("thread-download-")
78            .tempdir_in(scratch)?;
79        let mut files = [
80            tokio::fs::File::create(directory.path().join("source.pack")).await?,
81            tokio::fs::File::create(directory.path().join("source.idx")).await?,
82        ];
83        let mut operations = Vec::new();
84        let mut receipt_records = Vec::new();
85        let mut dependencies = Vec::new();
86        let mut metadata_bytes = 0usize;
87        let mut complete = false;
88        while let Some(item) = self.next().await? {
89            match item {
90                Item::Pack(chunk) => {
91                    let kind = chunk
92                        .extent
93                        .as_ref()
94                        .ok_or(Error::Invalid("chunk extent absent"))?
95                        .kind;
96                    let index = match pack_extent::Kind::try_from(kind) {
97                        Ok(pack_extent::Kind::NativePack) => 0,
98                        Ok(pack_extent::Kind::NativeIndex) => 1,
99                        _ => return Err(Error::Invalid("native source artifacts required")),
100                    };
101                    files[index].write_all(&chunk.data).await?;
102                }
103                Item::Operations(batch) => {
104                    metadata_bytes = metadata_bytes
105                        .checked_add(batch.encoded_len())
106                        .ok_or(Error::Invalid("source metadata length overflow"))?;
107                    if metadata_bytes > METADATA_BYTES {
108                        return Err(Error::Invalid("staged source metadata exceeds 16 MiB"));
109                    }
110                    for received in crate::authority_admission::match_batch(&batch)? {
111                        operations.push(received.original);
112                        receipt_records.extend(received.authority_admission);
113                    }
114                }
115                Item::ThreadGenesis(record) => {
116                    metadata_bytes = metadata_bytes
117                        .checked_add(record.encoded_len())
118                        .ok_or(Error::Invalid("source metadata length overflow"))?;
119                    if metadata_bytes > METADATA_BYTES || dependencies.len() >= 127 {
120                        return Err(Error::Invalid("dependency metadata exceeds bounds"));
121                    }
122                    dependencies.push(record);
123                }
124                Item::Complete(_) => complete = true,
125                Item::Sidecar(_) => return Err(Error::Invalid("source staging excludes sidecars")),
126            }
127        }
128        if !complete {
129            return Err(Error::Invalid("source staging requires Complete"));
130        }
131        for file in &mut files {
132            file.flush().await?;
133            file.sync_all().await?;
134        }
135        drop(files);
136        let ready = self.state.ready;
137        tokio::task::spawn_blocking(move || {
138            validate_with_receipts(directory, ready, operations, dependencies, receipt_records)
139        })
140        .await
141        .map_err(|error| Error::Preparation(error.to_string()))?
142    }
143}
144#[cfg(test)]
145fn validate(
146    directory: tempfile::TempDir,
147    ready: TransferReady,
148    operations: Vec<SignedOperation>,
149    dependencies: Vec<ThreadGenesisRecord>,
150) -> Result<StagedSource, Error> {
151    validate_with_receipts(directory, ready, operations, dependencies, Vec::new())
152}
153
154struct DisclosureInput {
155    directory: tempfile::TempDir,
156    operations: Vec<SignedOperation>,
157    dependency_records: Vec<ThreadGenesisRecord>,
158    receipt_records: Vec<crypto::thread_authority_admission::SignedAuthorityAdmission>,
159    allow_partial: bool,
160}
161
162pub(super) fn validate_with_receipts(
163    directory: tempfile::TempDir,
164    ready: TransferReady,
165    operations: Vec<SignedOperation>,
166    dependencies: Vec<ThreadGenesisRecord>,
167    receipt_records: Vec<crypto::thread_authority_admission::SignedAuthorityAdmission>,
168) -> Result<StagedSource, Error> {
169    let value = validate_disclosure_artifacts(
170        ready
171            .thread
172            .as_ref()
173            .ok_or(Error::Invalid("Thread absent"))?,
174        ready
175            .current
176            .as_ref()
177            .ok_or(Error::Invalid("revision absent"))?,
178        ready
179            .thread_genesis
180            .as_ref()
181            .ok_or(Error::Invalid("original genesis absent"))?,
182        DisclosureInput {
183            directory,
184            operations,
185            dependency_records: dependencies,
186            receipt_records,
187            allow_partial: !ready.full_closure_available,
188        },
189    )?;
190    Ok(StagedSource {
191        directory: value.directory,
192        ready,
193        operations: value.operations,
194        dependencies: value.dependencies,
195        state: value.state,
196        partial_trees: value.partial_trees,
197        authority_admissions: value.authority_admissions,
198    })
199}
200/// Structurally verified original source and actual artifact closure. This is
201/// not an author, audience, executor, or sharing-policy admission decision.
202pub struct ValidatedSourceArtifacts {
203    directory: tempfile::TempDir,
204    operations: Vec<SignedOperation>,
205    genesis: ThreadGenesisRecord,
206    dependencies: Vec<ThreadGenesisRecord>,
207    state: State,
208    partial_trees: Vec<heddle_object_model::object::PartialTree>,
209    authority_admissions:
210        BTreeMap<ContentHash, crypto::thread_authority_admission::SignedAuthorityAdmission>,
211}
212impl ValidatedSourceArtifacts {
213    pub fn artifact_paths(&self) -> [std::path::PathBuf; 2] {
214        [
215            self.directory.path().join("source.pack"),
216            self.directory.path().join("source.idx"),
217        ]
218    }
219    pub fn operations(&self) -> &[SignedOperation] {
220        &self.operations
221    }
222    pub fn geneses(&self) -> impl Iterator<Item = &ThreadGenesisRecord> {
223        std::iter::once(&self.genesis).chain(&self.dependencies)
224    }
225    pub fn state(&self) -> &State {
226        &self.state
227    }
228    pub fn authority_admissions(
229        &self,
230    ) -> &BTreeMap<ContentHash, crypto::thread_authority_admission::SignedAuthorityAdmission> {
231        &self.authority_admissions
232    }
233}
234pub(crate) fn validate_artifacts(
235    directory: tempfile::TempDir,
236    thread: &ThreadRef,
237    revision: &RevisionRef,
238    original: &ThreadGenesisRecord,
239    operations: Vec<SignedOperation>,
240    dependency_records: Vec<ThreadGenesisRecord>,
241    receipt_records: Vec<crypto::thread_authority_admission::SignedAuthorityAdmission>,
242) -> Result<ValidatedSourceArtifacts, Error> {
243    validate_disclosure_artifacts(
244        thread,
245        revision,
246        original,
247        DisclosureInput {
248            directory,
249            operations,
250            dependency_records,
251            receipt_records,
252            allow_partial: false,
253        },
254    )
255}
256
257fn validate_disclosure_artifacts(
258    thread: &ThreadRef,
259    revision: &RevisionRef,
260    original: &ThreadGenesisRecord,
261    input: DisclosureInput,
262) -> Result<ValidatedSourceArtifacts, Error> {
263    let DisclosureInput {
264        directory,
265        operations,
266        dependency_records,
267        receipt_records,
268        allow_partial,
269    } = input;
270    if operations.len() > 10_000
271        || dependency_records.len() >= 128
272        || receipt_records.len() > operations.len()
273    {
274        return Err(Error::Invalid("source original graph exceeds bounds"));
275    }
276    let mut metadata = original.encoded_len();
277    for record in &dependency_records {
278        metadata = metadata.saturating_add(record.encoded_len());
279    }
280    for operation in &operations {
281        metadata = metadata.saturating_add(operation.canonical.len() + operation.signature.len());
282    }
283    for receipt in &receipt_records {
284        metadata = metadata.saturating_add(receipt.canonical.len() + receipt.signature.len());
285    }
286    let mut evidence_ids = BTreeSet::new();
287    for wrapper in std::iter::once(original).chain(&dependency_records) {
288        for record in &wrapper.boundary_acceptances {
289            evidence_ids.insert(heddle_object_model::object::ContentHash::compute_typed(
290                heddle_object_model::object::original_boundary_acceptance::FORMAT,
291                &record.canonical_record,
292            ));
293            if evidence_ids.len() > crate::boundary_acceptance::MAX_ACCEPTANCES {
294                return Err(Error::Invalid("boundary evidence count exceeded"));
295            }
296        }
297    }
298    for receipt in &receipt_records {
299        if let Some(evidence) = &receipt.boundary_acceptance {
300            if evidence_ids.insert(
301                evidence
302                    .verify_signature()
303                    .map_err(preparation)?
304                    .id()
305                    .map_err(preparation)?,
306            ) {
307                metadata =
308                    metadata.saturating_add(evidence.canonical.len() + evidence.signature.len());
309            }
310            if evidence_ids.len() > crate::boundary_acceptance::MAX_ACCEPTANCES {
311                return Err(Error::Invalid("boundary evidence count exceeded"));
312            }
313        }
314    }
315    if metadata > METADATA_BYTES {
316        return Err(Error::Invalid("source metadata exceeds 16 MiB"));
317    }
318    if revision.spool != thread.spool {
319        return Err(Error::Invalid("source revision crosses Spool"));
320    }
321    let genesis = super::verify_origin(original, thread)?;
322    let Some(revision_ref::Revision::State(selected)) = revision.revision.as_ref() else {
323        return Err(Error::Invalid("exact native State required"));
324    };
325    let selected_thread = genesis.id().map_err(preparation)?;
326    if operations.is_empty() {
327        if !dependency_records.is_empty() || !receipt_records.is_empty() {
328            return Err(Error::Invalid(
329                "initial source cannot carry dependency originals",
330            ));
331        }
332        let state =
333            heddle_object_model::object::thread_replication::hosted_import::synthetic_initial_base(
334            )
335            .map_err(preparation)?;
336        let canonical = state.encode_current_msgpack().map_err(preparation)?;
337        heddle_object_model::object::thread_replication::hosted_import::initial_base_state(
338            &genesis, &canonical,
339        )
340        .map_err(preparation)?;
341        if selected.value.as_slice() != state.id().as_bytes() {
342            return Err(Error::Invalid(
343                "selected initial source differs from canonical seed",
344            ));
345        }
346        PackReader::open(
347            &directory.path().join("source.pack"),
348            &directory.path().join("source.idx"),
349        )
350        .map_err(preparation)?
351        .validate_source_closure_with_metadata(&state, &[], None, SOURCE_OBJECTS, SOURCE_BYTES)
352        .map_err(preparation)?;
353        return Ok(ValidatedSourceArtifacts {
354            directory,
355            operations,
356            genesis: original.clone(),
357            dependencies: Vec::new(),
358            state,
359            partial_trees: Vec::new(),
360            authority_admissions: BTreeMap::new(),
361        });
362    }
363    let mut geneses = BTreeMap::from([(selected_thread, genesis)]);
364    let mut dependencies = Vec::new();
365    for wrapper in dependency_records {
366        let record = wrapper
367            .genesis
368            .as_ref()
369            .ok_or(Error::Invalid("dependency signed genesis absent"))?;
370        let candidate = heddle_object_model::object::thread_replication::ThreadGenesis::decode(
371            &record.canonical_record,
372        )
373        .map_err(preparation)?;
374        let reference = ThreadRef {
375            spool: thread.spool.clone(),
376            id: Some(ThreadId {
377                value: candidate.id().map_err(preparation)?.as_bytes().to_vec(),
378            }),
379        };
380        let candidate = super::verify_origin(&wrapper, &reference)?;
381        let id = candidate.id().map_err(preparation)?;
382        if geneses.len() >= 128 || geneses.insert(id, candidate).is_some() {
383            return Err(Error::Invalid(
384                "duplicate or oversized dependency genesis set",
385            ));
386        }
387        dependencies.push(wrapper);
388    }
389    let mut claim_frontiers = BTreeMap::new();
390    for wrapper in std::iter::once(original).chain(&dependencies) {
391        let signed = wrapper
392            .genesis
393            .as_ref()
394            .ok_or(Error::Invalid("claim genesis absent"))?;
395        let genesis = heddle_object_model::object::thread_replication::ThreadGenesis::decode(
396            &signed.canonical_record,
397        )
398        .map_err(preparation)?;
399        let mut frontier = BTreeSet::new();
400        let claims = crate::replication::ownership::verify_claims(wrapper, &genesis)?;
401        let resolutions = crate::replication::ownership::verify_resolutions(wrapper, &genesis)?;
402        if claims.is_empty() && resolutions.is_empty() {
403            continue;
404        }
405        for claim in claims {
406            frontier.extend(
407                claim
408                    .original
409                    .verify()
410                    .map_err(preparation)?
411                    .source_frontier,
412            );
413        }
414        for resolution in resolutions {
415            frontier.extend(
416                heddle_object_model::object::thread_replication::ownership_resolution::ThreadOwnershipResolution::decode(&resolution.original.canonical)
417                    .map_err(preparation)?.frontier,
418            );
419        }
420        claim_frontiers.insert(genesis.id().map_err(preparation)?, frontier);
421    }
422    let mut originals = BTreeMap::new();
423    let mut decoded = BTreeMap::<ContentHash, ThreadOperation>::new();
424    let mut selected_operation = None;
425    let mut source_thread = selected_thread;
426    let mut inherited_bases = BTreeSet::new();
427    for _ in 0..128 {
428        let current = geneses
429            .get(&source_thread)
430            .ok_or(Error::Invalid("fork base source genesis absent"))?;
431        if selected.value.as_slice() != current.base.as_bytes() {
432            break;
433        }
434        let parent = current.parent.ok_or(Error::Invalid(
435            "non-system base has no original parent source",
436        ))?;
437        if !inherited_bases.insert(source_thread) {
438            return Err(Error::Invalid("fork base parent cycle"));
439        }
440        let ancestor = geneses
441            .get(&parent)
442            .ok_or(Error::Invalid("fork base parent original absent"))?;
443        if ancestor.spool != current.spool {
444            return Err(Error::Invalid("fork base crosses Spool"));
445        }
446        source_thread = parent;
447    }
448    if selected.value.as_slice()
449        == geneses
450            .get(&source_thread)
451            .ok_or(Error::Invalid("fork base source genesis absent"))?
452            .base
453            .as_bytes()
454    {
455        return Err(Error::Invalid("fork base source chain exceeds bound"));
456    }
457    for signed in &operations {
458        let operation = signed.verify().map_err(preparation)?;
459        let id = operation.id().map_err(preparation)?;
460        let state = operation
461            .source_state()
462            .map_err(preparation)?
463            .ok_or(Error::Invalid("non-source operation in source ancestry"))?;
464        if operation.thread == source_thread
465            && state.id().as_bytes().as_slice() == selected.value
466            && selected_operation.replace((id, state)).is_some()
467        {
468            return Err(Error::Invalid("ambiguous selected source proof"));
469        }
470        originals.insert(id, signed.clone());
471        if decoded.insert(id, operation).is_some() {
472            return Err(Error::Invalid("duplicate source proof"));
473        }
474    }
475    let mut authority_admissions = BTreeMap::new();
476    for receipt in receipt_records {
477        let statement = receipt.verify_signature().map_err(preparation)?;
478        let operation_id = statement.subject.operation_id().ok_or(Error::Invalid(
479            "source batch cannot carry ownership claim admission",
480        ))?;
481        let original = originals
482            .get(&operation_id)
483            .ok_or(Error::Invalid("unmatched source authority receipt"))?;
484        // Match immutable claims and signatures only. This self-described key
485        // is not enrolled here; the receiver must independently pin the issuer.
486        receipt.verify(original, &heddle_object_model::object::thread_replication::integration::TrustedHostedExecutor {
487            spool: statement.spool, spool_genesis: statement.spool_genesis, executor: statement.executor,
488        }).map_err(preparation)?;
489        if authority_admissions.insert(operation_id, receipt).is_some() {
490            return Err(Error::Invalid("duplicate source authority receipt"));
491        }
492    }
493    let (selected_id, state) =
494        selected_operation.ok_or(Error::Invalid("selected source proof absent"))?;
495    let mut pending = BTreeSet::from([selected_id]);
496    let mut seen = BTreeSet::new();
497    let mut used_threads = BTreeSet::new();
498    let mut edges = BTreeMap::new();
499    while let Some(id) = pending.pop_first() {
500        if !seen.insert(id) {
501            continue;
502        }
503        let operation = decoded
504            .get(&id)
505            .ok_or(Error::Invalid("incomplete source ancestry"))?;
506        let parents = operation
507            .parents
508            .iter()
509            .map(|id| {
510                decoded
511                    .get(id)
512                    .cloned()
513                    .ok_or(Error::Invalid("incomplete source ancestry"))
514            })
515            .collect::<Result<Vec<_>, _>>()?;
516        let genesis = geneses
517            .get(&operation.thread)
518            .ok_or(Error::Invalid("source dependency genesis absent"))?;
519        if used_threads.insert(operation.thread)
520            && let Some(frontier) = claim_frontiers.get(&operation.thread)
521        {
522            for head in frontier {
523                if decoded
524                    .get(head)
525                    .is_none_or(|source| source.thread != operation.thread)
526                {
527                    return Err(Error::Invalid(
528                        "ownership claim cutoff source proof absent or foreign",
529                    ));
530                }
531            }
532            pending.extend(frontier);
533        }
534        operation
535            .validate_parents(genesis, &parents)
536            .map_err(preparation)?;
537        let mut required = operation.parents.clone();
538        if let Some(receipt) = operation.local_integration().map_err(preparation)? {
539            let source = decoded
540                .get(&receipt.source_operation)
541                .ok_or(Error::Invalid(
542                    "local integration original source proof absent",
543                ))?;
544            receipt.validate_source(source).map_err(preparation)?;
545            required.insert(receipt.source_operation);
546            pending.insert(receipt.source_operation);
547        }
548        if let Some(receipt) = operation.integration().map_err(preparation)? {
549            let source = decoded
550                .get(&receipt.source_operation)
551                .ok_or(Error::Invalid(
552                    "hosted integration original source proof absent",
553                ))?;
554            receipt.validate_source(source).map_err(preparation)?;
555            required.insert(receipt.source_operation);
556            pending.insert(receipt.source_operation);
557        }
558        edges.insert(id, required);
559        pending.extend(
560            operation
561                .parents
562                .iter()
563                .filter(|id| !seen.contains(id))
564                .copied(),
565        );
566    }
567    if seen.len() != decoded.len()
568        || used_threads
569            .union(&inherited_bases)
570            .copied()
571            .collect::<BTreeSet<_>>()
572            != geneses.keys().copied().collect()
573    {
574        return Err(Error::Invalid("unselected source proofs"));
575    }
576    // A claim cutoff is an additional signed causal barrier. Ancestors retain
577    // their original local author; work outside it must follow the claim.
578    for (thread, frontier) in &claim_frontiers {
579        let mut history = BTreeSet::new();
580        let mut pending = frontier.clone();
581        while let Some(id) = pending.pop_first() {
582            if !history.insert(id) {
583                continue;
584            }
585            let operation = decoded
586                .get(&id)
587                .ok_or(Error::Invalid("claim cutoff ancestry absent"))?;
588            if operation.thread != *thread {
589                return Err(Error::Invalid("claim cutoff crosses Thread"));
590            }
591            pending.extend(&operation.parents);
592        }
593        {
594            for (id, operation) in &decoded {
595                if operation.thread == *thread && !history.contains(id) {
596                    if matches!(
597                        operation.source_author().map_err(preparation)?,
598                        Some(
599                            heddle_object_model::object::thread_replication::SourceAuthor::LocalKey
600                        )
601                    ) {
602                        return Err(Error::Invalid(
603                            "new local source lies outside signed ownership cutoff",
604                        ));
605                    }
606                    edges
607                        .get_mut(id)
608                        .ok_or(Error::Invalid("source topology entry absent"))?
609                        .extend(frontier);
610                }
611            }
612        }
613    }
614    let references = decoded
615        .values()
616        .map(|operation| {
617            operation
618                .reference_proof(
619                    geneses
620                        .get(&operation.thread)
621                        .ok_or(Error::Invalid("dependency genesis absent"))?,
622                )
623                .map_err(preparation)
624        })
625        .collect::<Result<Vec<_>, _>>()?
626        .into_iter()
627        .flatten()
628        .collect::<Vec<_>>();
629    let capture = decoded
630        .get(&selected_id)
631        .ok_or(Error::Invalid("selected source operation absent"))?
632        .source_result()
633        .map_err(preparation)?
634        .ok_or(Error::Invalid("selected operation has no source result"))?;
635    let pack = PackReader::open(
636        &directory.path().join("source.pack"),
637        &directory.path().join("source.idx"),
638    )
639    .map_err(preparation)?;
640    let partial_trees = if allow_partial {
641        pack.validate_visible_source_closure(&state, SOURCE_OBJECTS, SOURCE_BYTES)
642            .map_err(preparation)?
643            .partial_trees
644    } else {
645        pack.validate_source_closure_with_metadata(
646            &state,
647            &references,
648            capture.visibility.as_ref(),
649            SOURCE_OBJECTS,
650            SOURCE_BYTES,
651        )
652        .map_err(preparation)?;
653        Vec::new()
654    };
655    // Dependency-first installation makes foreign source authority available
656    // before admitting a local integration. Cycles cannot settle this graph.
657    let mut ready_ids: BTreeSet<_> = edges
658        .iter()
659        .filter(|(_, parents)| parents.is_empty())
660        .map(|(id, _)| *id)
661        .collect();
662    let mut children: BTreeMap<ContentHash, Vec<ContentHash>> = BTreeMap::new();
663    for (child, parents) in &edges {
664        for parent in parents {
665            children.entry(*parent).or_default().push(*child);
666        }
667    }
668    let mut ordered = Vec::new();
669    while let Some(id) = ready_ids.pop_first() {
670        ordered.push(
671            originals
672                .remove(&id)
673                .ok_or(Error::Invalid("duplicate source topology identity"))?,
674        );
675        if let Some(dependants) = children.get(&id) {
676            for child in dependants {
677                let parents = edges
678                    .get_mut(child)
679                    .ok_or(Error::Invalid("incomplete source topology"))?;
680                parents.remove(&id);
681                if parents.is_empty() {
682                    ready_ids.insert(*child);
683                }
684            }
685        }
686    }
687    if !originals.is_empty() {
688        return Err(Error::Invalid("source dependency cycle"));
689    }
690    Ok(ValidatedSourceArtifacts {
691        directory,
692        genesis: original.clone(),
693        operations: ordered,
694        authority_admissions,
695        dependencies,
696        state,
697        partial_trees,
698    })
699}
700fn preparation(error: impl std::fmt::Display) -> Error {
701    Error::Preparation(error.to_string())
702}
703
704#[cfg(test)]
705#[path = "staging_tests.rs"]
706mod tests;
heddle_thread_api/fetch/staging.rs

heddle_thread_api/fetch/
staging.rs
