From 8ff02610e45912bbf4c54ce9021e022e2ff55f9f Mon Sep 17 00:00:00 2001 From: Lewis Date: Mon, 1 Jun 2026 19:01:04 +0300 Subject: [PATCH] feat(store): inline-commit mode, committed-extent recovery, failsafe verify&rollback Lewis: May this revision serve well! --- .../src/blockstore/group_commit.rs | 345 +++++++++++++++--- .../tranquil-store/src/blockstore/repair.rs | 4 +- crates/tranquil-store/src/blockstore/store.rs | 192 +++++++--- 3 files changed, 432 insertions(+), 109 deletions(-) diff --git a/crates/tranquil-store/src/blockstore/group_commit.rs b/crates/tranquil-store/src/blockstore/group_commit.rs index c728512..ee9fab5 100644 --- a/crates/tranquil-store/src/blockstore/group_commit.rs +++ b/crates/tranquil-store/src/blockstore/group_commit.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU32, Ordering}; use std::thread; -use parking_lot::RwLock; +use parking_lot::{Mutex, RwLock}; use crate::clock::{Clock, LogicalNanos}; use crate::fsync_order::PostBlockstoreHook; @@ -218,6 +218,7 @@ pub struct GroupCommitConfig { pub checkpoint_interval_ms: u64, pub checkpoint_write_threshold: u64, pub verify_persisted_blocks: bool, + pub synchronous: bool, } impl Default for GroupCommitConfig { @@ -228,6 +229,7 @@ impl Default for GroupCommitConfig { checkpoint_interval_ms: 60_000, checkpoint_write_threshold: 100_000, verify_persisted_blocks: false, + synchronous: false, } } } @@ -259,9 +261,50 @@ fn log_thread_panic(payload: Box, context: &str) { tracing::error!(panic = msg, "{context}"); } -struct SingleShardWriter { - sender: flume::Sender, - handle: Option>, +trait InlineHandler: Send { + fn handle(&mut self, req: CommitRequest); + fn finalize(&mut self); +} + +struct InlineState { + manager: DataFileManager, + index: Arc, + config: GroupCommitConfig, + state: ActiveState, + ctx: ShardContext, + post_sync_hook: Option>, + last_checkpoint: LogicalNanos, + writes_since_checkpoint: u64, +} + +impl InlineHandler for InlineState { + fn handle(&mut self, req: CommitRequest) { + handle_request( + req, + &self.manager, + &self.index, + &self.config, + &mut self.state, + self.post_sync_hook.as_deref(), + &self.ctx, + &mut self.last_checkpoint, + &mut self.writes_since_checkpoint, + ); + } + + fn finalize(&mut self) { + shutdown_checkpoint(&self.index, &self.ctx.epoch, &self.ctx.hint_positions); + } +} + +enum SingleShardWriter { + Threaded { + sender: flume::Sender, + handle: Option>, + }, + Inline { + handler: Mutex>, + }, } impl SingleShardWriter { @@ -278,6 +321,23 @@ impl SingleShardWriter { ctx.hint_positions .update(ctx.shard_id, state.file_id, state.hint_position); + if config.synchronous { + let last_checkpoint = ctx.clock.monotonic(); + let inline = InlineState { + manager, + index, + config, + state, + ctx, + post_sync_hook, + last_checkpoint, + writes_since_checkpoint: 0, + }; + return Ok(Self::Inline { + handler: Mutex::new(Box::new(inline)), + }); + } + let (sender, receiver) = flume::bounded(config.channel_capacity); let handle = thread::Builder::new() @@ -295,29 +355,48 @@ impl SingleShardWriter { }) .map_err(|e| CommitError::from(io::Error::other(e)))?; - Ok(Self { + Ok(Self::Threaded { sender, handle: Some(handle), }) } + fn submit(&self, request: CommitRequest) -> Result<(), CommitError> { + match self { + Self::Threaded { sender, .. } => { + sender.send(request).map_err(|_| CommitError::ChannelClosed) + } + Self::Inline { handler } => { + handler.lock().handle(request); + Ok(()) + } + } + } + fn shutdown(&mut self) { - let _ = self.sender.send(CommitRequest::Shutdown); - if let Some(handle) = self.handle.take() - && let Err(payload) = handle.join() - { - log_thread_panic(payload, "group commit thread panicked"); + match self { + Self::Threaded { sender, handle } => { + let _ = sender.send(CommitRequest::Shutdown); + if let Some(handle) = handle.take() + && let Err(payload) = handle.join() + { + log_thread_panic(payload, "group commit thread panicked"); + } + } + Self::Inline { handler } => handler.lock().finalize(), } } } impl Drop for SingleShardWriter { fn drop(&mut self) { - let _ = self.sender.try_send(CommitRequest::Shutdown); - if let Some(handle) = self.handle.take() - && let Err(payload) = handle.join() - { - log_thread_panic(payload, "group commit thread panicked during drop"); + if let Self::Threaded { sender, handle } = self { + let _ = sender.try_send(CommitRequest::Shutdown); + if let Some(handle) = handle.take() + && let Err(payload) = handle.join() + { + log_thread_panic(payload, "group commit thread panicked during drop"); + } } } } @@ -478,19 +557,49 @@ impl GroupCommitWriter { &self.epoch } - pub fn sender_round_robin(&self) -> &flume::Sender { - let idx = self - .round_robin + fn round_robin_index(&self) -> usize { + self.round_robin .fetch_add(1, Ordering::Relaxed) - .wrapping_rem(self.shard_count) as usize; - &self.shards[idx].sender + .wrapping_rem(self.shard_count) as usize + } + + fn shard_index_for(&self, request: &CommitRequest) -> usize { + match request { + CommitRequest::PutBlocks { blocks, .. } => { + pick_shard_for_blocks(blocks, self.shard_count) + } + CommitRequest::ApplyCommit { + blocks, + deleted_cids, + .. + } => pick_shard_for_apply(blocks, deleted_cids, self.shard_count), + CommitRequest::Compact { .. } + | CommitRequest::RepairLeaked { .. } + | CommitRequest::RepairBlocks { .. } + | CommitRequest::Quiesce { .. } + | CommitRequest::Shutdown => self.round_robin_index(), + } + } + + pub fn submit_blocking(&self, request: CommitRequest) -> Result<(), CommitError> { + let idx = self.shard_index_for(&request); + self.shards[idx].submit(request) + } + + fn threaded_sender(shard: &SingleShardWriter) -> &flume::Sender { + match shard { + SingleShardWriter::Threaded { sender, .. } => sender, + SingleShardWriter::Inline { .. } => unreachable!( + "async commit path requires threaded group-commit; synchronous mode is gauntlet-blocking-only" + ), + } } pub fn sender_for_blocks( &self, blocks: &[([u8; CID_SIZE], Vec)], ) -> &flume::Sender { - &self.shards[pick_shard_for_blocks(blocks, self.shard_count)].sender + Self::threaded_sender(&self.shards[pick_shard_for_blocks(blocks, self.shard_count)]) } pub fn sender_for_apply( @@ -498,7 +607,9 @@ impl GroupCommitWriter { blocks: &[([u8; CID_SIZE], Vec)], deleted_cids: &[[u8; CID_SIZE]], ) -> &flume::Sender { - &self.shards[pick_shard_for_apply(blocks, deleted_cids, self.shard_count)].sender + Self::threaded_sender( + &self.shards[pick_shard_for_apply(blocks, deleted_cids, self.shard_count)], + ) } pub fn quiesce_all( @@ -510,13 +621,18 @@ impl GroupCommitWriter { .map(|shard| { let (response_tx, response_rx) = tokio::sync::oneshot::channel(); let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); - shard - .sender - .send(CommitRequest::Quiesce { - response: response_tx, - resume: resume_rx, - }) - .map_err(|_| CommitError::ChannelClosed)?; + let req = CommitRequest::Quiesce { + response: response_tx, + resume: resume_rx, + }; + match shard { + SingleShardWriter::Threaded { sender, .. } => { + sender.send(req).map_err(|_| CommitError::ChannelClosed)?; + } + SingleShardWriter::Inline { handler } => { + handler.lock().handle(req); + } + } Ok((response_rx, resume_tx)) }) .collect(); @@ -567,10 +683,13 @@ impl GroupCommitWriter { impl Drop for GroupCommitWriter { fn drop(&mut self) { self.shards.iter_mut().for_each(|s| { - let _ = s.sender.try_send(CommitRequest::Shutdown); + if let SingleShardWriter::Threaded { sender, .. } = s { + let _ = sender.try_send(CommitRequest::Shutdown); + } }); self.shards.iter_mut().for_each(|s| { - if let Some(handle) = s.handle.take() + if let SingleShardWriter::Threaded { handle, .. } = s + && let Some(handle) = handle.take() && let Err(payload) = handle.join() { log_thread_panic(payload, "group commit thread panicked during drop"); @@ -1105,6 +1224,95 @@ fn commit_loop( } } +#[allow(clippy::too_many_arguments)] +fn handle_request( + req: CommitRequest, + manager: &DataFileManager, + index: &BlockIndex, + config: &GroupCommitConfig, + state: &mut ActiveState, + post_sync_hook: Option<&dyn PostBlockstoreHook>, + ctx: &ShardContext, + last_checkpoint: &mut LogicalNanos, + writes_since_checkpoint: &mut u64, +) { + let epoch = &ctx.epoch; + match classify_request(req) { + ClassifyResult::Shutdown => {} + ClassifyResult::Compact { + file_id, + grace_period_ms, + response, + } => { + let result = compaction::compact_on_writer_thread( + manager, + index, + file_id, + epoch.current(), + grace_period_ms, + &ctx.file_ids, + &ctx.active_files, + &ctx.hint_positions, + epoch, + ctx.clock.wall_millis(), + ); + let _ = response.send(result); + } + ClassifyResult::Repair { + leaked_cids, + response, + } => { + let repaired = index.repair_leaked_refcounts( + &leaked_cids, + epoch.current(), + ctx.clock.wall_millis(), + ); + let _ = response.send(Ok(repaired)); + } + ClassifyResult::RepairBlocks { blocks, response } => { + let result = compaction::repair_blocks_on_writer_thread( + manager, + index, + &blocks, + epoch.current(), + &ctx.file_ids, + &ctx.hint_positions, + epoch, + ); + let _ = response.send(result); + } + ClassifyResult::Quiesce { response, resume } => { + let snapshot = capture_snapshot(manager, state, epoch); + let _ = response.send(snapshot); + drop(resume); + } + ClassifyResult::Batch(entry) => { + let entries = vec![entry]; + let result = process_batch(manager, index, &entries, state, ctx); + if let Ok((ref _dedup, ref proof)) = result { + run_post_sync_hook(post_sync_hook, proof); + } + if let Err(ref e) = result { + tracing::warn!(error = %e, "commit batch failed"); + } + if let Ok((ref dedup, _)) = result { + *writes_since_checkpoint = + writes_since_checkpoint.saturating_add(dedup.len() as u64); + } + dispatch_responses(entries, result.map(|(dedup, _proof)| dedup)); + maybe_checkpoint( + index, + epoch, + config, + &ctx.clock, + last_checkpoint, + writes_since_checkpoint, + &ctx.hint_positions, + ); + } + } +} + fn run_post_sync_hook(hook: Option<&dyn PostBlockstoreHook>, proof: &BlocksSynced) { if let Some(hook) = hook && let Err(e) = hook.on_blocks_synced(proof) @@ -1211,17 +1419,32 @@ fn verify_persisted_blocks( by_file.into_iter().try_for_each(|(file_id, locations)| { let path = manager.data_file_path(file_id); - let fd = match manager.io().open(&path, OpenOptions::read_only_existing()) { - Ok(fd) => fd, - Err(_) => return Ok(()), - }; - let file_size = match manager.io().file_size(fd) { - Ok(s) => s, - Err(_) => { - let _ = manager.io().close(fd); - return Ok(()); + let probe_offset = locations[0].1.offset; + let fd = match (0..VERIFY_RETRY_ATTEMPTS).find_map(|_| { + manager + .io() + .open(&path, OpenOptions::read_only_existing()) + .ok() + }) { + Some(fd) => fd, + None => { + return Err(CommitError::VerifyFailed { + file_id, + offset: probe_offset, + }); } }; + let file_size = + match (0..VERIFY_RETRY_ATTEMPTS).find_map(|_| manager.io().file_size(fd).ok()) { + Some(s) => s, + None => { + let _ = manager.io().close(fd); + return Err(CommitError::VerifyFailed { + file_id, + offset: probe_offset, + }); + } + }; let result = locations.into_iter().try_for_each(|(expected_cid, loc)| { verify_block_at(manager, fd, file_size, expected_cid, loc) }); @@ -1234,6 +1457,7 @@ fn verify_persisted_blocks( enum VerifyOutcome { NoFaultDetected, Faulted, + Inconclusive, } fn verify_block_at( @@ -1243,15 +1467,16 @@ fn verify_block_at( expected_cid: &[u8; CID_SIZE], loc: BlockLocation, ) -> Result<(), CommitError> { - let passed = (0..VERIFY_RETRY_ATTEMPTS).any(|_| { - matches!( - verify_once(manager, fd, file_size, expected_cid, loc), - VerifyOutcome::NoFaultDetected - ) + let verdict = (0..VERIFY_RETRY_ATTEMPTS).find_map(|_| { + match verify_once(manager, fd, file_size, expected_cid, loc) { + VerifyOutcome::NoFaultDetected => Some(true), + VerifyOutcome::Faulted => Some(false), + VerifyOutcome::Inconclusive => None, + } }); - match passed { - true => Ok(()), - false => Err(CommitError::VerifyFailed { + match verdict { + Some(true) => Ok(()), + Some(false) | None => Err(CommitError::VerifyFailed { file_id: loc.file_id, offset: loc.offset, }), @@ -1286,23 +1511,33 @@ fn verify_once( ); VerifyOutcome::Faulted } - Err(_) => VerifyOutcome::NoFaultDetected, + Err(_) => VerifyOutcome::Inconclusive, } } const VERIFY_RETRY_ATTEMPTS: u32 = 4; +const ROLLBACK_TRUNCATE_ATTEMPTS: u32 = 8; + +fn truncate_and_sync_durably(io: &S, fd: FileId, offset: u64) { + let ok = (0..ROLLBACK_TRUNCATE_ATTEMPTS) + .any(|_| io.truncate(fd, offset).and_then(|()| io.sync(fd)).is_ok()); + if !ok { + tracing::error!( + offset, + "rollback could not durably truncate uncommitted tail; recovery may resurrect it" + ); + } +} + fn rollback_batch( manager: &DataFileManager, state: &ActiveState, rotations: &[RotationState], ) { - let _ = manager.io().truncate(state.fd, state.position.raw()); - let _ = manager.io().sync(state.fd); - let _ = manager - .io() - .truncate(state.hint_fd, state.hint_position.raw()); - let _ = manager.io().sync(state.hint_fd); + truncate_and_sync_durably(manager.io(), state.fd, state.position.raw()); + truncate_and_sync_durably(manager.io(), state.hint_fd, state.hint_position.raw()); + let _ = manager.io().barrier(); rotations.iter().for_each(|rot| { manager.rollback_rotation(rot.file_id); let _ = manager.io().close(rot.hint_fd); diff --git a/crates/tranquil-store/src/blockstore/repair.rs b/crates/tranquil-store/src/blockstore/repair.rs index 19fc63c..d05351e 100644 --- a/crates/tranquil-store/src/blockstore/repair.rs +++ b/crates/tranquil-store/src/blockstore/repair.rs @@ -100,7 +100,9 @@ where let nodes_repaired = batch_by_bytes(to_repair) .into_iter() - .try_fold(0u64, |acc, batch| store.repair_blocks(batch).map(|n| acc + n)) + .try_fold(0u64, |acc, batch| { + store.repair_blocks(batch).map(|n| acc + n) + }) .map_err(|e| rebuild_err("repair_blocks", e))?; Ok(RepairOutcome { diff --git a/crates/tranquil-store/src/blockstore/store.rs b/crates/tranquil-store/src/blockstore/store.rs index 678efc0..eb4525e 100644 --- a/crates/tranquil-store/src/blockstore/store.rs +++ b/crates/tranquil-store/src/blockstore/store.rs @@ -123,6 +123,7 @@ pub struct TranquilBlockStore { epoch: EpochCounter, data_dir: PathBuf, clock: C, + synchronous: bool, } impl Clone for TranquilBlockStore { @@ -134,6 +135,7 @@ impl Clone for TranquilBlockStor epoch: self.epoch.clone(), data_dir: self.data_dir.clone(), clock: self.clock.clone(), + synchronous: self.synchronous, } } } @@ -299,6 +301,7 @@ impl TranquilBlockStore { let checkpoint_epoch = index.loaded_checkpoint_epoch(); let checkpoint_positions = index.loaded_checkpoint_positions(); + let synchronous = config.group_commit.synchronous; let writer = GroupCommitWriter::spawn_sharded( make_manager, Arc::clone(&index), @@ -331,6 +334,7 @@ impl TranquilBlockStore { epoch, data_dir, clock, + synchronous, }) } @@ -388,8 +392,15 @@ impl TranquilBlockStore { }) .is_ok(); - let result = - Self::scan_and_index(io, index, fd, file_id, start_offset, indexed_end, hint_exists); + let result = Self::scan_and_index( + io, + index, + fd, + file_id, + start_offset, + indexed_end, + hint_exists, + ); let _ = io.close(fd); @@ -462,7 +473,11 @@ impl TranquilBlockStore { RepoError::storage(e) })?; - let keep_end = last_valid_end.max(indexed_end); + let committed_end = BlockOffset::new(start_offset.raw().max(indexed_end.raw())); + let keep_end = match hint_exists { + true => committed_end, + false => last_valid_end.max(indexed_end), + }; if file_size > keep_end.raw() { tracing::info!( @@ -470,26 +485,30 @@ impl TranquilBlockStore { truncating_from = keep_end.raw(), file_size, scanned_count = scanned_entries.len(), - "truncating partial/unacked tail" + "truncating uncommitted tail" ); io.truncate(fd, keep_end.raw()) .map_err(RepoError::storage)?; io.sync(fd).map_err(RepoError::storage)?; } - if !scanned_entries.is_empty() { + let indexable: Vec<_> = scanned_entries + .into_iter() + .filter(|(_, loc)| loc.offset.raw() < keep_end.raw()) + .collect(); + if !indexable.is_empty() { tracing::info!( file_id = %file_id, - scanned = scanned_entries.len(), + scanned = indexable.len(), hint_exists, - "reindexing blocks past hint coverage" + "reindexing blocks within committed extent" ); let cursor = super::types::WriteCursor { file_id, offset: keep_end, }; index - .batch_put_if_absent(&scanned_entries, cursor) + .batch_put_if_absent(&indexable, cursor) .map_err(|e| RepoError::storage(io::Error::other(e.to_string())))?; } @@ -536,16 +555,15 @@ impl TranquilBlockStore { grace_period_ms: u64, ) -> Result { let (tx, rx) = tokio::sync::oneshot::channel(); - let sender = self - .writer - .with(|w| w.sender_round_robin().clone()) - .map_err(|_| CompactionError::ChannelClosed)?; - sender - .send(CommitRequest::Compact { - file_id, - grace_period_ms, - response: tx, + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::Compact { + file_id, + grace_period_ms, + response: tx, + }) }) + .map_err(|_| CompactionError::ChannelClosed)? .map_err(|_| CompactionError::ChannelClosed)?; let result = rx .blocking_recv() @@ -595,16 +613,15 @@ impl TranquilBlockStore { leaked_cids: &[(super::types::CidBytes, super::types::RefCount)], ) -> Result { let (tx, rx) = tokio::sync::oneshot::channel(); - let sender = self - .writer - .with(|w| w.sender_round_robin().clone()) - .map_err(commit_error_to_repo)?; - sender - .send(CommitRequest::RepairLeaked { - leaked_cids: leaked_cids.to_vec(), - response: tx, + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::RepairLeaked { + leaked_cids: leaked_cids.to_vec(), + response: tx, + }) }) - .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))?; + .map_err(commit_error_to_repo)? + .map_err(commit_error_to_repo)?; rx.blocking_recv() .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))? .map_err(commit_error_to_repo) @@ -618,15 +635,14 @@ impl TranquilBlockStore { return Ok(0); } let (tx, rx) = tokio::sync::oneshot::channel(); - let sender = self - .writer - .with(|w| w.sender_round_robin().clone()) - .map_err(|_| CompactionError::ChannelClosed)?; - sender - .send(CommitRequest::RepairBlocks { - blocks, - response: tx, + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::RepairBlocks { + blocks, + response: tx, + }) }) + .map_err(|_| CompactionError::ChannelClosed)? .map_err(|_| CompactionError::ChannelClosed)?; rx.blocking_recv() .map_err(|_| CompactionError::ChannelClosed)? @@ -663,17 +679,16 @@ impl TranquilBlockStore { if blocks.is_empty() { return Ok(()); } - let sender = self - .writer - .with(|w| w.sender_for_blocks(&blocks).clone()) - .map_err(commit_error_to_repo)?; let (tx, rx) = tokio::sync::oneshot::channel(); - sender - .send(CommitRequest::PutBlocks { - blocks, - response: tx, + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::PutBlocks { + blocks, + response: tx, + }) }) - .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))?; + .map_err(commit_error_to_repo)? + .map_err(commit_error_to_repo)?; rx.blocking_recv() .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))? .map_err(commit_error_to_repo)?; @@ -685,18 +700,17 @@ impl TranquilBlockStore { blocks: Vec<([u8; CID_SIZE], Vec)>, deleted_cids: Vec<[u8; CID_SIZE]>, ) -> Result<(), RepoError> { - let sender = self - .writer - .with(|w| w.sender_for_apply(&blocks, &deleted_cids).clone()) - .map_err(commit_error_to_repo)?; let (tx, rx) = tokio::sync::oneshot::channel(); - sender - .send(CommitRequest::ApplyCommit { - blocks, - deleted_cids, - response: tx, + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::ApplyCommit { + blocks, + deleted_cids, + response: tx, + }) }) - .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))?; + .map_err(commit_error_to_repo)? + .map_err(commit_error_to_repo)?; rx.blocking_recv() .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))? .map_err(commit_error_to_repo) @@ -706,6 +720,22 @@ impl TranquilBlockStore { &self, blocks: Vec<([u8; CID_SIZE], Vec)>, ) -> Result, RepoError> { + if self.synchronous { + let (tx, mut rx) = tokio::sync::oneshot::channel(); + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::PutBlocks { + blocks, + response: tx, + }) + }) + .map_err(commit_error_to_repo)? + .map_err(commit_error_to_repo)?; + return rx + .try_recv() + .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))? + .map_err(commit_error_to_repo); + } let sender = self .writer .with(|w| w.sender_for_blocks(&blocks).clone()) @@ -728,6 +758,23 @@ impl TranquilBlockStore { blocks: Vec<([u8; CID_SIZE], Vec)>, deleted_cids: Vec<[u8; CID_SIZE]>, ) -> Result<(), RepoError> { + if self.synchronous { + let (tx, mut rx) = tokio::sync::oneshot::channel(); + self.writer + .with(|w| { + w.submit_blocking(CommitRequest::ApplyCommit { + blocks, + deleted_cids, + response: tx, + }) + }) + .map_err(commit_error_to_repo)? + .map_err(commit_error_to_repo)?; + return rx + .try_recv() + .map_err(|_| commit_error_to_repo(CommitError::ChannelClosed))? + .map_err(commit_error_to_repo); + } let sender = self .writer .with(|w| w.sender_for_apply(&blocks, &deleted_cids).clone()) @@ -750,6 +797,9 @@ impl TranquilBlockStore { impl BlockStore for TranquilBlockStore { async fn get(&self, cid: &Cid) -> Result, RepoError> { let cid_bytes = cid_to_bytes(cid)?; + if self.synchronous { + return self.reader.get(&cid_bytes).map_err(read_error_to_repo); + } let reader = Arc::clone(&self.reader); tokio::task::spawn_blocking(move || reader.get(&cid_bytes)) .await @@ -767,6 +817,9 @@ impl BlockStore for TranquilBloc async fn has(&self, cid: &Cid) -> Result { let cid_bytes = cid_to_bytes(cid)?; + if self.synchronous { + return self.reader.has(&cid_bytes).map_err(read_error_to_repo); + } let reader = Arc::clone(&self.reader); tokio::task::spawn_blocking(move || reader.has(&cid_bytes)) .await @@ -797,6 +850,9 @@ impl BlockStore for TranquilBloc .iter() .map(cid_to_bytes) .collect::, _>>()?; + if self.synchronous { + return self.reader.get_many(&cid_bytes).map_err(read_error_to_repo); + } let reader = Arc::clone(&self.reader); tokio::task::spawn_blocking(move || reader.get_many(&cid_bytes)) .await @@ -1068,6 +1124,36 @@ mod tests { } } + #[tokio::test] + async fn synchronous_store_serves_async_block_store_trait() { + use jacquard_repo::storage::BlockStore; + + let tmp = tempfile::TempDir::new().unwrap(); + let mut cfg = BlockStoreConfig::new(tmp.path().join("data"), tmp.path().join("index")); + cfg.group_commit.synchronous = true; + let store = TranquilBlockStore::open(cfg).unwrap(); + + let payload = b"sync-trait-roundtrip".to_vec(); + let cid = crate::blockstore::hash_to_cid(&payload); + + store + .put_many(vec![(cid, bytes::Bytes::from(payload.clone()))]) + .await + .expect("async put_many must not panic on a synchronous store"); + + let got = store.get(&cid).await.expect("async get must succeed"); + assert_eq!( + got.as_deref(), + Some(payload.as_slice()), + "synchronous store must round-trip blocks through the async trait" + ); + + store + .decrement_refs(&[cid]) + .await + .expect("async decrement_refs must not panic on a synchronous store"); + } + fn instant_policy(max_attempts: u8) -> OpenRetryPolicy { OpenRetryPolicy { max_attempts: NonZeroU8::new(max_attempts).expect("max_attempts must be nonzero"),