diff --git a/iroh-blobs/src/api/blobs.rs b/iroh-blobs/src/api/blobs.rs
index 49e44cda..7cd0b1bc 100644
--- a/iroh-blobs/src/api/blobs.rs
+++ b/iroh-blobs/src/api/blobs.rs
@@ -32,6 +32,8 @@ use range_collections::{range_set::RangeSetRange, RangeSet2};
use ref_cast::RefCast;
use serde::{Deserialize, Serialize};
use tracing::trace;
+#[cfg(test)]
+mod export_pairs;
mod reader;
pub use reader::BlobReader;
@@ -1008,6 +1010,14 @@ impl fmt::Debug for ExportBaoProgress {
}
}
+fn encode_parent_pair(pair: &(hemera::Hash, hemera::Hash)) -> [u8; 2 * hemera::OUTPUT_BYTES] {
+ let mut data = [0; 2 * hemera::OUTPUT_BYTES];
+ let (left, right) = data.split_at_mut(hemera::OUTPUT_BYTES);
+ left.copy_from_slice(pair.0.as_bytes());
+ right.copy_from_slice(pair.1.as_bytes());
+ data
+}
+
impl ExportBaoProgress {
fn new(
fut: impl Future<Output = irpc::Result<mpsc::Receiver<EncodedItem>>> + Send + 'static,
@@ -1116,9 +1126,7 @@ impl ExportBaoProgress {
target.write(&size.to_le_bytes()).await?;
}
EncodedItem::Parent(parent) => {
- let mut data = vec![0u8; 128];
- data[..64].copy_from_slice(parent.pair.0.as_bytes());
- data[64..].copy_from_slice(parent.pair.1.as_bytes());
+ let data = encode_parent_pair(&parent.pair);
target.write(&data).await?;
}
EncodedItem::Leaf(leaf) => {
@@ -1148,11 +1156,9 @@ impl ExportBaoProgress {
progress.log_other_write(8);
}
EncodedItem::Parent(parent) => {
- let mut data = [0u8; 128];
- data[..64].copy_from_slice(parent.pair.0.as_bytes());
- data[64..].copy_from_slice(parent.pair.1.as_bytes());
+ let data = encode_parent_pair(&parent.pair);
writer.send(&data).await?;
- progress.log_other_write(128);
+ progress.log_other_write(data.len());
}
EncodedItem::Leaf(leaf) => {
let len = leaf.data.len();
@@ -1175,10 +1181,8 @@ impl ExportBaoProgress {
Some(Ok(size))
}
EncodedItem::Parent(parent) => {
- let mut data = vec![0u8; 128];
- data[..64].copy_from_slice(parent.pair.0.as_bytes());
- data[64..].copy_from_slice(parent.pair.1.as_bytes());
- Some(Ok(data.into()))
+ let data = encode_parent_pair(&parent.pair);
+ Some(Ok(Bytes::copy_from_slice(&data)))
}
EncodedItem::Leaf(leaf) => Some(Ok(leaf.data)),
EncodedItem::Done => None,
diff --git a/iroh-blobs/src/api/blobs/export_pairs.rs b/iroh-blobs/src/api/blobs/export_pairs.rs
new file mode 100644
index 00000000..7c0193ee
--- /dev/null
+++ b/iroh-blobs/src/api/blobs/export_pairs.rs
@@ -0,0 +1,344 @@
+//! Exercise the actual exporter with literal pairs and an independent store proof.
+mod support;
+
+use std::io;
+
+use cyber_bao::{
+ io::{slice::decode_slice, EncodeError},
+ Poseidon2Backend, CHUNK_SIZE,
+};
+use n0_future::StreamExt;
+use support::*;
+
+use super::{EncodedItem, ExportBaoProgress};
+use crate::{
+ api::{Error, ExportBaoError},
+ provider::events::ProgressError,
+ store::{readonly_mem::ReadonlyMemStore, IROH_BLOCK_SIZE},
+};
+
+async fn chunks(export: ExportBaoProgress) -> Vec<Vec<u8>> {
+ let mut stream = export.into_byte_stream();
+ let mut chunks = Vec::new();
+ while let Some(item) = stream.next().await {
+ chunks.push(item.unwrap().to_vec());
+ }
+ chunks
+}
+
+#[test]
+fn geometry_pins() {
+ assert_eq!(geometry(), (65536, 16));
+}
+
+#[tokio::test]
+async fn literal_parent_write() {
+ let mut bytes = Vec::new();
+ scripted(literal_items())
+ .await
+ .write(&mut bytes)
+ .await
+ .unwrap();
+ assert_eq!(bytes, literal_wire());
+}
+
+#[tokio::test]
+async fn literal_parent_progress() {
+ let mut sink = Sink::default();
+ let mut progress = Progress::default();
+ scripted(literal_items())
+ .await
+ .write_with_progress(&mut sink, &mut progress, &identity(), INDEX)
+ .await
+ .unwrap();
+ assert_eq!(sink.bytes, literal_wire());
+ assert_eq!(progress.starts, [(INDEX, identity(), 100)]);
+ assert_eq!(progress.overhead, 72);
+ assert_eq!(progress.payloads, [(INDEX, OFFSET, PAYLOAD.len())]);
+}
+
+#[tokio::test]
+async fn literal_parent_byte_stream() {
+ let chunks = chunks(scripted(literal_items()).await).await;
+ assert_eq!(
+ chunks.iter().map(Vec::len).collect::<Vec<_>>(),
+ [8, 64, PAYLOAD.len()]
+ );
+ assert_eq!(chunks.concat(), literal_wire());
+}
+
+#[tokio::test]
+async fn readonly_store_right_block() {
+ let (data, root, expected) = right_fixture();
+ let (block, count) = geometry();
+ assert_eq!(expected.len(), 8 + 2 * hemera::OUTPUT_BYTES + block);
+ // Optional evidence output is only enabled by the isolated audit command.
+ if let Some(dir) = std::env::var_os("RADIO_EXPORT_PAIRS_FIXTURE_DIR") {
+ let dir = std::path::PathBuf::from(dir);
+ std::fs::write(dir.join("right-proof.bin"), &expected).unwrap();
+ std::fs::write(dir.join("right-root.txt"), root.to_string()).unwrap();
+ }
+ let store = ReadonlyMemStore::new([&data]);
+ let query = ranges(count, 2 * count);
+ let mut written = Vec::new();
+ store
+ .export_bao(root, query.clone())
+ .write(&mut written)
+ .await
+ .unwrap();
+ assert_eq!(written, expected);
+ let mut sink = Sink::default();
+ let mut progress = Progress::default();
+ store
+ .export_bao(root, query.clone())
+ .write_with_progress(&mut sink, &mut progress, &root.into(), INDEX)
+ .await
+ .unwrap();
+ assert_eq!(sink.bytes, expected);
+ assert_eq!(progress.starts, [(INDEX, root.into(), data.len() as u64)]);
+ assert_eq!(progress.overhead, 72);
+ assert_eq!(progress.payloads, [(INDEX, block as u64, block)]);
+ let chunks = chunks(store.export_bao(root, query.clone())).await;
+ assert_eq!(
+ chunks.iter().map(Vec::len).collect::<Vec<_>>(),
+ [8, 64, block]
+ );
+ assert_eq!(chunks.concat(), expected);
+ let decoded =
+ decode_slice(&Poseidon2Backend, &written, &root, &query, IROH_BLOCK_SIZE).unwrap();
+ assert_eq!(decoded, [(block as u64, data[block..].to_vec())]);
+ store.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn readonly_store_parent_free() {
+ geometry();
+ let data = body(CHUNK_SIZE);
+ let root = hemera::tree::chunk_cv(&data, 0, true);
+ let mut expected = (CHUNK_SIZE as u64).to_le_bytes().to_vec();
+ expected.extend_from_slice(&data);
+ let store = ReadonlyMemStore::new([&data]);
+ let mut written = Vec::new();
+ store
+ .export_bao(root, ranges(0, 1))
+ .write(&mut written)
+ .await
+ .unwrap();
+ assert_eq!(written, expected);
+ let mut sink = Sink::default();
+ let mut progress = Progress::default();
+ store
+ .export_bao(root, ranges(0, 1))
+ .write_with_progress(&mut sink, &mut progress, &root.into(), INDEX)
+ .await
+ .unwrap();
+ assert_eq!(sink.bytes, expected);
+ assert_eq!(progress.overhead, 8);
+ assert_eq!(progress.payloads, [(INDEX, 0, CHUNK_SIZE)]);
+ let chunks = chunks(store.export_bao(root, ranges(0, 1))).await;
+ assert_eq!(
+ chunks.iter().map(Vec::len).collect::<Vec<_>>(),
+ [8, CHUNK_SIZE]
+ );
+ assert_eq!(chunks.concat(), expected);
+ store.shutdown().await.unwrap();
+}
+
+fn check_inner(error: ExportBaoError, kind: io::ErrorKind, detail: &str) {
+ let ExportBaoError::ExportBaoInner {
+ source: EncodeError::Io(source),
+ ..
+ } = error
+ else {
+ panic!("expected encoded I/O error: {error:?}");
+ };
+ assert_eq!(source.kind(), kind);
+ assert_eq!(source.to_string(), detail);
+}
+
+fn check_sink(error: ExportBaoError) {
+ let ExportBaoError::ExportBaoIo { source, .. } = error else {
+ panic!("expected sink I/O error: {error:?}");
+ };
+ assert_eq!(source.kind(), io::ErrorKind::BrokenPipe);
+ assert_eq!(source.to_string(), "pair sink failure");
+}
+
+fn error_items() -> Vec<EncodedItem> {
+ let mut items = literal_items();
+ items.truncate(2);
+ items.push(EncodedItem::Error(EncodeError::Io(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "encoded sentinel",
+ ))));
+ items
+}
+
+#[tokio::test]
+async fn encoded_error_propagates() {
+ let mut bytes = Vec::new();
+ check_inner(
+ scripted(error_items())
+ .await
+ .write(&mut bytes)
+ .await
+ .unwrap_err(),
+ io::ErrorKind::InvalidData,
+ "encoded sentinel",
+ );
+ assert_eq!(bytes, literal_wire()[..72]);
+ let mut sink = Sink::default();
+ let mut progress = Progress::default();
+ check_inner(
+ scripted(error_items())
+ .await
+ .write_with_progress(&mut sink, &mut progress, &identity(), INDEX)
+ .await
+ .unwrap_err(),
+ io::ErrorKind::InvalidData,
+ "encoded sentinel",
+ );
+ assert_eq!(sink.bytes, bytes);
+ assert_eq!(progress.overhead, 72);
+ assert!(progress.payloads.is_empty());
+ let mut stream = scripted(error_items()).await.into_byte_stream();
+ let mut prefix = Vec::new();
+ for _ in 0..2 {
+ prefix.extend_from_slice(&stream.next().await.unwrap().unwrap());
+ }
+ let Error::Io(source) = stream.next().await.unwrap().unwrap_err();
+ assert_eq!(prefix, bytes);
+ assert_eq!(source.kind(), io::ErrorKind::InvalidData);
+ assert_eq!(source.to_string(), "encoded sentinel");
+ assert!(stream.next().await.is_none());
+}
+
+#[tokio::test]
+async fn readonly_store_missing() {
+ let store = ReadonlyMemStore::new([b"present"]);
+ let detail = "export task ended unexpectedly";
+ let mut bytes = Vec::new();
+ check_inner(
+ store
+ .export_bao(identity(), ranges(0, 1))
+ .write(&mut bytes)
+ .await
+ .unwrap_err(),
+ io::ErrorKind::UnexpectedEof,
+ detail,
+ );
+ assert!(bytes.is_empty());
+ let mut sink = Sink::default();
+ let mut progress = Progress::default();
+ check_inner(
+ store
+ .export_bao(identity(), ranges(0, 1))
+ .write_with_progress(&mut sink, &mut progress, &identity(), INDEX)
+ .await
+ .unwrap_err(),
+ io::ErrorKind::UnexpectedEof,
+ detail,
+ );
+ assert!(sink.bytes.is_empty());
+ assert!(progress.starts.is_empty());
+ assert_eq!(progress.overhead, 0);
+ assert!(progress.payloads.is_empty());
+ let mut stream = store
+ .export_bao(identity(), ranges(0, 1))
+ .into_byte_stream();
+ let Error::Io(source) = stream.next().await.unwrap().unwrap_err();
+ assert_eq!(source.kind(), io::ErrorKind::UnexpectedEof);
+ assert_eq!(source.to_string(), detail);
+ assert!(stream.next().await.is_none());
+ store.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn writer_failure_accounting() {
+ for (call, overhead) in [(0, 0), (1, 8), (2, 72)] {
+ let mut sink = Sink {
+ fail_call: Some(call),
+ ..Sink::default()
+ };
+ check_sink(
+ scripted(literal_items())
+ .await
+ .write(&mut sink)
+ .await
+ .unwrap_err(),
+ );
+ assert_eq!(sink.calls, call + 1);
+ assert_eq!(sink.bytes, literal_wire()[..overhead]);
+ let mut sink = Sink {
+ fail_call: Some(call),
+ ..Sink::default()
+ };
+ let mut progress = Progress::default();
+ check_sink(
+ scripted(literal_items())
+ .await
+ .write_with_progress(&mut sink, &mut progress, &identity(), INDEX)
+ .await
+ .unwrap_err(),
+ );
+ assert_eq!(sink.calls, call + 1);
+ assert_eq!(sink.bytes, literal_wire()[..overhead]);
+ assert_eq!(progress.overhead, overhead);
+ assert_eq!(progress.starts, [(INDEX, identity(), 100)]);
+ assert!(progress.payloads.is_empty());
+ }
+}
+
+#[tokio::test]
+async fn partial_parent_failure() {
+ for use_progress in [false, true] {
+ let mut sink = Sink {
+ fail_call: Some(1),
+ partial: 17,
+ ..Sink::default()
+ };
+ let mut progress = Progress::default();
+ let export = scripted(literal_items()).await;
+ let error = if use_progress {
+ export
+ .write_with_progress(&mut sink, &mut progress, &identity(), INDEX)
+ .await
+ .unwrap_err()
+ } else {
+ export.write(&mut sink).await.unwrap_err()
+ };
+ check_sink(error);
+ assert_eq!(sink.bytes, literal_wire()[..25]);
+ assert_eq!(sink.calls, 2);
+ assert_eq!(progress.overhead, if use_progress { 8 } else { 0 });
+ assert!(progress.payloads.is_empty());
+ }
+}
+
+#[tokio::test]
+async fn payload_progress_failure() {
+ let mut items = literal_items();
+ items.pop();
+ items.extend(literal_items().into_iter().skip(2));
+ let mut sink = Sink::default();
+ let mut progress = Progress {
+ reject_payload: true,
+ ..Progress::default()
+ };
+ let error = scripted(items)
+ .await
+ .write_with_progress(&mut sink, &mut progress, &identity(), INDEX)
+ .await
+ .unwrap_err();
+ assert!(matches!(
+ error,
+ ExportBaoError::ClientError {
+ source: ProgressError::Permission { .. },
+ ..
+ }
+ ));
+ assert_eq!(sink.bytes, literal_wire());
+ assert_eq!(sink.calls, 3);
+ assert_eq!(progress.overhead, 72);
+ assert_eq!(progress.payloads, [(INDEX, OFFSET, PAYLOAD.len())]);
+}
diff --git a/iroh-blobs/src/api/blobs/export_pairs/support.rs b/iroh-blobs/src/api/blobs/export_pairs/support.rs
new file mode 100644
index 00000000..ef368971
--- /dev/null
+++ b/iroh-blobs/src/api/blobs/export_pairs/support.rs
@@ -0,0 +1,184 @@
+use std::io;
+
+use bytes::Bytes;
+use cyber_bao::{io::Parent, ChunkNum, ChunkRanges, TreeNode, CHUNK_SIZE};
+use hemera::tree::{chunk_cv, parent_cv};
+use iroh::endpoint::VarInt;
+use iroh_io::AsyncStreamWriter;
+use irpc::channel::mpsc;
+
+use super::super::{EncodedItem, ExportBaoProgress, Leaf, WriteProgress};
+use crate::{
+ provider::events::{AbortReason, ClientResult},
+ store::IROH_BLOCK_SIZE,
+ util::SendStream,
+ Hash,
+};
+
+pub(super) const INDEX: u64 = 7;
+pub(super) const OFFSET: u64 = 19;
+pub(super) const PAYLOAD: &[u8] = b"payload";
+
+pub(super) fn identity() -> Hash {
+ hemera::Hash::from_bytes([0xa5; 32]).into()
+}
+
+pub(super) fn literal_items() -> Vec<EncodedItem> {
+ vec![
+ EncodedItem::Size(100),
+ EncodedItem::Parent(Parent {
+ node: TreeNode::new(1),
+ pair: (
+ hemera::Hash::from_bytes(core::array::from_fn(|i| i as u8)),
+ hemera::Hash::from_bytes(core::array::from_fn(|i| (32 + i) as u8)),
+ ),
+ }),
+ EncodedItem::Leaf(Leaf {
+ offset: OFFSET,
+ data: Bytes::from_static(PAYLOAD),
+ }),
+ EncodedItem::Done,
+ ]
+}
+
+pub(super) fn literal_wire() -> Vec<u8> {
+ let mut wire = 100u64.to_le_bytes().to_vec();
+ wire.extend(0u8..64);
+ wire.extend_from_slice(PAYLOAD);
+ wire
+}
+
+pub(super) async fn scripted(items: Vec<EncodedItem>) -> ExportBaoProgress {
+ let (tx, rx) = mpsc::channel(items.len());
+ for item in items {
+ tx.send(item).await.unwrap();
+ }
+ drop(tx);
+ ExportBaoProgress::new(async move { Ok(rx) })
+}
+
+pub(super) fn geometry() -> (usize, usize) {
+ assert_eq!(CHUNK_SIZE, 4096, "pinned Hemera chunk geometry");
+ let bytes = IROH_BLOCK_SIZE.bytes();
+ assert_eq!(bytes, 65536, "pinned provider block geometry");
+ (bytes, bytes / CHUNK_SIZE)
+}
+
+pub(super) fn body(len: usize) -> Vec<u8> {
+ (0..len)
+ .map(|i| ((i * 31 + i / CHUNK_SIZE) % 256) as u8)
+ .collect()
+}
+
+pub(super) fn ranges(start: usize, end: usize) -> ChunkRanges {
+ ChunkRanges::from(ChunkNum(start as u64)..ChunkNum(end as u64))
+}
+
+// Four explicit levels, independent of radio's outboard/tree traversal.
+fn half_root(leaves: &[hemera::Hash]) -> hemera::Hash {
+ assert_eq!(leaves.len(), 16);
+ let p: [_; 8] = core::array::from_fn(|i| parent_cv(&leaves[2 * i], &leaves[2 * i + 1], false));
+ let q: [_; 4] = core::array::from_fn(|i| parent_cv(&p[2 * i], &p[2 * i + 1], false));
+ let r: [_; 2] = core::array::from_fn(|i| parent_cv(&q[2 * i], &q[2 * i + 1], false));
+ parent_cv(&r[0], &r[1], false)
+}
+
+pub(super) fn right_fixture() -> (Vec<u8>, hemera::Hash, Vec<u8>) {
+ let (block, chunks) = geometry();
+ let data = body(2 * block);
+ let leaves: Vec<_> = data
+ .chunks_exact(CHUNK_SIZE)
+ .enumerate()
+ .map(|(i, bytes)| chunk_cv(bytes, i as u64, false))
+ .collect();
+ let left = half_root(&leaves[..chunks]);
+ let right = half_root(&leaves[chunks..]);
+ let root = parent_cv(&left, &right, true);
+ let mut wire = (data.len() as u64).to_le_bytes().to_vec();
+ wire.extend_from_slice(left.as_bytes());
+ wire.extend_from_slice(right.as_bytes());
+ wire.extend_from_slice(&data[block..]);
+ (data, root, wire)
+}
+
+#[derive(Debug, Default)]
+pub(super) struct Sink {
+ pub bytes: Vec<u8>,
+ pub calls: usize,
+ pub fail_call: Option<usize>,
+ pub partial: usize,
+}
+
+impl Sink {
+ fn accept(&mut self, data: &[u8]) -> io::Result<()> {
+ let call = self.calls;
+ self.calls += 1;
+ if self.fail_call == Some(call) {
+ self.bytes
+ .extend_from_slice(&data[..self.partial.min(data.len())]);
+ return Err(io::Error::new(
+ io::ErrorKind::BrokenPipe,
+ "pair sink failure",
+ ));
+ }
+ self.bytes.extend_from_slice(data);
+ Ok(())
+ }
+}
+
+impl AsyncStreamWriter for Sink {
+ async fn write(&mut self, data: &[u8]) -> io::Result<()> {
+ self.accept(data)
+ }
+ async fn write_bytes(&mut self, data: Bytes) -> io::Result<()> {
+ self.accept(&data)
+ }
+ async fn sync(&mut self) -> io::Result<()> {
+ panic!("exporter unexpectedly synced")
+ }
+}
+
+impl SendStream for Sink {
+ async fn send_bytes(&mut self, bytes: Bytes) -> io::Result<()> {
+ self.accept(&bytes)
+ }
+ async fn send(&mut self, bytes: &[u8]) -> io::Result<()> {
+ self.accept(bytes)
+ }
+ async fn sync(&mut self) -> io::Result<()> {
+ panic!("exporter unexpectedly synced")
+ }
+ fn reset(&mut self, _: VarInt) -> io::Result<()> {
+ panic!("exporter unexpectedly reset")
+ }
+ async fn stopped(&mut self) -> io::Result<Option<VarInt>> {
+ panic!("exporter unexpectedly queried stop")
+ }
+ fn id(&self) -> u64 {
+ panic!("exporter unexpectedly queried id")
+ }
+}
+
+#[derive(Debug, Default)]
+pub(super) struct Progress {
+ pub starts: Vec<(u64, Hash, u64)>,
+ pub overhead: usize,
+ pub payloads: Vec<(u64, u64, usize)>,
+ pub reject_payload: bool,
+}
+
+impl WriteProgress for Progress {
+ async fn notify_payload_write(&mut self, index: u64, offset: u64, len: usize) -> ClientResult {
+ self.payloads.push((index, offset, len));
+ if self.reject_payload {
+ return Err(AbortReason::Permission.into());
+ }
+ Ok(())
+ }
+ fn log_other_write(&mut self, len: usize) {
+ self.overhead += len;
+ }
+ async fn send_transfer_started(&mut self, index: u64, hash: &Hash, size: u64) {
+ self.starts.push((index, *hash, size));
+ }
+}