use api::v2::client::RpcTransport;
pub use crate::contract::blob_read::Source as BlobSource;
use crate::{
Remote,
contract::*,
observation::{self, Error},
rpc, transport,
};
pub struct Blob {
pub source: BlobSource,
pub object_hash: Vec<u8>,
pub bytes: Vec<u8>,
}
impl<T: RpcTransport<Error = transport::Error>> Remote<T> {
pub async fn read_blobs(
&self,
thread: ThreadRef,
revision: RevisionRef,
sources: Vec<BlobSource>,
) -> Result<Vec<Blob>, Error> {
if thread.spool != revision.spool
|| thread.id.as_ref().is_none_or(|id| id.value.len() != 32)
{
return Err(Error::Invalid(
"content Thread differs from exact revision scope",
));
}
let budget = observation::budget(&self.description)?;
if sources.is_empty() || sources.len() > budget.max_items as usize {
return Err(Error::Invalid("invalid selection count"));
}
for source in &sources {
match source {
BlobSource::Path(path) if !path.is_empty() => {}
BlobSource::ObjectHash(hash) if hash.len() == 32 => {}
_ => return Err(Error::Invalid("invalid blob source")),
}
}
crate::reopen::retry(|| {
self.read_blobs_once(thread.clone(), revision.clone(), sources.clone(), budget)
})
.await
}
async fn read_blobs_once(
&self,
thread: ThreadRef,
revision: RevisionRef,
sources: Vec<BlobSource>,
budget: ReadBudget,
) -> Result<Vec<Blob>, Error> {
let selections = sources
.iter()
.enumerate()
.map(|(i, source)| ContentRead {
selection_id: i.to_string(),
selection: Some(content_read::Selection::Blob(BlobRead {
source: Some(source.clone()),
offset: 0,
length: 0,
})),
})
.collect();
let mut messages = self
.api
.observe::<rpc::ContentServiceReadContent>(&ReadContentRequest {
thread: Some(thread),
revision: Some(revision.clone()),
selections,
budget: Some(budget),
})
.await?;
let mut blobs: Vec<_> = sources
.into_iter()
.map(|source| Blob {
source,
object_hash: vec![],
bytes: vec![],
})
.collect();
let mut range_done = vec![false; blobs.len()];
let mut complete = vec![false; blobs.len()];
let mut totals = vec![None; blobs.len()];
let mut accepted = None;
let mut received_bytes = 0_u64;
let mut received_items = 0_u32;
while let Some(event) = messages.next().await? {
let echo = matches!(
event.payload,
Some(content_event::Payload::AcceptedBudget(_))
);
if let Some(content_event::Payload::AcceptedBudget(effective)) = &event.payload {
if accepted.is_some() || !event.selection_id.is_empty() {
return Err(Error::Invalid(
"duplicate or selection-scoped content budget echo",
));
}
if event
.revision
.as_ref()
.is_some_and(|echo_revision| echo_revision != &revision)
{
return Err(Error::Invalid("content revision mismatch"));
}
api::v2::validate_accepted_read_budget(&budget, Some(effective))
.map_err(|_| Error::Invalid("invalid accepted content budget"))?;
accepted = Some(*effective);
}
let limits = accepted.ok_or(Error::Invalid("missing initial content budget echo"))?;
let size = prost::Message::encoded_len(&event) as u64 + 5;
if size > u64::from(limits.max_frame_bytes)
|| size > limits.max_snapshot_bytes.saturating_sub(received_bytes)
|| received_items >= limits.max_items
{
return Err(Error::Invalid("content budget exceeded"));
}
received_bytes += size;
received_items += 1;
if echo {
continue;
}
if event.revision.as_ref() != Some(&revision) {
return Err(Error::Invalid("content revision mismatch"));
}
let index = event
.selection_id
.parse::<usize>()
.map_err(|_| Error::Invalid("unknown content selection"))?;
if event.selection_id != index.to_string() || index >= blobs.len() || complete[index] {
return Err(Error::Invalid("unknown or completed content selection"));
}
let blob = &mut blobs[index];
match event
.payload
.ok_or(Error::Invalid("missing content payload"))?
{
content_event::Payload::Blob(chunk) => {
if range_done[index]
|| chunk.offset != blob.bytes.len() as u64
|| chunk.total_size > limits.max_snapshot_bytes
|| chunk.data.len() as u64 > chunk.total_size.saturating_sub(chunk.offset)
|| chunk.object_hash.len() != 32
|| totals[index].is_some_and(|total| total != chunk.total_size)
|| (!blob.object_hash.is_empty() && blob.object_hash != chunk.object_hash)
|| matches!(&blob.source, BlobSource::ObjectHash(hash) if *hash != chunk.object_hash)
{
return Err(Error::Invalid("inconsistent blob range or identity"));
}
totals[index] = Some(chunk.total_size);
blob.object_hash = chunk.object_hash;
blob.bytes.extend(chunk.data);
if chunk.range_complete && blob.bytes.len() as u64 != chunk.total_size {
return Err(Error::Invalid("truncated complete blob"));
}
range_done[index] = chunk.range_complete;
}
content_event::Payload::SelectionComplete(status) => {
if !range_done[index]
|| status.coverage != Coverage::Complete as i32
|| status.computed_for.as_ref().is_some_and(|r| r != &revision)
{
return Err(Error::Invalid("incomplete blob selection"));
}
complete[index] = true;
}
_ => return Err(Error::Invalid("unexpected content payload")),
}
}
if accepted.is_none() {
return Err(Error::Invalid("missing initial content budget echo"));
}
if complete.iter().all(|done| *done) {
Ok(blobs)
} else {
Err(Error::Interrupted)
}
}
}
#[cfg(feature = "replication")]
pub fn structured_conflicts(
bytes: &[u8],
) -> Result<heddle_object_model::object::StructuredConflict, transport::Error> {
heddle_object_model::object::StructuredConflict::decode(bytes)
.map_err(|error| transport::Error::Io(error.to_string()))
}