heddle_thread_api/publication/
source.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Build and retain an exact selected-source transfer for explicit publication
3//! and retries. The application chooses the scratch directory and disclosure
4//! policy; preparation never reads historical States or adjacent checkouts.
5use std::{
6    fs::{File, OpenOptions},
7    path::Path,
8};
9
10use api::v2::client::RpcTransport;
11use heddle_object_model::object::{
12    EntryRedactions, ObjectSource, State, StateId, source_target::capture::ReferenceProof,
13};
14use heddle_pack::store::pack::{
15    StreamingPackBuilder, build_source_pack_with_references, build_visible_source_pack,
16};
17
18use super::{Error, PreparedPublication, PublicationOriginals};
19use crate::{Thread, contract::*, transport};
20
21pub struct SourceBudget {
22    pub max_objects: usize,
23    pub max_decoded_bytes: u64,
24}
25
26/// The caller retains the same options for retry. The source endpoint is the
27/// local Iroh identity; credential signing may use a distinct owner device key.
28pub struct PublicationOptions {
29    pub client_operation_id: String,
30    pub source: EndpointRef,
31    /// Empty omits the compare-and-set condition; otherwise exactly 32 bytes.
32    /// Explicit publication does not enable ongoing synchronization.
33    pub sharing_policy_version: Vec<u8>,
34    pub checkpoint: Option<TransferCheckpoint>,
35}
36
37pub struct SourcePack {
38    directory: tempfile::TempDir,
39    revision: StateId,
40    artifacts: [PackExtent; 2],
41}
42
43/// Read-side disclosure artifacts. A partial projection cannot be passed to
44/// `publish_source`, which requires a complete [`SourcePack`].
45pub struct VisibleSourcePack {
46    source: SourcePack,
47    complete: bool,
48}
49
50impl VisibleSourcePack {
51    /// Prepare an admitted source read, omitting hidden source entries and
52    /// reference descriptors when an entry restriction applies. Full reads
53    /// retain their independently verified reference closures.
54    pub fn prepare(
55        source: &impl ObjectSource,
56        selected: &State,
57        references: &[ReferenceProof],
58        redactions: &EntryRedactions,
59        scratch_root: &Path,
60        budget: SourceBudget,
61    ) -> Result<Self, Error> {
62        let (source, complete) = SourcePack::prepare_disclosure(
63            source,
64            selected,
65            references,
66            Some(redactions),
67            scratch_root,
68            budget,
69        )?;
70        Ok(Self { source, complete })
71    }
72
73    /// Whether the pack proves full source and selected descriptor availability.
74    pub fn is_complete(&self) -> bool {
75        self.complete
76    }
77
78    /// Whole independently hashed pack/index artifacts in transmission order.
79    pub fn artifacts(&self) -> &[PackExtent; 2] {
80        self.source.artifacts()
81    }
82
83    /// Stream the retained read artifacts while this value owns their scratch.
84    pub async fn open_artifacts(&self) -> Result<[tokio::fs::File; 2], Error> {
85        self.source.open_artifacts().await
86    }
87}
88
89impl SourcePack {
90    /// Performs disk I/O and compression. Async applications run preparation on
91    /// their blocking-work executor. The output owns and removes its scratch.
92    pub fn prepare(
93        source: &impl ObjectSource,
94        selected: &State,
95        scratch_root: &Path,
96        budget: SourceBudget,
97    ) -> Result<Self, Error> {
98        Self::prepare_with_references(source, selected, &[], scratch_root, budget)
99    }
100
101    /// Include only descriptor closures selected by independently verified source
102    /// operation proofs. A reference never authorizes another source revision.
103    pub fn prepare_with_references(
104        source: &impl ObjectSource,
105        selected: &State,
106        references: &[ReferenceProof],
107        scratch_root: &Path,
108        budget: SourceBudget,
109    ) -> Result<Self, Error> {
110        Self::prepare_disclosure(source, selected, references, None, scratch_root, budget)
111            .map(|(source, _)| source)
112    }
113
114    fn prepare_disclosure(
115        source: &impl ObjectSource,
116        selected: &State,
117        references: &[ReferenceProof],
118        redactions: Option<&EntryRedactions>,
119        scratch_root: &Path,
120        budget: SourceBudget,
121    ) -> Result<(Self, bool), Error> {
122        let directory = tempfile::Builder::new()
123            .prefix("thread-source-")
124            .tempdir_in(scratch_root)?;
125        let pack_path = directory.path().join("source.pack");
126        let index_path = directory.path().join("source.idx");
127        let pack = OpenOptions::new()
128            .read(true)
129            .write(true)
130            .create_new(true)
131            .open(&pack_path)?;
132        let builder = StreamingPackBuilder::new(
133            pack,
134            index_path.clone(),
135            Default::default(),
136            directory.path().join("buckets"),
137        )
138        .map_err(store_error)?;
139        let (pack, _, complete) = match redactions {
140            Some(redactions) => build_visible_source_pack(
141                builder,
142                source,
143                selected,
144                references,
145                redactions,
146                budget.max_objects,
147                budget.max_decoded_bytes,
148            ),
149            None => build_source_pack_with_references(
150                builder,
151                source,
152                selected,
153                references,
154                budget.max_objects,
155                budget.max_decoded_bytes,
156            )
157            .map(|(output, stats)| (output, stats, true)),
158        }
159        .map_err(store_error)?;
160        drop(pack);
161        let artifacts = [
162            artifact(&pack_path, pack_extent::Kind::NativePack)?,
163            artifact(&index_path, pack_extent::Kind::NativeIndex)?,
164        ];
165        Ok((
166            Self {
167                directory,
168                revision: selected.id(),
169                artifacts,
170            },
171            complete,
172        ))
173    }
174
175    /// Whole, independently hashed artifacts in transmission order. Source
176    /// preparation bounds the decoded closure before producing this inventory.
177    pub fn artifacts(&self) -> &[PackExtent; 2] {
178        &self.artifacts
179    }
180
181    /// Open the exact prepared artifacts without buffering their bodies. Keep
182    /// this SourcePack alive until the readers finish; dropping it removes its
183    /// temporary files, including after a cancelled transfer.
184    pub async fn open_artifacts(&self) -> Result<[tokio::fs::File; 2], Error> {
185        Ok([
186            tokio::fs::File::open(self.directory.path().join("source.pack")).await?,
187            tokio::fs::File::open(self.directory.path().join("source.idx")).await?,
188        ])
189    }
190
191    pub fn revision(&self) -> StateId {
192        self.revision
193    }
194
195    /// The same inventory digest verified by the publication receipt.
196    pub fn inventory_digest(&self) -> Result<[u8; 32], Error> {
197        super::inventory_digest(&self.artifacts)
198    }
199}
200
201impl<T: RpcTransport<Error = transport::Error>> Thread<'_, T> {
202    /// One exchange, directly to this Thread's endpoint. The source capture must
203    /// carry its complete original proofs. Preparation and binding are local;
204    /// no head lookup, proxy call or implicit capture happens here.
205    pub async fn publish_source(
206        &self,
207        source: &SourcePack,
208        originals: &PublicationOriginals,
209        options: PublicationOptions,
210    ) -> Result<PublicationReceipt, Error> {
211        let opening = self.publication_opening(source, options, originals)?;
212        let [pack, index] = source.open_artifacts().await?;
213        self.remote
214            .publish_content(&opening, originals, [pack, index])
215            .await
216    }
217    /// Local preparation over exact source/originals; no network authorization or
218    /// implicit acceptance. Call sign_acceptance explicitly, then send_prepared.
219    pub fn prepare_publication(
220        &self,
221        source: &SourcePack,
222        originals: PublicationOriginals,
223        options: PublicationOptions,
224        spool_genesis: heddle_object_model::object::ContentHash,
225    ) -> Result<PreparedPublication, Error> {
226        Ok(PreparedPublication::new(
227            self.publication_opening(source, options, &originals)?,
228            originals,
229            spool_genesis,
230        )?)
231    }
232
233    pub async fn send_prepared(
234        &self,
235        source: &SourcePack,
236        prepared: &PreparedPublication,
237    ) -> Result<PublicationReceipt, Error> {
238        let Some(publish_content_client_frame::Body::Open(open)) = &prepared.opening().body else {
239            return Err(Error::Invalid("prepared Open required"));
240        };
241        if open.thread.as_ref() != Some(&self.reference)
242            || open.packs.as_slice() != source.artifacts()
243            || prepared.plan().intent().revision != source.revision()
244        {
245            return Err(Error::Invalid(
246                "prepared publication differs from selected source",
247            ));
248        }
249        let artifacts = source.open_artifacts().await?;
250        self.remote
251            .publish_content(prepared.opening(), prepared.originals(), artifacts)
252            .await
253    }
254
255    fn publication_opening(
256        &self,
257        source: &SourcePack,
258        options: PublicationOptions,
259        originals: &PublicationOriginals,
260    ) -> Result<PublishContentClientFrame, Error> {
261        if self
262            .reference
263            .spool
264            .as_ref()
265            .is_none_or(|spool| spool.id.is_empty())
266            || self
267                .reference
268                .id
269                .as_ref()
270                .is_none_or(|id| id.value.len() != 32)
271            || options.source.public_key.len() != 32
272            || (!options.sharing_policy_version.is_empty()
273                && options.sharing_policy_version.len() != 32)
274        {
275            return Err(Error::Invalid(
276                "Thread, source endpoint and optional 32-byte policy version required",
277            ));
278        }
279        let destination = self
280            .remote
281            .description
282            .endpoint
283            .clone()
284            .ok_or(Error::Invalid("remote endpoint identity missing"))?;
285        let mut import_authority = None;
286        for bundle in originals
287            .operations
288            .iter()
289            .filter_map(|batch| batch.import_authority.as_ref())
290        {
291            if import_authority.is_some_and(|previous| previous != bundle) {
292                return Err(Error::Invalid(
293                    "publication originals have different import histories",
294                ));
295            }
296            crate::hybrid::bundle(Some(bundle)).map_err(Error::Invalid)?;
297            import_authority = Some(bundle);
298        }
299        let mut native_authority = None;
300        for bundle in originals
301            .operations
302            .iter()
303            .filter_map(|batch| batch.native_authority.as_ref())
304        {
305            if native_authority.is_some_and(|previous| previous != bundle) {
306                return Err(Error::Invalid(
307                    "publication originals have different native histories",
308                ));
309            }
310            native_authority = Some(bundle);
311        }
312        crate::hybrid::bundles(import_authority, native_authority).map_err(Error::Invalid)?;
313        let protocol = if import_authority.is_some()
314            || native_authority.is_some()
315            || originals
316                .geneses
317                .iter()
318                .any(|g| g.native_genesis_authority.is_some())
319        {
320            api::import_authority::require_hybrid_peer(self.remote.description.protocol.as_ref())
321                .map_err(|_| {
322                Error::Invalid(
323                    "HYBRID originals require a capable publication peer (api#307 cutover)",
324                )
325            })?;
326            Some(crate::hybrid::protocol())
327        } else {
328            crate::hybrid::sync_protocol()
329        };
330        Ok(PublishContentClientFrame {
331            client_operation_id: options.client_operation_id,
332            body: Some(publish_content_client_frame::Body::Open(
333                PublishContentOpen {
334                    native_authority: native_authority.cloned(),
335                    thread: Some(self.reference.clone()),
336                    revision: Some(RevisionRef {
337                        spool: self.reference.spool.clone(),
338                        revision: Some(revision_ref::Revision::State(
339                            api::heddle::api::common::StateId {
340                                value: source.revision.as_bytes().to_vec(),
341                            },
342                        )),
343                    }),
344                    sharing_policy_version: options.sharing_policy_version,
345                    packs: source.artifacts.to_vec(),
346                    checkpoint: options.checkpoint,
347                    source: Some(options.source),
348                    destination: Some(destination),
349                    semantic_indexes: Vec::new(),
350                    protocol,
351                    import_authority: import_authority.cloned(),
352                },
353            )),
354        })
355    }
356}
357
358fn artifact(path: &Path, kind: pack_extent::Kind) -> Result<PackExtent, Error> {
359    let mut file = File::open(path)?;
360    let length = file.metadata()?.len();
361    let mut hash = blake3::Hasher::new();
362    hash.update_reader(&mut file)?;
363    let address = ObjectAddress {
364        algorithm: "blake3".into(),
365        digest: hash.finalize().as_bytes().to_vec(),
366    };
367    Ok(PackExtent {
368        pack: Some(address.clone()),
369        kind: kind as i32,
370        offset: 0,
371        length,
372        extent_digest: Some(address),
373    })
374}
375fn store_error(error: impl std::fmt::Display) -> Error {
376    transport::Error::Io(error.to_string()).into()
377}
heddle_thread_api/publication/source.rs

heddle_thread_api/publication/
source.rs
