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)?;
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)?,
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    ) -> Result<PublishContentClientFrame, Error> {
260        if self
261            .reference
262            .spool
263            .as_ref()
264            .is_none_or(|spool| spool.id.is_empty())
265            || self
266                .reference
267                .id
268                .as_ref()
269                .is_none_or(|id| id.value.len() != 32)
270            || options.source.public_key.len() != 32
271            || (!options.sharing_policy_version.is_empty()
272                && options.sharing_policy_version.len() != 32)
273        {
274            return Err(Error::Invalid(
275                "Thread, source endpoint and optional 32-byte policy version required",
276            ));
277        }
278        let destination = self
279            .remote
280            .description
281            .endpoint
282            .clone()
283            .ok_or(Error::Invalid("remote endpoint identity missing"))?;
284        Ok(PublishContentClientFrame {
285            client_operation_id: options.client_operation_id,
286            body: Some(publish_content_client_frame::Body::Open(
287                PublishContentOpen {
288                    thread: Some(self.reference.clone()),
289                    revision: Some(RevisionRef {
290                        spool: self.reference.spool.clone(),
291                        revision: Some(revision_ref::Revision::State(
292                            api::heddle::api::common::StateId {
293                                value: source.revision.as_bytes().to_vec(),
294                            },
295                        )),
296                    }),
297                    sharing_policy_version: options.sharing_policy_version,
298                    packs: source.artifacts.to_vec(),
299                    checkpoint: options.checkpoint,
300                    source: Some(options.source),
301                    destination: Some(destination),
302                },
303            )),
304        })
305    }
306}
307
308fn artifact(path: &Path, kind: pack_extent::Kind) -> Result<PackExtent, Error> {
309    let mut file = File::open(path)?;
310    let length = file.metadata()?.len();
311    let mut hash = blake3::Hasher::new();
312    hash.update_reader(&mut file)?;
313    let address = ObjectAddress {
314        algorithm: "blake3".into(),
315        digest: hash.finalize().as_bytes().to_vec(),
316    };
317    Ok(PackExtent {
318        pack: Some(address.clone()),
319        kind: kind as i32,
320        offset: 0,
321        length,
322        extent_digest: Some(address),
323    })
324}
325fn store_error(error: impl std::fmt::Display) -> Error {
326    transport::Error::Io(error.to_string()).into()
327}
heddle_thread_api/publication/source.rs

heddle_thread_api/publication/
source.rs
