use std::{
collections::BTreeMap,
future::Future,
io,
ops::Deref,
sync::atomic::{AtomicU64, Ordering},
};
use anyhow::{anyhow, Context, Result};
use bao_tree::{
io::{fsm, mixed::EncodedItem, BaoContentItem, Leaf},
BaoTree, ChunkNum, ChunkRanges, TreeNode,
};
use bytes::Bytes;
use iroh_blobs::{
api::{
self,
blobs::{Bitfield, BlobStatus, ExportProgressItem},
proto::{
BatchResponse, BlobDeleteRequest, BlobStatusRequest, Command, CreateTagRequest,
DeleteTagsRequest, ExportBaoRequest, ExportPathRequest, ExportRangesItem,
ExportRangesRequest, ImportBaoRequest, ImportByteStreamUpdate, ImportBytesRequest,
ListTagsRequest, RenameTagRequest, Scope, SetTagRequest,
},
tags::TagInfo,
TempTag,
},
protocol::ChunkRangesExt,
store::{mem::MemStore, IROH_BLOCK_SIZE},
Hash, HashAndFormat,
};
use iroh_io::{AsyncSliceReader, TokioStreamReader, TokioStreamWriter};
use irpc::channel::mpsc;
use range_collections::range_set::RangeSetRange;
use crate::wasm_docs_persistence::JsReplicaStore;
const STORE_COMMAND_CAPACITY: usize = 32;
const BAO_IMPORT_CHUNK_BYTES: usize = 1024;
const LOCAL_BLOB_CHUNK_BYTES: usize = 16 * 1024;
const STAGING_CHUNK_BYTES: usize = 1024 * 1024;
static NEXT_STAGING_SESSION: AtomicU64 = AtomicU64::new(1);
#[derive(Debug, Clone)]
pub(crate) struct IndexedDbBlobStore {
inner: MemStore,
}
impl IndexedDbBlobStore {
pub(crate) async fn open(adapter: JsReplicaStore) -> Result<Self> {
for manifest in adapter.blob_manifests().await? {
if manifest.size == 0
|| manifest.chunk_bytes == 0
|| manifest.chunk_count == 0
|| manifest.size.div_ceil(u64::from(manifest.chunk_bytes))
!= u64::from(manifest.chunk_count)
{
return Err(anyhow!(
"persistent browser blob {} has an invalid manifest",
manifest.content_hash
));
}
if !manifest.bao_ready {
let hash: Hash = manifest
.content_hash
.parse()
.context("parsing persistent browser blob hash")?;
build_outboard(&adapter, hash, manifest.size).await?;
}
}
let (sender, receiver) = tokio::sync::mpsc::channel(STORE_COMMAND_CAPACITY);
n0_future::task::spawn(Actor::new(adapter, receiver).run());
Ok(Self {
inner: MemStore::from_sender(sender.into()),
})
}
}
impl Deref for IndexedDbBlobStore {
type Target = iroh_blobs::api::Store;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl AsRef<iroh_blobs::api::Store> for IndexedDbBlobStore {
fn as_ref(&self) -> &iroh_blobs::api::Store {
self
}
}
impl From<IndexedDbBlobStore> for iroh_blobs::api::Store {
fn from(value: IndexedDbBlobStore) -> Self {
value.inner.into()
}
}
struct Actor {
adapter: JsReplicaStore,
commands: tokio::sync::mpsc::Receiver<Command>,
tasks: n0_future::task::JoinSet<()>,
idle_waiters: Vec<irpc::channel::oneshot::Sender<()>>,
tags: BTreeMap<iroh_blobs::api::Tag, HashAndFormat>,
next_scope: AtomicU64,
}
impl Actor {
fn new(adapter: JsReplicaStore, commands: tokio::sync::mpsc::Receiver<Command>) -> Self {
Self {
adapter,
commands,
tasks: n0_future::task::JoinSet::new(),
idle_waiters: Vec::new(),
tags: BTreeMap::new(),
next_scope: AtomicU64::new(1),
}
}
async fn run(mut self) {
loop {
tokio::select! {
command = self.commands.recv() => {
let Some(command) = command else {
break;
};
if self.handle(command).await {
break;
}
}
Some(result) = self.tasks.join_next(), if !self.tasks.is_empty() => {
let _ = result;
if self.tasks.is_empty() {
for waiter in self.idle_waiters.drain(..) {
waiter.send(()).await.ok();
}
}
}
}
}
}
fn spawn(&mut self, future: impl Future<Output = ()> + 'static) {
self.tasks.spawn(future);
}
async fn handle(&mut self, command: Command) -> bool {
match command {
Command::WaitIdle(command) => {
if self.tasks.is_empty() {
command.tx.send(()).await.ok();
} else {
self.idle_waiters.push(command.tx);
}
}
Command::SyncDb(command) => {
let result = self.adapter.flush().await.map_err(api_error);
command.tx.send(result).await.ok();
}
Command::Shutdown(command) => {
command.tx.send(()).await.ok();
return true;
}
Command::BlobStatus(command) => {
let BlobStatusRequest { hash } = command.inner;
let status = self.blob_status(hash).await.unwrap_or(BlobStatus::NotFound);
command.tx.send(status).await.ok();
}
Command::ListBlobs(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
match adapter.blob_manifests().await {
Ok(manifests) => {
for manifest in manifests {
match manifest.content_hash.parse::<Hash>() {
Ok(hash) => {
if command.tx.send(Ok(hash)).await.is_err() {
break;
}
}
Err(error) => {
command.tx.send(Err(api_error(error.into()))).await.ok();
break;
}
}
}
}
Err(error) => {
command.tx.send(Err(api_error(error))).await.ok();
}
}
});
}
Command::ImportBao(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
import_bao(adapter, command).await;
});
}
Command::ExportBao(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
export_bao(adapter, command).await;
});
}
Command::ImportBytes(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
import_bytes(adapter, command).await;
});
}
Command::ImportByteStream(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
import_byte_stream(adapter, command).await;
});
}
Command::Observe(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
let content_hash = command.inner.hash.to_string();
let mut previous = Bitfield::empty();
let mut first = true;
loop {
let bitfield = match adapter.blob_resume_state(&content_hash).await {
Ok(Some(state)) => bitfield_for_state(&state),
Ok(None) | Err(_) => Bitfield::empty(),
};
let complete = bitfield.is_complete();
if first || bitfield != previous {
first = false;
previous = bitfield.clone();
if command.tx.send(bitfield).await.is_err() {
break;
}
}
if complete {
break;
}
gloo_timers::future::TimeoutFuture::new(100).await;
}
});
}
Command::ExportRanges(command) => {
let adapter = self.adapter.clone();
self.spawn(async move {
export_ranges(adapter, command).await;
});
}
Command::ImportPath(command) => {
command
.tx
.send(api::proto::AddProgressItem::Error(unsupported(
"path import is unavailable in browser builds",
)))
.await
.ok();
}
Command::ExportPath(command) => {
let ExportPathRequest { .. } = command.inner;
command
.tx
.send(ExportProgressItem::Error(api_error(anyhow!(
"path export is unavailable in browser builds"
))))
.await
.ok();
}
Command::DeleteBlobs(command) => {
let BlobDeleteRequest { hashes, force } = command.inner;
let adapter = self.adapter.clone();
self.spawn(async move {
let result = async {
for hash in hashes {
adapter.delete_blob(&hash.to_string(), force).await?;
}
Result::<()>::Ok(())
}
.await
.map_err(api_error);
command.tx.send(result).await.ok();
});
}
Command::ClearProtected(command) => {
command.tx.send(Ok(())).await.ok();
}
Command::Batch(command) => {
let scope = Scope::GLOBAL;
command.tx.send(scope).await.ok();
self.spawn(async move {
let mut receiver = command.rx;
while let Ok(Some(message)) = receiver.recv().await {
if matches!(message, BatchResponse::Ping) {
continue;
}
}
});
}
Command::CreateTempTag(command) => {
command
.tx
.send(TempTag::new(command.inner.value, None))
.await
.ok();
}
Command::ListTempTags(command) => {
command.tx.send(Vec::new()).await.ok();
}
Command::ListTags(command) => {
let ListTagsRequest {
from,
to,
raw,
hash_seq,
} = command.inner;
let tags = self
.tags
.iter()
.filter(|(tag, value)| {
from.as_ref().map_or(true, |from| *tag >= from)
&& to.as_ref().map_or(true, |to| *tag < to)
&& ((raw && value.format.is_raw())
|| (hash_seq && value.format.is_hash_seq()))
})
.map(|(name, value)| {
Ok(TagInfo::new(
name.as_ref(),
HashAndFormat {
hash: value.hash,
format: value.format,
},
))
})
.collect();
command.tx.send(tags).await.ok();
}
Command::SetTag(command) => {
let SetTagRequest { name, value } = command.inner;
self.tags.insert(name, value);
command.tx.send(Ok(())).await.ok();
}
Command::CreateTag(command) => {
let CreateTagRequest { value } = command.inner;
let name = iroh_blobs::api::Tag::from(
format!("wasm-{}", self.next_scope.fetch_add(1, Ordering::Relaxed)).as_bytes(),
);
self.tags.insert(name.clone(), value);
command.tx.send(Ok(name)).await.ok();
}
Command::RenameTag(command) => {
let RenameTagRequest { from, to } = command.inner;
let result = self
.tags
.remove(&from)
.map(|value| {
self.tags.insert(to, value);
})
.ok_or_else(|| api_error(anyhow!("tag not found")));
command.tx.send(result).await.ok();
}
Command::DeleteTags(command) => {
let DeleteTagsRequest { from, to } = command.inner;
let before = self.tags.len();
self.tags.retain(|tag, _| {
!from.as_ref().map_or(true, |from| tag >= from)
|| !to.as_ref().map_or(true, |to| tag < to)
});
command
.tx
.send(Ok((before - self.tags.len()) as u64))
.await
.ok();
}
}
false
}
async fn blob_status(&self, hash: Hash) -> Result<BlobStatus> {
blob_status(&self.adapter, hash).await
}
}
async fn blob_status(adapter: &JsReplicaStore, hash: Hash) -> Result<BlobStatus> {
if hash == Hash::EMPTY {
return Ok(BlobStatus::Complete { size: 0 });
}
Ok(match adapter.blob_resume_state(&hash.to_string()).await? {
Some(state) if state.complete => BlobStatus::Complete {
size: state.expected_size,
},
Some(state) => BlobStatus::Partial {
size: bitfield_for_state(&state).validated_size(),
},
None => BlobStatus::NotFound,
})
}
fn bitfield_for_state(state: &crate::wasm_docs_persistence::PersistentBlobResumeState) -> Bitfield {
if state.complete {
return Bitfield::complete(state.expected_size);
}
let mut ranges = ChunkRanges::empty();
for chunk_index in &state.received_chunks {
let start = u64::from(*chunk_index) * u64::from(state.chunk_bytes);
let end = (start + u64::from(state.chunk_bytes)).min(state.expected_size);
ranges |= ChunkRanges::from(ChunkNum::full_chunks(start)..ChunkNum::chunks(end));
}
Bitfield::new(ranges, state.expected_size)
}
async fn import_bao(adapter: JsReplicaStore, mut command: api::proto::ImportBaoMsg) {
let ImportBaoRequest { hash, size } = command.inner;
let result = async {
if hash == Hash::EMPTY && size.get() == 0 {
while command.rx.recv().await?.is_some() {}
return Ok(());
}
let tree = BaoTree::new(size.get(), IROH_BLOCK_SIZE);
adapter
.begin_blob_import(&hash.to_string(), size.get(), BAO_IMPORT_CHUNK_BYTES)
.await?;
while let Some(item) = command.rx.recv().await? {
match item {
BaoContentItem::Parent(parent) => {
let mut pair = [0u8; 64];
pair[..32].copy_from_slice(parent.pair.0.as_bytes());
pair[32..].copy_from_slice(parent.pair.1.as_bytes());
let offset = tree
.pre_order_offset(parent.node)
.ok_or_else(|| anyhow!("invalid Bao parent node"))?;
adapter
.write_blob_outboard_node(&hash.to_string(), offset, &pair)
.await?;
}
BaoContentItem::Leaf(leaf) => {
adapter
.write_blob_range(
&hash.to_string(),
size.get(),
BAO_IMPORT_CHUNK_BYTES,
leaf.offset,
&leaf.data,
)
.await?;
}
}
}
if adapter.finalize_blob_if_complete(&hash.to_string()).await? {
build_outboard(&adapter, hash, size.get()).await?;
}
Result::<()>::Ok(())
}
.await
.map_err(api_error);
command.tx.send(result).await.ok();
}
async fn export_bao(adapter: JsReplicaStore, command: api::proto::ExportBaoMsg) {
let ExportBaoRequest { hash, ranges } = command.inner;
if hash == Hash::EMPTY {
command.tx.send(EncodedItem::Size(0)).await.ok();
command.tx.send(EncodedItem::Done).await.ok();
return;
}
let Some(manifest) = adapter.blob_manifests().await.ok().and_then(|manifests| {
manifests
.into_iter()
.find(|manifest| manifest.content_hash == hash.to_string())
}) else {
command
.tx
.send(EncodedItem::Error(bao_tree::io::EncodeError::Io(
io::Error::new(io::ErrorKind::NotFound, "blob not found"),
)))
.await
.ok();
return;
};
if !manifest.bao_ready {
if let Err(error) = build_outboard(&adapter, hash, manifest.size).await {
command
.tx
.send(EncodedItem::Error(bao_tree::io::EncodeError::Io(
io::Error::other(error),
)))
.await
.ok();
return;
}
}
command.tx.send(EncodedItem::Size(manifest.size)).await.ok();
let tree = BaoTree::new(manifest.size, IROH_BLOCK_SIZE);
let data = IndexedDbDataReader {
adapter: adapter.clone(),
hash,
size: manifest.size,
};
let outboard = IndexedDbOutboard {
adapter,
hash,
tree,
};
if let Err(error) = encode_to_items(data, outboard, ranges, command.tx.clone()).await {
command.tx.send(EncodedItem::Error(error)).await.ok();
return;
}
command.tx.send(EncodedItem::Done).await.ok();
}
async fn export_ranges(adapter: JsReplicaStore, command: api::proto::ExportRangesMsg) {
let ExportRangesRequest { hash, ranges } = command.inner;
if hash == Hash::EMPTY {
command.tx.send(ExportRangesItem::Size(0)).await.ok();
return;
}
let result = async {
let state = adapter
.blob_resume_state(&hash.to_string())
.await?
.ok_or_else(|| anyhow!("blob not found"))?;
command
.tx
.send(ExportRangesItem::Size(state.expected_size))
.await?;
let present = bitfield_for_state(&state);
for range in ranges.iter() {
let (start, end) = match range {
RangeSetRange::Range(range) => (
(*range.start).min(state.expected_size),
(*range.end).min(state.expected_size),
),
RangeSetRange::RangeFrom(range) => {
((*range.start).min(state.expected_size), state.expected_size)
}
};
let requested = ChunkRanges::bytes(start..end);
if !present.ranges.is_superset(&requested) {
return Err(anyhow!("requested blob range is not locally complete"));
}
let mut offset = start;
while offset < end {
let len = usize::try_from((end - offset).min(1024))
.context("converting IndexedDB range size")?;
let bytes = adapter
.read_blob_range(&hash.to_string(), offset, len)
.await?
.ok_or_else(|| anyhow!("blob range disappeared during export"))?;
command
.tx
.send(ExportRangesItem::Data(Leaf {
offset,
data: Bytes::from(bytes),
}))
.await?;
offset += len as u64;
}
}
Result::<()>::Ok(())
}
.await;
if let Err(error) = result {
command
.tx
.send(ExportRangesItem::Error(api_error(error)))
.await
.ok();
}
}
async fn encode_to_items(
data: IndexedDbDataReader,
outboard: IndexedDbOutboard,
ranges: ChunkRanges,
output: mpsc::Sender<EncodedItem>,
) -> std::result::Result<(), bao_tree::io::EncodeError> {
let tree = outboard.tree;
let hash = outboard.hash;
let (writer, reader) = tokio::io::duplex(LOCAL_BLOB_CHUNK_BYTES * 2);
let encode_ranges = ranges.clone();
let encoder = async move {
fsm::encode_ranges_validated(data, outboard, &encode_ranges, TokioStreamWriter(writer))
.await
};
let decoder = async move {
let mut decoder =
fsm::ResponseDecoder::new(hash.into(), ranges, tree, TokioStreamReader(reader));
loop {
match decoder.next().await {
fsm::ResponseDecoderNext::More((next, item)) => {
let item = match item.map_err(|error| {
bao_tree::io::EncodeError::Io(io::Error::new(
io::ErrorKind::InvalidData,
error.to_string(),
))
})? {
BaoContentItem::Parent(parent) => EncodedItem::Parent(parent),
BaoContentItem::Leaf(leaf) => EncodedItem::Leaf(leaf),
};
output.send(item).await.map_err(|_| {
bao_tree::io::EncodeError::Io(io::ErrorKind::BrokenPipe.into())
})?;
decoder = next;
}
fsm::ResponseDecoderNext::Done(_) => break,
}
}
Ok(())
};
let (encoded, decoded) = tokio::join!(encoder, decoder);
encoded?;
decoded
}
async fn import_bytes(adapter: JsReplicaStore, command: api::proto::ImportBytesMsg) {
let ImportBytesRequest {
data,
format,
scope: _,
} = command.inner;
let size = data.len() as u64;
command
.tx
.send(api::proto::AddProgressItem::Size(size))
.await
.ok();
command
.tx
.send(api::proto::AddProgressItem::CopyDone)
.await
.ok();
let hash = Hash::new(&data);
let result = persist_complete_bytes(&adapter, hash, data).await;
match result {
Ok(()) => {
command
.tx
.send(api::proto::AddProgressItem::Done(TempTag::new(
HashAndFormat { hash, format },
None,
)))
.await
.ok();
}
Err(error) => {
command
.tx
.send(api::proto::AddProgressItem::Error(unsupported(&format!(
"{error:#}"
))))
.await
.ok();
}
}
}
async fn import_byte_stream(adapter: JsReplicaStore, mut command: api::proto::ImportByteStreamMsg) {
let session_id = format!(
"iroh-blob-{}-{}",
js_sys::Date::now().to_bits(),
NEXT_STAGING_SESSION.fetch_add(1, Ordering::Relaxed)
);
let result = async {
let mut hasher = blake3::Hasher::new();
let mut buffered = Vec::with_capacity(STAGING_CHUNK_BYTES);
let mut size = 0u64;
let mut chunk_index = 0u32;
loop {
match command.rx.recv().await? {
Some(ImportByteStreamUpdate::Bytes(mut bytes)) => {
size = size
.checked_add(bytes.len() as u64)
.ok_or_else(|| anyhow!("streamed blob size overflow"))?;
hasher.update(&bytes);
while !bytes.is_empty() {
let take = (STAGING_CHUNK_BYTES - buffered.len()).min(bytes.len());
buffered.extend_from_slice(&bytes.split_to(take));
if buffered.len() == STAGING_CHUNK_BYTES {
adapter
.write_blob_staging_chunk(&session_id, chunk_index, &buffered)
.await?;
buffered.clear();
chunk_index = chunk_index
.checked_add(1)
.ok_or_else(|| anyhow!("streamed blob chunk count overflow"))?;
}
}
command
.tx
.send(api::proto::AddProgressItem::CopyProgress(size))
.await?;
}
Some(ImportByteStreamUpdate::Done) => break,
None => bail_stream("blob byte stream ended before Done")?,
}
}
if size == 0 {
command
.tx
.send(api::proto::AddProgressItem::Size(0))
.await?;
command
.tx
.send(api::proto::AddProgressItem::CopyDone)
.await?;
command
.tx
.send(api::proto::AddProgressItem::Done(TempTag::leaking_empty(
command.inner.format,
)))
.await?;
return Result::<()>::Ok(());
}
if !buffered.is_empty() {
adapter
.write_blob_staging_chunk(&session_id, chunk_index, &buffered)
.await?;
}
let hash = Hash::from(hasher.finalize());
command
.tx
.send(api::proto::AddProgressItem::Size(size))
.await?;
command
.tx
.send(api::proto::AddProgressItem::CopyDone)
.await?;
adapter
.finalize_blob_staging(&session_id, &hash.to_string(), size, STAGING_CHUNK_BYTES)
.await?;
build_outboard(&adapter, hash, size).await?;
command
.tx
.send(api::proto::AddProgressItem::Done(TempTag::new(
HashAndFormat {
hash,
format: command.inner.format,
},
None,
)))
.await?;
Ok(())
}
.await;
if let Err(error) = result {
adapter.abort_blob_staging(&session_id).await.ok();
command
.tx
.send(api::proto::AddProgressItem::Error(unsupported(&format!(
"{error:#}"
))))
.await
.ok();
}
}
async fn persist_complete_bytes(adapter: &JsReplicaStore, hash: Hash, data: Bytes) -> Result<()> {
if data.is_empty() {
return Ok(());
}
let size = data.len() as u64;
adapter
.begin_blob_import(&hash.to_string(), size, LOCAL_BLOB_CHUNK_BYTES)
.await?;
for (chunk_index, chunk) in data.chunks(LOCAL_BLOB_CHUNK_BYTES).enumerate() {
adapter
.write_blob_chunk(
&hash.to_string(),
size,
LOCAL_BLOB_CHUNK_BYTES,
chunk_index as u32,
chunk,
)
.await?;
}
adapter
.finalize_blob_import(&hash.to_string(), size, LOCAL_BLOB_CHUNK_BYTES)
.await?;
build_outboard(adapter, hash, size).await
}
async fn build_outboard(adapter: &JsReplicaStore, hash: Hash, size: u64) -> Result<()> {
let data = IndexedDbDataReader {
adapter: adapter.clone(),
hash,
size,
};
let writer = IndexedDbOutboardWriter {
adapter: adapter.clone(),
hash,
};
let tree = BaoTree::new(size, IROH_BLOCK_SIZE);
let mut outboard = bao_tree::io::outboard::PreOrderOutboard {
root: hash.into(),
tree,
data: writer,
};
use bao_tree::io::fsm::CreateOutboard;
outboard
.init_from(std::io::Cursor::new(data))
.await
.context("building persistent Bao outboard")?;
if Hash::from(outboard.root) != hash {
return Err(anyhow!("persistent Bao root does not match blob hash"));
}
adapter.mark_blob_bao_ready(&hash.to_string()).await
}
#[derive(Clone)]
struct IndexedDbDataReader {
adapter: JsReplicaStore,
hash: Hash,
size: u64,
}
impl AsyncSliceReader for IndexedDbDataReader {
async fn read_at(&mut self, offset: u64, len: usize) -> io::Result<Bytes> {
self.adapter
.read_blob_range(&self.hash.to_string(), offset, len)
.await
.map_err(io::Error::other)?
.map(Bytes::from)
.ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "blob range unavailable"))
}
async fn size(&mut self) -> io::Result<u64> {
Ok(self.size)
}
}
struct IndexedDbOutboard {
adapter: JsReplicaStore,
hash: Hash,
tree: BaoTree,
}
impl fsm::Outboard for IndexedDbOutboard {
fn root(&self) -> bao_tree::blake3::Hash {
self.hash.into()
}
fn tree(&self) -> BaoTree {
self.tree
}
async fn load(
&mut self,
node: TreeNode,
) -> io::Result<Option<(bao_tree::blake3::Hash, bao_tree::blake3::Hash)>> {
let pair = self
.adapter
.read_blob_outboard_node(
&self.hash.to_string(),
self.tree.pre_order_offset(node).ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidData, "invalid Bao node")
})?,
)
.await
.map_err(io::Error::other)?;
Ok(pair.map(|pair| {
(
<[u8; 32]>::try_from(&pair[..32]).unwrap().into(),
<[u8; 32]>::try_from(&pair[32..]).unwrap().into(),
)
}))
}
}
struct IndexedDbOutboardWriter {
adapter: JsReplicaStore,
hash: Hash,
}
impl iroh_io::AsyncSliceWriter for IndexedDbOutboardWriter {
async fn write_at(&mut self, offset: u64, data: &[u8]) -> io::Result<()> {
if offset % 64 != 0 || data.len() != 64 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Bao outboard writes must contain one aligned hash pair",
));
}
self.adapter
.write_blob_outboard_node(
&self.hash.to_string(),
offset / 64,
data.try_into().unwrap(),
)
.await
.map_err(io::Error::other)
}
async fn write_bytes_at(&mut self, offset: u64, data: Bytes) -> io::Result<()> {
self.write_at(offset, &data).await
}
async fn set_len(&mut self, _len: u64) -> io::Result<()> {
Ok(())
}
async fn sync(&mut self) -> io::Result<()> {
Ok(())
}
}
fn api_error(error: anyhow::Error) -> api::Error {
api::Error::other(error)
}
fn unsupported(message: &str) -> io::Error {
io::Error::new(io::ErrorKind::Unsupported, message.to_owned())
}
fn bail_stream<T>(message: &str) -> Result<T> {
Err(anyhow!(message.to_owned()))
}