mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-07-21 23:43:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(store): inline-commit mode, committed-extent recovery, failsafe verify&rollback

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-06-01 19:01:04 +03:00
parent 22f82489d5
commit 8ff02610e4
3 changed files with 432 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-store/src/blockstore/group_commit.rs
+290 -55
View File
@@ -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<dyn std::any::Any + Send>, context: &str) {
tracing::error!(panic = msg, "{context}");
}
struct SingleShardWriter {
sender: flume::Sender<CommitRequest>,
handle: Option<thread::JoinHandle<()>>,
trait InlineHandler: Send {
fn handle(&mut self, req: CommitRequest);
fn finalize(&mut self);
}
struct InlineState<S: StorageIO + 'static, C: Clock> {
manager: DataFileManager<S>,
index: Arc<BlockIndex>,
config: GroupCommitConfig,
state: ActiveState,
ctx: ShardContext<C>,
post_sync_hook: Option<Arc<dyn PostBlockstoreHook>>,
last_checkpoint: LogicalNanos,
writes_since_checkpoint: u64,
}
impl<S: StorageIO + 'static, C: Clock> InlineHandler for InlineState<S, C> {
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<CommitRequest>,
handle: Option<thread::JoinHandle<()>>,
},
Inline {
handler: Mutex<Box<dyn InlineHandler>>,
},
}
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<CommitRequest> {
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<CommitRequest> {
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<u8>)],
) -> &flume::Sender<CommitRequest> {
&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<u8>)],
deleted_cids: &[[u8; CID_SIZE]],
) -> &flume::Sender<CommitRequest> {
&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<S: StorageIO, C: Clock>(
}
}
#[allow(clippy::too_many_arguments)]
fn handle_request<S: StorageIO, C: Clock>(
req: CommitRequest,
manager: &DataFileManager<S>,
index: &BlockIndex,
config: &GroupCommitConfig,
state: &mut ActiveState,
post_sync_hook: Option<&dyn PostBlockstoreHook>,
ctx: &ShardContext<C>,
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<S: StorageIO>(
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<S: StorageIO>(
enum VerifyOutcome {
NoFaultDetected,
Faulted,
Inconclusive,
}
fn verify_block_at<S: StorageIO>(
@@ -1243,15 +1467,16 @@ fn verify_block_at<S: StorageIO>(
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<S: StorageIO>(
);
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<S: StorageIO>(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<S: StorageIO>(
manager: &DataFileManager<S>,
state: &ActiveState,
rotations: &[RotationState<S>],
) {
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);
crates/tranquil-store/src/blockstore/repair.rs
+3 -1
View File
@@ -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 {
crates/tranquil-store/src/blockstore/store.rs
+139 -53
View File
@@ -123,6 +123,7 @@ pub struct TranquilBlockStore<S: StorageIO + Send + Sync + 'static, C: Clock> {
epoch: EpochCounter,
data_dir: PathBuf,
clock: C,
synchronous: bool,
}
impl<S: StorageIO + Send + Sync + 'static, C: Clock> Clone for TranquilBlockStore<S, C> {
@@ -134,6 +135,7 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> 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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
epoch,
data_dir,
clock,
synchronous,
})
}
@@ -388,8 +392,15 @@ impl<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
})
.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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
grace_period_ms: u64,
) -> Result<CompactionResult, CompactionError> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
leaked_cids: &[(super::types::CidBytes, super::types::RefCount)],
) -> Result<u64, RepoError> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
blocks: Vec<([u8; CID_SIZE], Vec<u8>)>,
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
&self,
blocks: Vec<([u8; CID_SIZE], Vec<u8>)>,
) -> Result<Vec<BlockLocation>, 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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
blocks: Vec<([u8; CID_SIZE], Vec<u8>)>,
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<S: StorageIO + Send + Sync + 'static, C: Clock> TranquilBlockStore<S, C> {
impl<S: StorageIO + Send + Sync + 'static, C: Clock> BlockStore for TranquilBlockStore<S, C> {
async fn get(&self, cid: &Cid) -> Result<Option<Bytes>, 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<S: StorageIO + Send + Sync + 'static, C: Clock> BlockStore for TranquilBloc
async fn has(&self, cid: &Cid) -> Result<bool, RepoError> {
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<S: StorageIO + Send + Sync + 'static, C: Clock> BlockStore for TranquilBloc
.iter()
.map(cid_to_bytes)
.collect::<Result<Vec<_>, _>>()?;
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"),