use std::{
collections::{BTreeMap, BTreeSet},
path::Path,
};
use api::v2::client::MessageReader;
use crypto::thread_operation::SignedOperation;
use heddle_object_model::object::{ContentHash, State, thread_replication::ThreadOperation};
use heddle_pack::store::pack::PackReader;
use prost::Message;
use tokio::io::AsyncWriteExt;
use super::{Download, Error, Item};
use crate::{contract::*, transport};
const METADATA_BYTES: usize = 16 * 1024 * 1024;
const SOURCE_BYTES: u64 = 256 * 1024 * 1024;
const SOURCE_OBJECTS: usize = 100_000;
pub struct StagedSource {
pub(super) directory: tempfile::TempDir,
pub(super) ready: TransferReady,
pub(super) operations: Vec<SignedOperation>,
pub(super) dependencies: Vec<ThreadGenesisRecord>,
pub(super) state: State,
pub(super) partial_trees: Vec<heddle_object_model::object::PartialTree>,
pub(super) authority_admissions:
BTreeMap<ContentHash, crypto::thread_authority_admission::SignedAuthorityAdmission>,
}
impl StagedSource {
pub fn artifact_paths(&self) -> [std::path::PathBuf; 2] {
[
self.directory.path().join("source.pack"),
self.directory.path().join("source.idx"),
]
}
pub fn operations(&self) -> &[SignedOperation] {
&self.operations
}
pub fn dependency_geneses(&self) -> &[ThreadGenesisRecord] {
&self.dependencies
}
pub fn ready(&self) -> &TransferReady {
&self.ready
}
pub fn state(&self) -> &State {
&self.state
}
pub fn is_complete(&self) -> bool {
self.ready.full_closure_available
}
}
impl<R: MessageReader<Error = transport::Error>> Download<R> {
pub async fn stage(mut self, scratch: &Path) -> Result<StagedSource, Error> {
if self.state.facets != [SharedFacet::Source as i32] {
return Err(Error::Invalid("staging requires the source facet alone"));
}
let total = self
.state
.ready
.packs
.iter()
.try_fold(0u64, |sum, extent| sum.checked_add(extent.length))
.ok_or(Error::Invalid("source artifact length overflow"))?;
if total > SOURCE_BYTES {
return Err(Error::Invalid("staged source exceeds 256 MiB"));
}
self.state.limits.max_operations = self.state.limits.max_operations.min(10_000);
let directory = tempfile::Builder::new()
.prefix("thread-download-")
.tempdir_in(scratch)?;
let mut files = [
tokio::fs::File::create(directory.path().join("source.pack")).await?,
tokio::fs::File::create(directory.path().join("source.idx")).await?,
];
let mut operations = Vec::new();
let mut receipt_records = Vec::new();
let mut dependencies = Vec::new();
let mut metadata_bytes = 0usize;
let mut complete = false;
while let Some(item) = self.next().await? {
match item {
Item::Pack(chunk) => {
let kind = chunk
.extent
.as_ref()
.ok_or(Error::Invalid("chunk extent absent"))?
.kind;
let index = match pack_extent::Kind::try_from(kind) {
Ok(pack_extent::Kind::NativePack) => 0,
Ok(pack_extent::Kind::NativeIndex) => 1,
_ => return Err(Error::Invalid("native source artifacts required")),
};
files[index].write_all(&chunk.data).await?;
}
Item::Operations(batch) => {
metadata_bytes = metadata_bytes
.checked_add(batch.encoded_len())
.ok_or(Error::Invalid("source metadata length overflow"))?;
if metadata_bytes > METADATA_BYTES {
return Err(Error::Invalid("staged source metadata exceeds 16 MiB"));
}
for received in crate::authority_admission::match_batch(&batch)? {
operations.push(received.original);
receipt_records.extend(received.authority_admission);
}
}
Item::ThreadGenesis(record) => {
metadata_bytes = metadata_bytes
.checked_add(record.encoded_len())
.ok_or(Error::Invalid("source metadata length overflow"))?;
if metadata_bytes > METADATA_BYTES || dependencies.len() >= 127 {
return Err(Error::Invalid("dependency metadata exceeds bounds"));
}
dependencies.push(record);
}
Item::Complete(_) => complete = true,
Item::Sidecar(_) => return Err(Error::Invalid("source staging excludes sidecars")),
}
}
if !complete {
return Err(Error::Invalid("source staging requires Complete"));
}
for file in &mut files {
file.flush().await?;
file.sync_all().await?;
}
drop(files);
let ready = self.state.ready;
tokio::task::spawn_blocking(move || {
validate_with_receipts(directory, ready, operations, dependencies, receipt_records)
})
.await
.map_err(|error| Error::Preparation(error.to_string()))?
}
}
#[cfg(test)]
fn validate(
directory: tempfile::TempDir,
ready: TransferReady,
operations: Vec<SignedOperation>,
dependencies: Vec<ThreadGenesisRecord>,
) -> Result<StagedSource, Error> {
validate_with_receipts(directory, ready, operations, dependencies, Vec::new())
}
struct DisclosureInput {
directory: tempfile::TempDir,
operations: Vec<SignedOperation>,
dependency_records: Vec<ThreadGenesisRecord>,
receipt_records: Vec<crypto::thread_authority_admission::SignedAuthorityAdmission>,
allow_partial: bool,
}
pub(super) fn validate_with_receipts(
directory: tempfile::TempDir,
ready: TransferReady,
operations: Vec<SignedOperation>,
dependencies: Vec<ThreadGenesisRecord>,
receipt_records: Vec<crypto::thread_authority_admission::SignedAuthorityAdmission>,
) -> Result<StagedSource, Error> {
let value = validate_disclosure_artifacts(
ready
.thread
.as_ref()
.ok_or(Error::Invalid("Thread absent"))?,
ready
.current
.as_ref()
.ok_or(Error::Invalid("revision absent"))?,
ready
.thread_genesis
.as_ref()
.ok_or(Error::Invalid("original genesis absent"))?,
DisclosureInput {
directory,
operations,
dependency_records: dependencies,
receipt_records,
allow_partial: !ready.full_closure_available,
},
)?;
Ok(StagedSource {
directory: value.directory,
ready,
operations: value.operations,
dependencies: value.dependencies,
state: value.state,
partial_trees: value.partial_trees,
authority_admissions: value.authority_admissions,
})
}
pub struct ValidatedSourceArtifacts {
directory: tempfile::TempDir,
operations: Vec<SignedOperation>,
genesis: ThreadGenesisRecord,
dependencies: Vec<ThreadGenesisRecord>,
state: State,
partial_trees: Vec<heddle_object_model::object::PartialTree>,
authority_admissions:
BTreeMap<ContentHash, crypto::thread_authority_admission::SignedAuthorityAdmission>,
}
impl ValidatedSourceArtifacts {
pub fn artifact_paths(&self) -> [std::path::PathBuf; 2] {
[
self.directory.path().join("source.pack"),
self.directory.path().join("source.idx"),
]
}
pub fn operations(&self) -> &[SignedOperation] {
&self.operations
}
pub fn geneses(&self) -> impl Iterator<Item = &ThreadGenesisRecord> {
std::iter::once(&self.genesis).chain(&self.dependencies)
}
pub fn state(&self) -> &State {
&self.state
}
pub fn authority_admissions(
&self,
) -> &BTreeMap<ContentHash, crypto::thread_authority_admission::SignedAuthorityAdmission> {
&self.authority_admissions
}
}
pub(crate) fn validate_artifacts(
directory: tempfile::TempDir,
thread: &ThreadRef,
revision: &RevisionRef,
original: &ThreadGenesisRecord,
operations: Vec<SignedOperation>,
dependency_records: Vec<ThreadGenesisRecord>,
receipt_records: Vec<crypto::thread_authority_admission::SignedAuthorityAdmission>,
) -> Result<ValidatedSourceArtifacts, Error> {
validate_disclosure_artifacts(
thread,
revision,
original,
DisclosureInput {
directory,
operations,
dependency_records,
receipt_records,
allow_partial: false,
},
)
}
fn validate_disclosure_artifacts(
thread: &ThreadRef,
revision: &RevisionRef,
original: &ThreadGenesisRecord,
input: DisclosureInput,
) -> Result<ValidatedSourceArtifacts, Error> {
let DisclosureInput {
directory,
operations,
dependency_records,
receipt_records,
allow_partial,
} = input;
if operations.len() > 10_000
|| dependency_records.len() >= 128
|| receipt_records.len() > operations.len()
{
return Err(Error::Invalid("source original graph exceeds bounds"));
}
let mut metadata = original.encoded_len();
for record in &dependency_records {
metadata = metadata.saturating_add(record.encoded_len());
}
for operation in &operations {
metadata = metadata.saturating_add(operation.canonical.len() + operation.signature.len());
}
for receipt in &receipt_records {
metadata = metadata.saturating_add(receipt.canonical.len() + receipt.signature.len());
}
let mut evidence_ids = BTreeSet::new();
for wrapper in std::iter::once(original).chain(&dependency_records) {
for record in &wrapper.boundary_acceptances {
evidence_ids.insert(heddle_object_model::object::ContentHash::compute_typed(
heddle_object_model::object::original_boundary_acceptance::FORMAT,
&record.canonical_record,
));
if evidence_ids.len() > crate::boundary_acceptance::MAX_ACCEPTANCES {
return Err(Error::Invalid("boundary evidence count exceeded"));
}
}
}
for receipt in &receipt_records {
if let Some(evidence) = &receipt.boundary_acceptance {
if evidence_ids.insert(
evidence
.verify_signature()
.map_err(preparation)?
.id()
.map_err(preparation)?,
) {
metadata =
metadata.saturating_add(evidence.canonical.len() + evidence.signature.len());
}
if evidence_ids.len() > crate::boundary_acceptance::MAX_ACCEPTANCES {
return Err(Error::Invalid("boundary evidence count exceeded"));
}
}
}
if metadata > METADATA_BYTES {
return Err(Error::Invalid("source metadata exceeds 16 MiB"));
}
if revision.spool != thread.spool {
return Err(Error::Invalid("source revision crosses Spool"));
}
let genesis = super::verify_origin(original, thread)?;
let Some(revision_ref::Revision::State(selected)) = revision.revision.as_ref() else {
return Err(Error::Invalid("exact native State required"));
};
let selected_thread = genesis.id().map_err(preparation)?;
if operations.is_empty() {
if !dependency_records.is_empty() || !receipt_records.is_empty() {
return Err(Error::Invalid(
"initial source cannot carry dependency originals",
));
}
let state =
heddle_object_model::object::thread_replication::hosted_import::synthetic_initial_base(
)
.map_err(preparation)?;
let canonical = state.encode_current_msgpack().map_err(preparation)?;
heddle_object_model::object::thread_replication::hosted_import::initial_base_state(
&genesis, &canonical,
)
.map_err(preparation)?;
if selected.value.as_slice() != state.id().as_bytes() {
return Err(Error::Invalid(
"selected initial source differs from canonical seed",
));
}
PackReader::open(
&directory.path().join("source.pack"),
&directory.path().join("source.idx"),
)
.map_err(preparation)?
.validate_source_closure_with_metadata(&state, &[], None, SOURCE_OBJECTS, SOURCE_BYTES)
.map_err(preparation)?;
return Ok(ValidatedSourceArtifacts {
directory,
operations,
genesis: original.clone(),
dependencies: Vec::new(),
state,
partial_trees: Vec::new(),
authority_admissions: BTreeMap::new(),
});
}
let mut geneses = BTreeMap::from([(selected_thread, genesis)]);
let mut dependencies = Vec::new();
for wrapper in dependency_records {
let record = wrapper
.genesis
.as_ref()
.ok_or(Error::Invalid("dependency signed genesis absent"))?;
let candidate = heddle_object_model::object::thread_replication::ThreadGenesis::decode(
&record.canonical_record,
)
.map_err(preparation)?;
let reference = ThreadRef {
spool: thread.spool.clone(),
id: Some(ThreadId {
value: candidate.id().map_err(preparation)?.as_bytes().to_vec(),
}),
};
let candidate = super::verify_origin(&wrapper, &reference)?;
let id = candidate.id().map_err(preparation)?;
if geneses.len() >= 128 || geneses.insert(id, candidate).is_some() {
return Err(Error::Invalid(
"duplicate or oversized dependency genesis set",
));
}
dependencies.push(wrapper);
}
let mut claim_frontiers = BTreeMap::new();
for wrapper in std::iter::once(original).chain(&dependencies) {
let signed = wrapper
.genesis
.as_ref()
.ok_or(Error::Invalid("claim genesis absent"))?;
let genesis = heddle_object_model::object::thread_replication::ThreadGenesis::decode(
&signed.canonical_record,
)
.map_err(preparation)?;
let mut frontier = BTreeSet::new();
let claims = crate::replication::ownership::verify_claims(wrapper, &genesis)?;
let resolutions = crate::replication::ownership::verify_resolutions(wrapper, &genesis)?;
if claims.is_empty() && resolutions.is_empty() {
continue;
}
for claim in claims {
frontier.extend(
claim
.original
.verify()
.map_err(preparation)?
.source_frontier,
);
}
for resolution in resolutions {
frontier.extend(
heddle_object_model::object::thread_replication::ownership_resolution::ThreadOwnershipResolution::decode(&resolution.original.canonical)
.map_err(preparation)?.frontier,
);
}
claim_frontiers.insert(genesis.id().map_err(preparation)?, frontier);
}
let mut originals = BTreeMap::new();
let mut decoded = BTreeMap::<ContentHash, ThreadOperation>::new();
let mut selected_operation = None;
let mut source_thread = selected_thread;
let mut inherited_bases = BTreeSet::new();
for _ in 0..128 {
let current = geneses
.get(&source_thread)
.ok_or(Error::Invalid("fork base source genesis absent"))?;
if selected.value.as_slice() != current.base.as_bytes() {
break;
}
let parent = current.parent.ok_or(Error::Invalid(
"non-system base has no original parent source",
))?;
if !inherited_bases.insert(source_thread) {
return Err(Error::Invalid("fork base parent cycle"));
}
let ancestor = geneses
.get(&parent)
.ok_or(Error::Invalid("fork base parent original absent"))?;
if ancestor.spool != current.spool {
return Err(Error::Invalid("fork base crosses Spool"));
}
source_thread = parent;
}
if selected.value.as_slice()
== geneses
.get(&source_thread)
.ok_or(Error::Invalid("fork base source genesis absent"))?
.base
.as_bytes()
{
return Err(Error::Invalid("fork base source chain exceeds bound"));
}
for signed in &operations {
let operation = signed.verify().map_err(preparation)?;
let id = operation.id().map_err(preparation)?;
let state = operation
.source_state()
.map_err(preparation)?
.ok_or(Error::Invalid("non-source operation in source ancestry"))?;
if operation.thread == source_thread
&& state.id().as_bytes().as_slice() == selected.value
&& selected_operation.replace((id, state)).is_some()
{
return Err(Error::Invalid("ambiguous selected source proof"));
}
originals.insert(id, signed.clone());
if decoded.insert(id, operation).is_some() {
return Err(Error::Invalid("duplicate source proof"));
}
}
let mut authority_admissions = BTreeMap::new();
for receipt in receipt_records {
let statement = receipt.verify_signature().map_err(preparation)?;
let operation_id = statement.subject.operation_id().ok_or(Error::Invalid(
"source batch cannot carry ownership claim admission",
))?;
let original = originals
.get(&operation_id)
.ok_or(Error::Invalid("unmatched source authority receipt"))?;
receipt.verify(original, &heddle_object_model::object::thread_replication::integration::TrustedHostedExecutor {
spool: statement.spool, spool_genesis: statement.spool_genesis, executor: statement.executor,
}).map_err(preparation)?;
if authority_admissions.insert(operation_id, receipt).is_some() {
return Err(Error::Invalid("duplicate source authority receipt"));
}
}
let (selected_id, state) =
selected_operation.ok_or(Error::Invalid("selected source proof absent"))?;
let mut pending = BTreeSet::from([selected_id]);
let mut seen = BTreeSet::new();
let mut used_threads = BTreeSet::new();
let mut edges = BTreeMap::new();
while let Some(id) = pending.pop_first() {
if !seen.insert(id) {
continue;
}
let operation = decoded
.get(&id)
.ok_or(Error::Invalid("incomplete source ancestry"))?;
let parents = operation
.parents
.iter()
.map(|id| {
decoded
.get(id)
.cloned()
.ok_or(Error::Invalid("incomplete source ancestry"))
})
.collect::<Result<Vec<_>, _>>()?;
let genesis = geneses
.get(&operation.thread)
.ok_or(Error::Invalid("source dependency genesis absent"))?;
if used_threads.insert(operation.thread)
&& let Some(frontier) = claim_frontiers.get(&operation.thread)
{
for head in frontier {
if decoded
.get(head)
.is_none_or(|source| source.thread != operation.thread)
{
return Err(Error::Invalid(
"ownership claim cutoff source proof absent or foreign",
));
}
}
pending.extend(frontier);
}
operation
.validate_parents(genesis, &parents)
.map_err(preparation)?;
let mut required = operation.parents.clone();
if let Some(receipt) = operation.local_integration().map_err(preparation)? {
let source = decoded
.get(&receipt.source_operation)
.ok_or(Error::Invalid(
"local integration original source proof absent",
))?;
receipt.validate_source(source).map_err(preparation)?;
required.insert(receipt.source_operation);
pending.insert(receipt.source_operation);
}
if let Some(receipt) = operation.integration().map_err(preparation)? {
let source = decoded
.get(&receipt.source_operation)
.ok_or(Error::Invalid(
"hosted integration original source proof absent",
))?;
receipt.validate_source(source).map_err(preparation)?;
required.insert(receipt.source_operation);
pending.insert(receipt.source_operation);
}
edges.insert(id, required);
pending.extend(
operation
.parents
.iter()
.filter(|id| !seen.contains(id))
.copied(),
);
}
if seen.len() != decoded.len()
|| used_threads
.union(&inherited_bases)
.copied()
.collect::<BTreeSet<_>>()
!= geneses.keys().copied().collect()
{
return Err(Error::Invalid("unselected source proofs"));
}
for (thread, frontier) in &claim_frontiers {
let mut history = BTreeSet::new();
let mut pending = frontier.clone();
while let Some(id) = pending.pop_first() {
if !history.insert(id) {
continue;
}
let operation = decoded
.get(&id)
.ok_or(Error::Invalid("claim cutoff ancestry absent"))?;
if operation.thread != *thread {
return Err(Error::Invalid("claim cutoff crosses Thread"));
}
pending.extend(&operation.parents);
}
{
for (id, operation) in &decoded {
if operation.thread == *thread && !history.contains(id) {
if matches!(
operation.source_author().map_err(preparation)?,
Some(
heddle_object_model::object::thread_replication::SourceAuthor::LocalKey
)
) {
return Err(Error::Invalid(
"new local source lies outside signed ownership cutoff",
));
}
edges
.get_mut(id)
.ok_or(Error::Invalid("source topology entry absent"))?
.extend(frontier);
}
}
}
}
let references = decoded
.values()
.map(|operation| {
operation
.reference_proof(
geneses
.get(&operation.thread)
.ok_or(Error::Invalid("dependency genesis absent"))?,
)
.map_err(preparation)
})
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.flatten()
.collect::<Vec<_>>();
let capture = decoded
.get(&selected_id)
.ok_or(Error::Invalid("selected source operation absent"))?
.source_result()
.map_err(preparation)?
.ok_or(Error::Invalid("selected operation has no source result"))?;
let pack = PackReader::open(
&directory.path().join("source.pack"),
&directory.path().join("source.idx"),
)
.map_err(preparation)?;
let partial_trees = if allow_partial {
pack.validate_visible_source_closure(&state, SOURCE_OBJECTS, SOURCE_BYTES)
.map_err(preparation)?
.partial_trees
} else {
pack.validate_source_closure_with_metadata(
&state,
&references,
capture.visibility.as_ref(),
SOURCE_OBJECTS,
SOURCE_BYTES,
)
.map_err(preparation)?;
Vec::new()
};
let mut ready_ids: BTreeSet<_> = edges
.iter()
.filter(|(_, parents)| parents.is_empty())
.map(|(id, _)| *id)
.collect();
let mut children: BTreeMap<ContentHash, Vec<ContentHash>> = BTreeMap::new();
for (child, parents) in &edges {
for parent in parents {
children.entry(*parent).or_default().push(*child);
}
}
let mut ordered = Vec::new();
while let Some(id) = ready_ids.pop_first() {
ordered.push(
originals
.remove(&id)
.ok_or(Error::Invalid("duplicate source topology identity"))?,
);
if let Some(dependants) = children.get(&id) {
for child in dependants {
let parents = edges
.get_mut(child)
.ok_or(Error::Invalid("incomplete source topology"))?;
parents.remove(&id);
if parents.is_empty() {
ready_ids.insert(*child);
}
}
}
}
if !originals.is_empty() {
return Err(Error::Invalid("source dependency cycle"));
}
Ok(ValidatedSourceArtifacts {
directory,
genesis: original.clone(),
operations: ordered,
authority_admissions,
dependencies,
state,
partial_trees,
})
}
fn preparation(error: impl std::fmt::Display) -> Error {
Error::Preparation(error.to_string())
}
#[cfg(test)]
#[path = "staging_tests.rs"]
mod tests;