heddle_thread_api/
content.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Bounded full-blob reads at one exact revision. Large/ranged transfers use the
3//! same generated ContentServiceReadContent stream and a caller-owned sink.
4use api::v2::client::RpcTransport;
5
6pub use crate::contract::blob_read::Source as BlobSource;
7use crate::{
8    Remote,
9    contract::*,
10    observation::{self, Error},
11    rpc, transport,
12};
13
14pub struct Blob {
15    pub source: BlobSource,
16    pub object_hash: Vec<u8>,
17    pub bytes: Vec<u8>,
18}
19
20impl<T: RpcTransport<Error = transport::Error>> Remote<T> {
21    /// One request for any mixture of paths and missing object hashes. The
22    /// The selected Thread and revision pin content and authorization without a
23    /// mutable Thread-tip lookup.
24    pub async fn read_blobs(
25        &self,
26        thread: ThreadRef,
27        revision: RevisionRef,
28        sources: Vec<BlobSource>,
29    ) -> Result<Vec<Blob>, Error> {
30        if thread.spool != revision.spool
31            || thread.id.as_ref().is_none_or(|id| id.value.len() != 32)
32        {
33            return Err(Error::Invalid(
34                "content Thread differs from exact revision scope",
35            ));
36        }
37        let budget = observation::budget(&self.description)?;
38        if sources.is_empty() || sources.len() > budget.max_items as usize {
39            return Err(Error::Invalid("invalid selection count"));
40        }
41        for source in &sources {
42            match source {
43                BlobSource::Path(path) if !path.is_empty() => {}
44                BlobSource::ObjectHash(hash) if hash.len() == 32 => {}
45                _ => return Err(Error::Invalid("invalid blob source")),
46            }
47        }
48        crate::reopen::retry(|| {
49            self.read_blobs_once(thread.clone(), revision.clone(), sources.clone(), budget)
50        })
51        .await
52    }
53
54    async fn read_blobs_once(
55        &self,
56        thread: ThreadRef,
57        revision: RevisionRef,
58        sources: Vec<BlobSource>,
59        budget: ReadBudget,
60    ) -> Result<Vec<Blob>, Error> {
61        let selections = sources
62            .iter()
63            .enumerate()
64            .map(|(i, source)| ContentRead {
65                selection_id: i.to_string(),
66                selection: Some(content_read::Selection::Blob(BlobRead {
67                    source: Some(source.clone()),
68                    offset: 0,
69                    length: 0,
70                })),
71            })
72            .collect();
73        let mut messages = self
74            .api
75            .observe::<rpc::ContentServiceReadContent>(&ReadContentRequest {
76                thread: Some(thread),
77                revision: Some(revision.clone()),
78                selections,
79                budget: Some(budget),
80            })
81            .await?;
82        let mut blobs: Vec<_> = sources
83            .into_iter()
84            .map(|source| Blob {
85                source,
86                object_hash: vec![],
87                bytes: vec![],
88            })
89            .collect();
90        let mut range_done = vec![false; blobs.len()];
91        let mut complete = vec![false; blobs.len()];
92        let mut totals = vec![None; blobs.len()];
93        let mut received_bytes = 0_u64;
94        let mut received_items = 0_u32;
95        while let Some(event) = messages.next().await? {
96            let size = prost::Message::encoded_len(&event) as u64;
97            if size > u64::from(budget.max_frame_bytes)
98                || size > budget.max_snapshot_bytes.saturating_sub(received_bytes)
99                || received_items >= budget.max_items
100            {
101                return Err(Error::Invalid("content budget exceeded"));
102            }
103            received_bytes += size;
104            received_items += 1;
105            if event.revision.as_ref() != Some(&revision) {
106                return Err(Error::Invalid("content revision mismatch"));
107            }
108            let index = event
109                .selection_id
110                .parse::<usize>()
111                .map_err(|_| Error::Invalid("unknown content selection"))?;
112            if event.selection_id != index.to_string() || index >= blobs.len() || complete[index] {
113                return Err(Error::Invalid("unknown or completed content selection"));
114            }
115            let blob = &mut blobs[index];
116            match event
117                .payload
118                .ok_or(Error::Invalid("missing content payload"))?
119            {
120                content_event::Payload::Blob(chunk) => {
121                    if range_done[index]
122                        || chunk.offset != blob.bytes.len() as u64
123                        || chunk.total_size > budget.max_snapshot_bytes
124                        || chunk.data.len() as u64 > chunk.total_size.saturating_sub(chunk.offset)
125                        || chunk.object_hash.len() != 32
126                        || totals[index].is_some_and(|total| total != chunk.total_size)
127                        || (!blob.object_hash.is_empty() && blob.object_hash != chunk.object_hash)
128                        || matches!(&blob.source, BlobSource::ObjectHash(hash) if *hash != chunk.object_hash)
129                    {
130                        return Err(Error::Invalid("inconsistent blob range or identity"));
131                    }
132                    totals[index] = Some(chunk.total_size);
133                    blob.object_hash = chunk.object_hash;
134                    blob.bytes.extend(chunk.data);
135                    if chunk.range_complete && blob.bytes.len() as u64 != chunk.total_size {
136                        return Err(Error::Invalid("truncated complete blob"));
137                    }
138                    range_done[index] = chunk.range_complete;
139                }
140                content_event::Payload::SelectionComplete(status) => {
141                    if !range_done[index]
142                        || status.coverage != Coverage::Complete as i32
143                        || status.computed_for.as_ref().is_some_and(|r| r != &revision)
144                    {
145                        return Err(Error::Invalid("incomplete blob selection"));
146                    }
147                    complete[index] = true;
148                }
149                _ => return Err(Error::Invalid("unexpected content payload")),
150            }
151            if complete.iter().all(|done| *done) {
152                messages.cancel();
153                return Ok(blobs);
154            }
155        }
156        Err(Error::Interrupted)
157    }
158}
159
160/// Decode native conflict attachment bytes without inventing lifecycle evidence.
161/// Region geometry is immutable; absent retained resolution evidence stays unspecified.
162#[cfg(feature = "replication")]
163pub fn structured_conflicts(
164    bytes: &[u8],
165) -> Result<heddle_object_model::object::StructuredConflict, transport::Error> {
166    heddle_object_model::object::StructuredConflict::decode(bytes)
167        .map_err(|error| transport::Error::Io(error.to_string()))
168}
heddle_thread_api/content.rs

heddle_thread_api/
content.rs
