use std::io;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::config::IoBackend;
use crate::shard::{Shard, TestSyncClock};
fn remaining_until(deadline: std::time::Instant, now: std::time::Instant) -> Option<Duration> {
deadline
.checked_duration_since(now)
.filter(|remaining| !remaining.is_zero())
}
pub static DATA_ATTEMPTS: AtomicU64 = AtomicU64::new(0);
pub static DATA_SUCCESSES: AtomicU64 = AtomicU64::new(0);
pub static TAG_ATTEMPTS: AtomicU64 = AtomicU64::new(0);
pub static TAG_SUCCESSES: AtomicU64 = AtomicU64::new(0);
pub static FIXED_ATTEMPTS: AtomicU64 = AtomicU64::new(0);
pub static FIXED_SUCCESSES: AtomicU64 = AtomicU64::new(0);
pub static FAIL_NEXT_DATA_SYNC: AtomicBool = AtomicBool::new(false);
pub static FAIL_NEXT_TAG_SYNC: AtomicBool = AtomicBool::new(false);
pub static FAIL_NEXT_FORCE_WRITE: AtomicBool = AtomicBool::new(false);
pub static FAIL_NEXT_ENCRYPTED_DATA_WRITE_AFTER_PWRITE: AtomicBool = AtomicBool::new(false);
#[derive(Default)]
pub(crate) struct SyncGate {
state: Mutex<(bool, bool)>,
cv: std::sync::Condvar,
#[cfg(test)]
wake_state: Mutex<TestWakeState>,
#[cfg(test)]
wake_cv: std::sync::Condvar,
}
#[cfg(test)]
#[derive(Default)]
struct TestWakeState {
waiter_ready: bool,
requested: u64,
observed: u64,
}
impl SyncGate {
pub(crate) fn wait_until_blocked(&self) {
self.wait_until_blocked_for(SYNC_GATE_TIMEOUT);
}
fn wait_until_blocked_for(&self, timeout: Duration) {
let deadline = std::time::Instant::now() + timeout;
let mut state = self.state.lock().expect("sync gate poisoned");
loop {
if state.0 {
return;
}
let remaining = remaining_until(deadline, std::time::Instant::now())
.unwrap_or_else(|| panic!("data sync gate was never reached within {timeout:?}"));
#[cfg(test)]
self.mark_test_waiter_ready();
let (next, _) = self
.cv
.wait_timeout(state, remaining)
.expect("sync gate poisoned");
#[cfg(test)]
self.acknowledge_test_wake();
state = next;
}
}
pub(crate) fn block(&self) {
self.block_for(SYNC_GATE_TIMEOUT);
}
fn block_for(&self, timeout: Duration) {
let deadline = std::time::Instant::now() + timeout;
let mut state = self.state.lock().expect("sync gate poisoned");
state.0 = true;
self.cv.notify_all();
loop {
if state.1 {
return;
}
let remaining = remaining_until(deadline, std::time::Instant::now())
.unwrap_or_else(|| panic!("data sync gate was not released within {timeout:?}"));
#[cfg(test)]
self.mark_test_waiter_ready();
let (next, _) = self
.cv
.wait_timeout(state, remaining)
.expect("sync gate poisoned");
#[cfg(test)]
self.acknowledge_test_wake();
state = next;
}
}
pub(crate) fn open(&self) {
let mut state = self.state.lock().expect("sync gate poisoned");
state.1 = true;
self.cv.notify_all();
}
#[cfg(test)]
fn wait_until_blocked_for_test(&self, timeout: Duration) {
self.wait_until_blocked_for(timeout);
}
#[cfg(test)]
fn wake_for_test(&self) {
let deadline = std::time::Instant::now() + Duration::from_secs(1);
let mut wake = self.wake_state.lock().expect("sync gate poisoned");
while !wake.waiter_ready {
let remaining = remaining_until(deadline, std::time::Instant::now())
.expect("test waiter never entered the condition-variable wait");
let (next, timed) = self
.wake_cv
.wait_timeout(wake, remaining)
.expect("sync gate poisoned");
assert!(
!timed.timed_out() || next.waiter_ready,
"test waiter never entered the condition-variable wait"
);
wake = next;
}
wake.requested += 1;
let requested = wake.requested;
drop(wake);
let state = self.state.lock().expect("sync gate poisoned");
self.cv.notify_all();
drop(state);
let mut wake = self.wake_state.lock().expect("sync gate poisoned");
while wake.observed < requested {
let remaining = remaining_until(deadline, std::time::Instant::now())
.expect("test waiter did not observe the spurious wake");
let (next, timed) = self
.wake_cv
.wait_timeout(wake, remaining)
.expect("sync gate poisoned");
assert!(
!timed.timed_out() || next.observed >= requested,
"test waiter did not observe the spurious wake"
);
wake = next;
}
}
#[cfg(test)]
fn mark_blocked_for_test(&self) {
let mut state = self.state.lock().expect("sync gate poisoned");
state.0 = true;
self.cv.notify_all();
}
#[cfg(test)]
fn mark_test_waiter_ready(&self) {
let mut wake = self.wake_state.lock().expect("sync gate poisoned");
wake.waiter_ready = true;
self.wake_cv.notify_all();
}
#[cfg(test)]
fn acknowledge_test_wake(&self) {
let mut wake = self.wake_state.lock().expect("sync gate poisoned");
if wake.observed < wake.requested {
wake.observed = wake.requested;
self.wake_cv.notify_all();
}
}
}
static DATA_SYNC_GATE: Mutex<Option<Arc<SyncGate>>> = Mutex::new(None);
static BARRIER_PUBLICATION_GATE: Mutex<Option<PublicationGate>> = Mutex::new(None);
const SYNC_GATE_TIMEOUT: Duration = Duration::from_secs(5);
pub(crate) struct DataSyncGateGuard {
gate: Arc<SyncGate>,
}
impl DataSyncGateGuard {
pub(crate) fn wait_until_blocked(&self) {
self.gate.wait_until_blocked();
}
pub(crate) fn release(&self) {
release_gate(&self.gate);
}
#[cfg(test)]
fn gate(&self) -> Arc<SyncGate> {
Arc::clone(&self.gate)
}
}
impl Drop for DataSyncGateGuard {
fn drop(&mut self) {
self.release();
}
}
fn release_gate(gate: &Arc<SyncGate>) {
let mut active = DATA_SYNC_GATE.lock().expect("sync gate poisoned");
if active
.as_ref()
.is_some_and(|current| Arc::ptr_eq(current, gate))
{
*active = None;
}
drop(active);
gate.open();
}
pub fn reset() {
DATA_ATTEMPTS.store(0, Ordering::SeqCst);
DATA_SUCCESSES.store(0, Ordering::SeqCst);
TAG_ATTEMPTS.store(0, Ordering::SeqCst);
TAG_SUCCESSES.store(0, Ordering::SeqCst);
FIXED_ATTEMPTS.store(0, Ordering::SeqCst);
FIXED_SUCCESSES.store(0, Ordering::SeqCst);
FAIL_NEXT_DATA_SYNC.store(false, Ordering::SeqCst);
FAIL_NEXT_TAG_SYNC.store(false, Ordering::SeqCst);
FAIL_NEXT_FORCE_WRITE.store(false, Ordering::SeqCst);
FAIL_NEXT_ENCRYPTED_DATA_WRITE_AFTER_PWRITE.store(false, Ordering::SeqCst);
let gate = DATA_SYNC_GATE.lock().expect("sync gate poisoned").take();
if let Some(gate) = gate {
gate.open();
}
let gate = BARRIER_PUBLICATION_GATE
.lock()
.expect("publication gate poisoned")
.take();
if let Some(gate) = gate {
gate.gate.open();
}
}
pub(crate) fn hold_data_sync() -> DataSyncGateGuard {
let gate = Arc::new(SyncGate::default());
*DATA_SYNC_GATE.lock().expect("sync gate poisoned") = Some(Arc::clone(&gate));
DataSyncGateGuard { gate }
}
pub(crate) struct BarrierPublicationGateGuard {
generation: u64,
gate: Arc<SyncGate>,
}
struct PublicationGate {
generation: u64,
gate: Arc<SyncGate>,
}
impl BarrierPublicationGateGuard {
pub(crate) fn wait_until_blocked(&self) {
self.gate.wait_until_blocked();
}
pub(crate) fn release(&self) {
let mut active = BARRIER_PUBLICATION_GATE
.lock()
.expect("publication gate poisoned");
if active.as_ref().is_some_and(|current| {
current.generation == self.generation && Arc::ptr_eq(¤t.gate, &self.gate)
}) {
*active = None;
}
drop(active);
self.gate.open();
}
}
impl Drop for BarrierPublicationGateGuard {
fn drop(&mut self) {
self.release();
}
}
pub(crate) fn hold_barrier_publication(generation: u64) -> BarrierPublicationGateGuard {
let gate = Arc::new(SyncGate::default());
*BARRIER_PUBLICATION_GATE
.lock()
.expect("publication gate poisoned") = Some(PublicationGate {
generation,
gate: Arc::clone(&gate),
});
BarrierPublicationGateGuard { generation, gate }
}
pub(crate) fn wait_barrier_publication_gate(generation: u64) {
let gate = BARRIER_PUBLICATION_GATE
.lock()
.expect("publication gate poisoned")
.as_ref()
.filter(|current| current.generation == generation)
.map(|current| Arc::clone(¤t.gate));
if let Some(gate) = gate {
gate.block();
}
}
pub(crate) fn wait_data_sync_gate() {
let gate = DATA_SYNC_GATE.lock().expect("sync gate poisoned").clone();
if let Some(gate) = gate {
gate.block();
}
}
#[test]
fn sync_gate_remaining_deadline_does_not_restart_after_spurious_wake() {
let now = std::time::Instant::now();
let deadline = now + Duration::from_secs(5);
assert_eq!(
remaining_until(deadline, now + Duration::from_secs(2)),
Some(Duration::from_secs(3)),
"a spurious wake retains the original deadline"
);
assert_eq!(remaining_until(deadline, deadline), None);
assert_eq!(
remaining_until(deadline, deadline + Duration::from_nanos(1)),
None
);
}
#[test]
fn held_sync_gate_drop_releases_blocked_worker() {
reset();
let held: DataSyncGateGuard = hold_data_sync();
let gate = held.gate();
let worker = std::thread::spawn(move || gate.block());
held.wait_until_blocked();
drop(held);
worker
.join()
.expect("blocked worker must be released by guard drop");
}
#[test]
fn sync_gate_spurious_wake_is_observed_before_release() {
use std::sync::mpsc;
let gate = Arc::new(SyncGate::default());
let notifier_gate = Arc::clone(&gate);
let (tx, rx) = mpsc::channel();
let notifier = std::thread::spawn(move || {
notifier_gate.wake_for_test();
notifier_gate.mark_blocked_for_test();
tx.send(()).expect("report acknowledged wake");
});
gate.wait_until_blocked_for_test(Duration::from_secs(1));
rx.recv_timeout(Duration::from_secs(1))
.expect("notifier must observe the waiter before releasing it");
notifier.join().expect("notifier thread panicked");
}
pub fn data_attempts() -> u64 {
DATA_ATTEMPTS.load(Ordering::SeqCst)
}
pub fn data_successes() -> u64 {
DATA_SUCCESSES.load(Ordering::SeqCst)
}
#[cfg_attr(not(feature = "encryption"), allow(dead_code))]
pub fn tag_attempts() -> u64 {
TAG_ATTEMPTS.load(Ordering::SeqCst)
}
#[cfg_attr(not(feature = "encryption"), allow(dead_code))]
pub fn tag_successes() -> u64 {
TAG_SUCCESSES.load(Ordering::SeqCst)
}
pub fn fixed_attempts() -> u64 {
FIXED_ATTEMPTS.load(Ordering::SeqCst)
}
pub fn fixed_successes() -> u64 {
FIXED_SUCCESSES.load(Ordering::SeqCst)
}
pub fn note_data_attempt() {
DATA_ATTEMPTS.fetch_add(1, Ordering::SeqCst);
}
pub fn note_data_success() {
DATA_SUCCESSES.fetch_add(1, Ordering::SeqCst);
}
#[cfg_attr(not(feature = "encryption"), allow(dead_code))]
pub fn note_tag_attempt() {
TAG_ATTEMPTS.fetch_add(1, Ordering::SeqCst);
}
#[cfg_attr(not(feature = "encryption"), allow(dead_code))]
pub fn note_tag_success() {
TAG_SUCCESSES.fetch_add(1, Ordering::SeqCst);
}
pub fn note_fixed_attempt() {
FIXED_ATTEMPTS.fetch_add(1, Ordering::SeqCst);
}
pub fn note_fixed_success() {
FIXED_SUCCESSES.fetch_add(1, Ordering::SeqCst);
}
pub fn take_fail_data() -> bool {
FAIL_NEXT_DATA_SYNC.swap(false, Ordering::SeqCst)
}
#[cfg_attr(not(feature = "encryption"), allow(dead_code))]
pub fn take_fail_tag() -> bool {
FAIL_NEXT_TAG_SYNC.swap(false, Ordering::SeqCst)
}
pub fn take_fail_force_write() -> bool {
FAIL_NEXT_FORCE_WRITE.swap(false, Ordering::SeqCst)
}
#[cfg_attr(not(feature = "encryption"), allow(dead_code))]
pub fn take_fail_encrypted_data_write_after_pwrite() -> bool {
FAIL_NEXT_ENCRYPTED_DATA_WRITE_AFTER_PWRITE.swap(false, Ordering::SeqCst)
}
pub fn injected_sync_err(kind: &'static str) -> crate::error::DbError {
io::Error::other(format!("test: injected {kind} sync failure")).into()
}
pub fn open_plain_shard(dir: &Path) -> Shard {
Shard::open(
0,
dir,
1 << 20,
64 * 1024,
false,
false,
false,
IoBackend::default(),
Arc::new(AtomicU64::new(0)),
)
.expect("open test shard")
}
fn open_plain_shard_with_thresholds(
dir: &Path,
sync_batch_size: u32,
sync_interval: Option<Duration>,
) -> Shard {
Shard::open_with_sync_thresholds(
0,
dir,
1 << 20,
64 * 1024,
false,
false,
false,
IoBackend::default(),
sync_batch_size,
sync_interval,
Arc::new(AtomicU64::new(0)),
)
.expect("open threshold test shard")
}
fn open_plain_shard_with_thresholds_and_clock(
dir: &Path,
sync_batch_size: u32,
sync_interval: Option<Duration>,
) -> (Shard, TestSyncClock) {
let clock = TestSyncClock::install();
let shard = open_plain_shard_with_thresholds(dir, sync_batch_size, sync_interval);
(shard, clock)
}
fn append(shard: &Shard, key: &[u8], value: &[u8], tombstone: bool) {
shard
.lock()
.append_entry(shard.id, key, value, tombstone)
.expect("append_entry");
}
fn append_and_sync_if_needed(shard: &Shard, key: &[u8], value: &[u8]) -> crate::DbResult<()> {
let should_sync = {
let mut inner = shard.lock();
inner.append_entry(shard.id, key, value, false)?;
inner.should_sync()
};
if should_sync {
shard.flush()?;
}
Ok(())
}
#[test]
fn sync_batch_flushes_after_eight_pending_writes() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard_with_thresholds(dir.path(), 8, None);
for key in 0_u64..7 {
append_and_sync_if_needed(&shard, &key.to_be_bytes(), &[7; 8]).unwrap();
}
assert_eq!(data_attempts(), 0, "seven writes stay below the threshold");
append_and_sync_if_needed(&shard, &7_u64.to_be_bytes(), &[7; 8]).unwrap();
assert_eq!(
data_successes(),
1,
"the eighth write completes one barrier"
);
assert!(
!shard.is_dirty(),
"the threshold barrier covers all eight writes"
);
}
#[test]
fn sync_interval_flushes_pending_write_after_elapsed_interval() {
reset();
let dir = tempfile::tempdir().unwrap();
let (shard, clock) =
open_plain_shard_with_thresholds_and_clock(dir.path(), 0, Some(Duration::from_millis(20)));
append_and_sync_if_needed(&shard, &1_u64.to_be_bytes(), &[7; 8]).unwrap();
assert_eq!(
data_attempts(),
0,
"a fresh shard has not reached its interval"
);
clock.set_millis(20);
append_and_sync_if_needed(&shard, &2_u64.to_be_bytes(), &[8; 8]).unwrap();
assert_eq!(
data_successes(),
1,
"elapsed interval flushes pending writes"
);
}
#[test]
fn sync_interval_explicit_flush_resets_success_clock() {
reset();
let dir = tempfile::tempdir().unwrap();
let (shard, clock) =
open_plain_shard_with_thresholds_and_clock(dir.path(), 0, Some(Duration::from_millis(20)));
clock.set_millis(100);
append(&shard, &1_u64.to_be_bytes(), &[7; 8], false);
shard.flush().unwrap();
let after_explicit = data_successes();
clock.set_millis(119);
append_and_sync_if_needed(&shard, &2_u64.to_be_bytes(), &[8; 8]).unwrap();
assert_eq!(
data_successes(),
after_explicit,
"an explicit barrier restarts the interval clock before the next write"
);
clock.set_millis(121);
append_and_sync_if_needed(&shard, &3_u64.to_be_bytes(), &[9; 8]).unwrap();
assert_eq!(
data_successes(),
after_explicit + 1,
"the controlled clock reaches the reset interval exactly"
);
}
#[test]
fn sync_clock_disabled_thresholds_do_not_read_clock() {
reset();
let dir = tempfile::tempdir().unwrap();
let (shard, clock) = open_plain_shard_with_thresholds_and_clock(dir.path(), 0, None);
append_and_sync_if_needed(&shard, &1_u64.to_be_bytes(), &[7; 8]).unwrap();
assert_eq!(
clock.reads(),
0,
"disabled thresholds do not read the clock"
);
}
#[test]
fn barrier_publication_serializes_generation_and_clock() {
use std::sync::mpsc;
reset();
let dir = tempfile::tempdir().unwrap();
let (shard, clock) = open_plain_shard_with_thresholds_and_clock(dir.path(), 8, None);
append(&shard, &1_u64.to_be_bytes(), &[7; 8], false);
let lower = shard.prepare_barrier_for_test().unwrap();
append(&shard, &2_u64.to_be_bytes(), &[8; 8], false);
let higher = shard.prepare_barrier_for_test().unwrap();
clock.set_millis(10);
let gate = hold_barrier_publication(1);
let lower_publisher = std::thread::spawn(move || lower.complete());
gate.wait_until_blocked();
clock.set_millis(20);
let (tx, rx) = mpsc::channel();
let higher_publisher = std::thread::spawn(move || {
tx.send(higher.complete())
.expect("report higher publication result");
});
assert!(
matches!(
rx.recv_timeout(Duration::from_millis(100)),
Err(mpsc::RecvTimeoutError::Timeout)
),
"the higher publisher waits for the lower publisher's clock store"
);
gate.release();
lower_publisher
.join()
.expect("lower publisher panicked")
.unwrap();
rx.recv_timeout(Duration::from_secs(1))
.expect("higher publisher must finish")
.unwrap();
higher_publisher.join().expect("higher publisher panicked");
assert_eq!(shard.generations_for_test().1, 2);
assert_eq!(
shard.last_synced_at_for_test(),
21,
"the final clock belongs to the higher generation"
);
}
#[test]
fn write_sync_failure_retries_on_next_threshold_write() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard_with_thresholds(dir.path(), 8, None);
for key in 0_u64..7 {
append_and_sync_if_needed(&shard, &key.to_be_bytes(), &[7; 8]).unwrap();
}
FAIL_NEXT_DATA_SYNC.store(true, Ordering::SeqCst);
assert!(append_and_sync_if_needed(&shard, &7_u64.to_be_bytes(), &[7; 8]).is_err());
assert!(
shard.is_dirty(),
"a failed barrier leaves the generation pending"
);
append_and_sync_if_needed(&shard, &8_u64.to_be_bytes(), &[7; 8]).unwrap();
assert_eq!(
data_attempts(),
2,
"the ninth write retries the failed barrier"
);
assert_eq!(data_successes(), 1);
assert!(!shard.is_dirty());
}
#[test]
fn write_sync_in_flight_keeps_later_writes_eligible() {
use std::sync::mpsc;
reset();
let dir = tempfile::tempdir().unwrap();
let shard = Arc::new(open_plain_shard_with_thresholds(dir.path(), 8, None));
for key in 0_u64..7 {
append_and_sync_if_needed(&shard, &key.to_be_bytes(), &[7; 8]).unwrap();
}
let gate = hold_data_sync();
let flushing_shard = Arc::clone(&shard);
let (tx, rx) = mpsc::channel();
let flusher = std::thread::spawn(move || {
let result = append_and_sync_if_needed(&flushing_shard, &7_u64.to_be_bytes(), &[7; 8]);
tx.send(result).expect("report threshold barrier result");
});
gate.wait_until_blocked();
for key in 8_u64..16 {
append(&shard, &key.to_be_bytes(), &[7; 8], false);
}
let (written, synced) = shard.generations_for_test();
assert_eq!(
written.saturating_sub(synced),
16,
"eight later writes remain pending"
);
gate.release();
rx.recv_timeout(Duration::from_secs(1))
.expect("threshold barrier must finish")
.unwrap();
flusher.join().expect("threshold flusher panicked");
assert!(
shard.lock().should_sync(),
"writes after the held barrier stay eligible"
);
}
#[test]
fn sync_batch_stale_barrier_does_not_refresh_success_clock() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard_with_thresholds(dir.path(), 8, None);
append(&shard, &1_u64.to_be_bytes(), &[7; 8], false);
let first = shard.prepare_barrier_for_test().unwrap();
append(&shard, &2_u64.to_be_bytes(), &[8; 8], false);
let second = shard.prepare_barrier_for_test().unwrap();
second.complete().unwrap();
shard.set_last_synced_at_for_test(7);
first.complete().unwrap();
assert_eq!(
shard.last_synced_at_for_test(),
7,
"a stale barrier must not refresh the interval clock"
);
}
#[test]
fn dirty_shard_clean_skip_second_flush() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
shard.flush().unwrap();
let after_first = data_attempts();
assert!(after_first >= 1, "first flush must attempt a data barrier");
shard.flush().unwrap();
assert_eq!(
data_attempts(),
after_first,
"second flush of a clean shard must attempt zero data barriers"
);
assert!(!shard.is_dirty());
}
#[test]
fn barrier_complete_releases_shard_lock_before_fsync() {
use std::sync::mpsc;
use std::time::Duration;
reset();
let gate = hold_data_sync();
let dir = tempfile::tempdir().unwrap();
let shard = Arc::new(open_plain_shard(dir.path()));
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
let flushing_shard = Arc::clone(&shard);
let flusher = std::thread::spawn(move || flushing_shard.flush());
gate.wait_until_blocked();
let (tx, rx) = mpsc::channel();
let writer_shard = Arc::clone(&shard);
let writer = std::thread::spawn(move || {
append(&writer_shard, &2u64.to_be_bytes(), &[8; 8], false);
tx.send(()).expect("report append completion");
});
let append_completed = rx.recv_timeout(Duration::from_secs(1));
gate.release();
flusher.join().expect("flush thread panicked").unwrap();
writer.join().expect("append thread panicked");
assert!(
append_completed.is_ok(),
"append must complete while the data fsync is blocked"
);
assert!(
shard.is_dirty(),
"append after barrier preparation must require a new barrier"
);
let before_next = data_attempts();
shard.flush().unwrap();
assert_eq!(
data_attempts(),
before_next + 1,
"the later append must require its own barrier"
);
}
#[test]
fn barrier_reverse_completion_keeps_high_watermark() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
let first = shard.prepare_barrier_for_test().unwrap();
append(&shard, &2u64.to_be_bytes(), &[8; 8], false);
let second = shard.prepare_barrier_for_test().unwrap();
second.complete().unwrap();
let (_, after_second) = shard.generations_for_test();
first.complete().unwrap();
let (written, synced) = shard.generations_for_test();
assert_eq!(
synced, after_second,
"late completion must not lower synced_gen"
);
assert_eq!(
synced, written,
"the newest completed barrier remains durable"
);
}
#[test]
fn barrier_late_completion_after_rotate_is_noop() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
let first = shard.prepare_barrier_for_test().unwrap();
append(&shard, &2u64.to_be_bytes(), &[8; 8], false);
let second = shard.prepare_barrier_for_test().unwrap();
shard.rotate_active_for_test(8).unwrap();
let (_, after_rotate) = shard.generations_for_test();
second.complete().unwrap();
let (_, after_second) = shard.generations_for_test();
first.complete().unwrap();
let (_, synced) = shard.generations_for_test();
assert_eq!(
after_second, after_rotate,
"a stale completion after rotation must leave synced_gen unchanged"
);
assert_eq!(
synced, after_rotate,
"the older stale completion after rotation must leave synced_gen unchanged"
);
}
#[cfg(feature = "encryption")]
#[test]
fn barrier_tag_failure_does_not_publish_generation() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = Shard::open_encrypted(
0,
dir.path(),
1 << 20,
64 * 1024,
false,
false,
false,
IoBackend::default(),
Some(Arc::new(
crate::crypto::PageCipher::new(&[0xAB; 32]).unwrap(),
)),
Arc::new(AtomicU64::new(0)),
)
.unwrap();
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
let first = shard.prepare_barrier_for_test().unwrap();
append(&shard, &2u64.to_be_bytes(), &[8; 8], false);
let second = shard.prepare_barrier_for_test().unwrap();
FAIL_NEXT_TAG_SYNC.store(true, Ordering::SeqCst);
assert!(second.complete().is_err());
let (_, after_failed_second) = shard.generations_for_test();
assert_eq!(
after_failed_second, 0,
"a failed tag fsync cannot publish a generation"
);
first.complete().unwrap();
let (written, synced) = shard.generations_for_test();
assert!(
synced < written,
"the failed newer barrier must remain dirty"
);
}
#[test]
fn in_lock_barrier_blocks_same_shard_writer_until_sync_returns() {
use std::sync::mpsc;
use std::time::Duration;
reset();
let gate = hold_data_sync();
let dir = tempfile::tempdir().unwrap();
let shard = Arc::new(open_plain_shard(dir.path()));
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
let barrier_shard = Arc::clone(&shard);
let barrier = std::thread::spawn(move || barrier_shard.lock().barrier());
gate.wait_until_blocked();
let (tx, rx) = mpsc::channel();
let writer_shard = Arc::clone(&shard);
let writer = std::thread::spawn(move || {
append(&writer_shard, &2u64.to_be_bytes(), &[8; 8], false);
tx.send(()).expect("report append completion");
});
let writer_blocked = matches!(
rx.recv_timeout(Duration::from_millis(100)),
Err(mpsc::RecvTimeoutError::Timeout)
);
gate.release();
let barrier_result = barrier.join().expect("barrier thread panicked");
let writer_result = rx.recv_timeout(Duration::from_secs(1));
writer.join().expect("writer thread panicked");
assert!(
writer_blocked,
"the in-lock barrier must retain the shard mutex until fsync returns"
);
barrier_result.unwrap();
writer_result.expect("writer must run after the in-lock barrier returns");
}
#[test]
fn barrier_prepare_drain_failure_is_counted() {
reset();
crate::flush_pass::begin();
let dir = tempfile::tempdir().unwrap();
let shard = Shard::open(
0,
dir.path(),
1 << 20,
64 * 1024,
false,
true,
false,
IoBackend::default(),
Arc::new(AtomicU64::new(0)),
)
.unwrap();
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
FAIL_NEXT_FORCE_WRITE.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
assert_eq!(crate::flush_pass::counts_for_test(), (1, 1));
}
#[test]
fn clean_shard_flush_records_a_skipped_barrier() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
shard.flush().unwrap();
crate::flush_pass::begin();
shard.flush().unwrap();
assert_eq!(
crate::flush_pass::counts_for_test(),
(0, 0),
"a clean shard attempts no barrier"
);
assert_eq!(
crate::flush_pass::skips_for_test(),
1,
"a clean shard is accounted as a skipped barrier"
);
}
#[test]
fn dirty_shard_put_and_tombstone_rearm() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
assert!(shard.is_dirty());
shard.flush().unwrap();
assert!(!shard.is_dirty());
let clean = data_attempts();
shard.flush_buf().unwrap();
shard.flush_for_replication_catchup().unwrap();
assert!(!shard.is_dirty(), "buffer-only paths must not clear dirty");
shard.flush().unwrap();
assert_eq!(
data_attempts(),
clean,
"still clean after buffer-only flushes"
);
append(&shard, &1u64.to_be_bytes(), &[], true);
assert!(shard.is_dirty(), "tombstone must dirty");
shard.flush().unwrap();
assert!(data_attempts() > clean);
assert!(!shard.is_dirty());
}
#[test]
fn dirty_shard_failed_barrier_retries() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
FAIL_NEXT_DATA_SYNC.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
assert!(
shard.is_dirty(),
"failed barrier must leave the shard dirty"
);
assert_eq!(data_successes(), 0);
shard.flush().unwrap();
assert!(!shard.is_dirty());
assert_eq!(data_successes(), 1);
}
#[test]
fn dirty_shard_create_starts_clean() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
assert!(!shard.is_dirty());
shard.flush().unwrap();
assert_eq!(data_attempts(), 0);
}
#[test]
fn dirty_shard_reopen_starts_dirty() {
reset();
let dir = tempfile::tempdir().unwrap();
{
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
}
reset();
let shard = open_plain_shard(dir.path());
assert!(
shard.is_dirty(),
"reopened existing active file starts dirty"
);
shard.flush().unwrap();
assert_eq!(data_successes(), 1);
assert!(!shard.is_dirty());
shard.flush().unwrap();
assert_eq!(data_successes(), 1, "second tick after reopen is clean");
}
#[test]
fn dirty_shard_rotate_and_write_active_hint_clear() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
assert!(shard.is_dirty());
shard.rotate_active_for_test(8).unwrap();
assert!(!shard.is_dirty(), "rotate barriers before the file switch");
append(&shard, &2u64.to_be_bytes(), &[8; 8], false);
shard.write_active_hint(8).unwrap();
assert!(!shard.is_dirty(), "write_active_hint is a full barrier");
}
#[cfg(feature = "replication")]
#[test]
fn dirty_shard_catchup_snapshot_clears() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
shard.catchup_snapshot().unwrap();
assert!(!shard.is_dirty());
}
#[test]
fn dirty_shard_recovery_tail_truncation_stays_dirty_until_barrier() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
shard.flush().unwrap();
let inner = shard.lock();
let file_id = inner.active.file_id;
let path = inner.active.path.clone();
drop(inner);
use std::io::Write;
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
f.write_all(&[0xFF; 16]).unwrap();
drop(f);
drop(shard);
reset();
let shard = open_plain_shard(dir.path());
shard.flush().unwrap();
assert!(!shard.is_dirty());
reset();
let tail = crate::recovery::ActiveTail {
shard_idx: 0,
file_id,
last_valid_offset: {
let meta_len = std::fs::metadata(dir.path().join(format!("{file_id:06}.data")))
.unwrap()
.len();
meta_len.saturating_sub(16)
},
};
shard.apply_recovery_tail(&tail).unwrap();
assert!(shard.is_dirty());
shard.flush().unwrap();
assert_eq!(data_successes(), 1);
shard.flush().unwrap();
assert_eq!(data_successes(), 1);
}
#[cfg(feature = "replication")]
#[test]
fn dirty_shard_append_raw_entry_is_dirty() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = open_plain_shard(dir.path());
shard.flush().unwrap();
assert!(!shard.is_dirty(), "new shard starts clean");
shard
.lock()
.append_raw_entry(0, 8, &[0u8; 32])
.expect("append_raw_entry");
assert!(
shard.is_dirty(),
"replication apply must bump write_gen so the tick barriers the follower"
);
shard.flush().unwrap();
assert!(!shard.is_dirty());
let after = data_successes();
shard.flush().unwrap();
assert_eq!(
data_successes(),
after,
"clean follower shard must skip the next barrier"
);
}
#[test]
fn force_flush_io_error_keeps_padding_so_next_append_does_not_reuse_page() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = Shard::open(
0,
dir.path(),
1 << 20,
64 * 1024,
false,
true,
false,
IoBackend::default(),
Arc::new(AtomicU64::new(0)),
)
.expect("open page-aligned shard");
let (loc1, _) = shard
.lock()
.append_entry(0, &1u64.to_be_bytes(), &[7; 8], false)
.unwrap();
assert!(
loc1.offset < 4096,
"first append must land in page 0, got offset {}",
loc1.offset
);
FAIL_NEXT_FORCE_WRITE.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
let (loc2, _) = shard
.lock()
.append_entry(0, &2u64.to_be_bytes(), &[8; 8], false)
.unwrap();
assert!(
loc2.offset >= 4096,
"force-flush padding must stay after I/O error so the next append does not reuse the sealed page; got offset {}",
loc2.offset
);
shard.flush().unwrap();
assert!(!shard.is_dirty());
}
#[test]
fn padding_dead_runtime_counts_each_force_flushed_page_once() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = Shard::open(
0,
dir.path(),
1 << 20,
8192,
false,
true,
false,
IoBackend::default(),
Arc::new(AtomicU64::new(0)),
)
.expect("open page-aligned shard");
let entry_size = crate::entry::entry_size(8, 8);
for key in 1_u64..=3 {
append(&shard, &key.to_be_bytes(), &[7; 8], false);
shard.flush().unwrap();
}
let inner = shard.lock();
let padding = inner
.dead_bytes
.get(&inner.active.file_id)
.copied()
.unwrap_or(0);
assert_eq!(
padding,
3 * (4096 - entry_size),
"each force-flushed partial page contributes its padding exactly once"
);
}
#[test]
fn padding_dead_failure_keeps_failed_force_padding_until_later_success() {
reset();
let dir = tempfile::tempdir().unwrap();
let shard = Shard::open(
0,
dir.path(),
1 << 20,
8192,
false,
true,
false,
IoBackend::default(),
Arc::new(AtomicU64::new(0)),
)
.expect("open page-aligned shard");
let entry_size = crate::entry::entry_size(8, 8);
let file_id = shard.lock().active.file_id;
append(&shard, &1_u64.to_be_bytes(), &[7; 8], false);
FAIL_NEXT_FORCE_WRITE.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
0,
"a failed force write must not publish padding as dead bytes"
);
append(&shard, &2_u64.to_be_bytes(), &[8; 8], false);
shard.flush().unwrap();
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
2 * (4096 - entry_size),
"the later covering force write charges both retained and new padding"
);
}
#[cfg(feature = "encryption")]
#[test]
fn encrypted_force_flush_io_error_keeps_padding() {
reset();
let dir = tempfile::tempdir().unwrap();
let cipher = Some(Arc::new(
crate::crypto::PageCipher::new(&[0xAB; 32]).expect("create cipher"),
));
let shard = Shard::open_encrypted(
0,
dir.path(),
1 << 20,
64 * 1024,
false,
false,
false,
IoBackend::default(),
cipher,
Arc::new(AtomicU64::new(0)),
)
.unwrap();
let (loc1, _) = shard
.lock()
.append_entry(0, &1u64.to_be_bytes(), &[7; 8], false)
.unwrap();
assert!(loc1.offset < 4096);
FAIL_NEXT_FORCE_WRITE.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
let (loc2, _) = shard
.lock()
.append_entry(0, &2u64.to_be_bytes(), &[8; 8], false)
.unwrap();
assert!(
loc2.offset >= 4096,
"encrypted force-flush must not reuse a page whose nonce was already sealed; got offset {}",
loc2.offset
);
shard.flush().unwrap();
assert!(!shard.is_dirty());
}
#[cfg(feature = "encryption")]
#[test]
fn padding_dead_failure_encrypted_non_force_publishes_retained_padding() {
reset();
let dir = tempfile::tempdir().unwrap();
let cipher = Some(Arc::new(
crate::crypto::PageCipher::new(&[0xAB; 32]).expect("create cipher"),
));
let shard = Shard::open_encrypted(
0,
dir.path(),
1 << 20,
8192,
false,
false,
false,
IoBackend::default(),
cipher,
Arc::new(AtomicU64::new(0)),
)
.unwrap();
let entry_size = crate::entry::entry_size(8, 8);
let file_id = shard.lock().active.file_id;
append(&shard, &1_u64.to_be_bytes(), &[7; 8], false);
FAIL_NEXT_FORCE_WRITE.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
0,
"a failed encrypted force write must not publish padding as dead bytes"
);
for key in 2_u64..=129 {
append(&shard, &key.to_be_bytes(), &[8; 8], false);
}
shard.lock().flush_write_buf().unwrap();
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
4096 - entry_size,
"the successful encrypted non-force write charges retained force padding once"
);
append(&shard, &130_u64.to_be_bytes(), &[9; 8], false);
shard.rotate_active_for_test(8).unwrap();
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
2 * (4096 - entry_size),
"the retired file records the retained and later force padding exactly once"
);
}
#[cfg(feature = "encryption")]
#[test]
fn padding_dead_failure_encrypted_after_data_write_waits_for_tag_publication() {
reset();
let dir = tempfile::tempdir().unwrap();
let cipher = Some(Arc::new(
crate::crypto::PageCipher::new(&[0xAB; 32]).expect("create cipher"),
));
let shard = Shard::open_encrypted(
0,
dir.path(),
1 << 20,
8192,
false,
false,
false,
IoBackend::default(),
cipher,
Arc::new(AtomicU64::new(0)),
)
.unwrap();
let entry_size = crate::entry::entry_size(8, 8);
let file_id = shard.lock().active.file_id;
append(&shard, &1_u64.to_be_bytes(), &[7; 8], false);
FAIL_NEXT_ENCRYPTED_DATA_WRITE_AFTER_PWRITE.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
0,
"padding must remain uncredited after data writes but before tag publication"
);
shard.lock().flush_write_buf().unwrap();
assert_eq!(
shard.lock().dead_bytes.get(&file_id).copied().unwrap_or(0),
4096 - entry_size,
"the retry charges the retained padding exactly once after tags publish"
);
}
#[test]
fn dirty_shard_engine_flush_attempts_every_shard() {
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = crate::Config::test();
cfg.shard_count = 2;
cfg.hints = Some(false);
let engine = crate::engine::Engine::open(dir.path(), cfg).unwrap();
{
let s0 = &engine.shards()[0];
s0.lock()
.append_entry(0, &1u64.to_be_bytes(), &[7; 8], false)
.unwrap();
let s1 = &engine.shards()[1];
s1.lock()
.append_entry(1, &2u64.to_be_bytes(), &[8; 8], false)
.unwrap();
}
FAIL_NEXT_DATA_SYNC.store(true, Ordering::SeqCst);
assert!(engine.flush().is_err());
assert!(engine.shards()[0].is_dirty(), "failed shard 0 stays dirty");
assert!(
!engine.shards()[1].is_dirty(),
"shard 1 must still have been synced in the same call"
);
assert_eq!(data_successes(), 1);
}
#[cfg(feature = "encryption")]
#[test]
fn dirty_shard_tag_failure_keeps_dirty() {
reset();
let dir = tempfile::tempdir().unwrap();
let gsn = Arc::new(AtomicU64::new(0));
let cipher = Some(Arc::new(
crate::crypto::PageCipher::new(&[0xAB; 32]).expect("create cipher"),
));
let shard = Shard::open_encrypted(
0,
dir.path(),
1 << 20,
64 * 1024,
false,
false,
false,
IoBackend::default(),
cipher,
gsn,
)
.unwrap();
append(&shard, &1u64.to_be_bytes(), &[7; 8], false);
FAIL_NEXT_TAG_SYNC.store(true, Ordering::SeqCst);
assert!(shard.flush().is_err());
assert!(shard.is_dirty());
assert!(tag_attempts() >= 1);
assert_eq!(tag_successes(), 0);
shard.flush().unwrap();
assert!(!shard.is_dirty());
assert!(tag_successes() >= 1);
}
#[test]
fn dirty_shard_compaction_keeps_buffered_newer_value() {
use crate::Config;
use crate::ConstTree;
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = Config::test();
cfg.shard_count = 1;
cfg.max_file_size = 4096;
cfg.write_buffer_size = 4096;
cfg.compaction_threshold = 0.0;
cfg.hints = Some(true);
let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
let k = 1u64.to_be_bytes();
for i in 0..256u64 {
tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
}
tree.flush().unwrap(); tree.put(&k, &99u64.to_be_bytes()).unwrap(); let before = data_attempts();
let compacted = tree.compact().unwrap();
assert!(
compacted > 0,
"compaction must rewrite the immutable file that holds v1"
);
assert!(
data_attempts() > before,
"compact must barrier the dirty active file before deleting sources"
);
let got = tree.get(&k).expect("buffered newer value must survive");
assert_eq!(got, 99u64.to_be_bytes());
}
#[test]
fn compaction_predelete_barrier_keeps_sources_until_fsync_and_later_append_dirty() {
use crate::{Config, ConstTree};
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = Config::test();
cfg.shard_count = 1;
cfg.max_file_size = 4096;
cfg.write_buffer_size = 4096;
cfg.compaction_threshold = 0.0;
cfg.hints = Some(true);
let tree = Arc::new(ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap());
let key = 1u64.to_be_bytes();
for i in 0..256u64 {
tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
}
tree.flush().unwrap();
tree.put(&key, &99u64.to_be_bytes()).unwrap();
let shard_dir = dir.path().join("shard_000");
let sources = std::fs::read_dir(&shard_dir)
.unwrap()
.map(|entry| entry.unwrap().path())
.filter(|path| path.extension().is_some_and(|ext| ext == "data"))
.collect::<Vec<_>>();
let gate = hold_data_sync();
let compaction_tree = Arc::clone(&tree);
let compaction = std::thread::spawn(move || compaction_tree.compact());
gate.wait_until_blocked();
let sources_retained = sources.iter().all(|path| path.exists());
let append_result = tree.put(&key, &100u64.to_be_bytes());
gate.release();
let compaction_result = compaction.join().expect("compaction thread panicked");
assert!(
sources_retained,
"compaction must retain every source until its pre-delete barrier completes"
);
append_result.unwrap();
assert!(compaction_result.unwrap() > 0);
let before_later_barrier = data_attempts();
tree.durable_flush().unwrap();
assert_eq!(
data_attempts(),
before_later_barrier + 1,
"the append after compaction barrier preparation needs its own barrier"
);
}
#[test]
fn fixed_flush_gate_skips_clean_shard() {
use crate::fixed::config::FixedConfig;
use crate::fixed::engine::FixedEngine;
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = FixedConfig::test();
cfg.shard_count = 1;
cfg.enable_fsync = false;
let engine = FixedEngine::open(dir.path(), cfg, 8, 16).unwrap();
engine.flush().unwrap();
let baseline = crate::flush_test::fixed_attempts();
engine.flush().unwrap();
assert_eq!(
crate::flush_test::fixed_attempts(),
baseline,
"clean Fixed shard must not be fdatasynced again"
);
}
#[test]
fn fixed_flush_gate_foreign_batch_leaves_unsynced() {
use crate::fixed::config::FixedConfig;
use crate::fixed::shard::FixedShardInner;
use crate::fixed::slot;
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = FixedConfig::test();
cfg.enable_fsync = false;
let mut shard = FixedShardInner::open(dir.path(), 0, 8, 16, &cfg).unwrap();
shard
.apply_foreign_slot_deferred(
0,
slot::pack_meta(slot::STATUS_OCCUPIED, 1),
b"key_0001",
b"value___00000001",
)
.unwrap();
shard.sync_replication_batch().unwrap();
assert_eq!(shard.pending_writes_for_test(), 0);
assert!(
shard.is_unsynced(),
"F1: pending_writes reset without a real sync must not clear unsynced"
);
}
#[test]
fn fixed_flush_gate_enable_fsync_tick_is_noop() {
use crate::fixed::config::FixedConfig;
use crate::fixed::engine::FixedEngine;
use crate::fixed::slot;
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = FixedConfig::test();
cfg.shard_count = 1;
cfg.enable_fsync = true;
let engine = FixedEngine::open(dir.path(), cfg, 8, 16).unwrap();
{
let mut inner = crate::sync::lock(&engine.shards()[0].inner);
inner
.write_slot_with_meta(
0,
slot::pack_meta(slot::STATUS_OCCUPIED, 1),
b"key_0001",
b"value___00000001",
)
.unwrap();
}
let after_write = crate::flush_test::fixed_successes();
assert!(after_write >= 1, "per-write enable_fsync must have synced");
engine.flush().unwrap();
assert_eq!(
crate::flush_test::fixed_successes(),
after_write,
"tick must not issue a redundant fdatasync when enable_fsync already cleared unsynced"
);
}
#[cfg(feature = "armour")]
#[test]
fn db_dirty_shards_counts_the_bitcask_loss_window() {
use crate::{CollectionMeta, Config, NoHook};
use armour_core::GetType;
use zerocopy::{FromBytes, Immutable, IntoBytes};
#[derive(Clone, Copy, Debug, GetType, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct Row {
n: u64,
}
impl CollectionMeta for Row {
type SelfId = [u8; 8];
const NAME: &'static str = "bitcask_dirty_shards";
}
reset();
let dir = tempfile::tempdir().unwrap();
let db = crate::armour::Db::open_test(dir.path()).unwrap();
let mut cfg = Config::test();
cfg.shard_count = 1;
let tree = db.open_zero_tree::<Row, 8, _>(cfg, NoHook, &[]).unwrap();
crate::armour::collection::Collection::periodic_flush(&*tree).unwrap();
assert_eq!(db.dirty_shards(), 0, "a barrier closes the loss window");
tree.put(&1u64.to_be_bytes(), &Row { n: 1 }).unwrap();
assert_eq!(
db.dirty_shards(),
1,
"an unsynced write reopens the loss window on exactly one shard"
);
crate::armour::collection::Collection::periodic_flush(&*tree).unwrap();
assert_eq!(
db.dirty_shards(),
0,
"the tick barrier closes the loss window again"
);
}
#[cfg(feature = "armour")]
#[test]
fn db_dirty_shards_counts_unsynced_fixed_shards() {
use crate::fixed::config::FixedConfig;
use crate::{CollectionMeta, NoHook};
use armour_core::GetType;
use zerocopy::{FromBytes, Immutable, IntoBytes};
#[derive(Clone, Copy, Debug, GetType, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct Row {
n: u64,
}
impl CollectionMeta for Row {
type SelfId = [u8; 8];
const NAME: &'static str = "fixed_dirty_shards";
}
reset();
let dir = tempfile::tempdir().unwrap();
let db = crate::armour::Db::open_test(dir.path()).unwrap();
let mut cfg = FixedConfig::test();
cfg.shard_count = 1;
cfg.enable_fsync = false;
cfg.sync_interval = Duration::from_secs(3600);
cfg.sync_batch_size = 10_000;
let tree = db
.open_zero_tree_fixed::<Row, 8, _>(cfg, NoHook, &[])
.unwrap();
assert_eq!(
db.dirty_shards(),
0,
"a fresh collection has no loss window"
);
tree.put(&1u64.to_be_bytes(), &Row { n: 1 }).unwrap();
assert_eq!(
db.dirty_shards(),
1,
"an unsynced FixedStore write leaves exactly one dirty shard"
);
crate::armour::collection::Collection::periodic_flush(&*tree).unwrap();
assert_eq!(
db.dirty_shards(),
0,
"the tick barrier closes the loss window"
);
}
#[cfg(feature = "armour")]
#[test]
fn fixed_flush_gate_sync_interval_first_write_after_idle() {
use crate::fixed::config::FixedConfig;
use crate::{CollectionMeta, NoHook};
use armour_core::GetType;
use std::time::Duration;
use zerocopy::{FromBytes, Immutable, IntoBytes};
#[derive(Clone, Copy, Debug, GetType, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct Row {
n: u64,
}
impl CollectionMeta for Row {
type SelfId = [u8; 8];
const NAME: &'static str = "fixed_flush_gate_idle";
}
reset();
let dir = tempfile::tempdir().unwrap();
let db = crate::armour::Db::open_test(dir.path()).unwrap();
let mut cfg = FixedConfig::test();
cfg.shard_count = 1;
cfg.enable_fsync = false;
cfg.sync_interval = Duration::from_millis(20);
cfg.sync_batch_size = 10_000; let tree = db
.open_zero_tree_fixed::<Row, 8, _>(cfg, NoHook, &[])
.unwrap();
tree.put(&1u64.to_be_bytes(), &Row { n: 1 }).unwrap();
tree.flush().unwrap(); reset();
tree.flush().unwrap(); assert_eq!(crate::flush_test::fixed_attempts(), 0);
std::thread::sleep(Duration::from_millis(200));
reset();
tree.put(&2u64.to_be_bytes(), &Row { n: 2 }).unwrap();
assert_eq!(
crate::flush_test::fixed_successes(),
1,
"first write after idle longer than sync_interval issues exactly one fdatasync on the return path"
);
}
#[test]
fn fixed_flush_gate_engine_attempts_every_shard() {
use crate::fixed::config::FixedConfig;
use crate::fixed::engine::FixedEngine;
use crate::fixed::shard::{SyncSite, fail_next_sync};
use crate::fixed::slot;
reset();
let dir = tempfile::tempdir().unwrap();
let mut cfg = FixedConfig::test();
cfg.shard_count = 2;
cfg.enable_fsync = false;
let engine = FixedEngine::open(dir.path(), cfg, 8, 16).unwrap();
for shard in engine.shards() {
let mut inner = crate::sync::lock(&shard.inner);
inner
.write_slot_with_meta(
0,
slot::pack_meta(slot::STATUS_OCCUPIED, 1),
b"key_0001",
b"value___00000001",
)
.unwrap();
}
let failure = fail_next_sync(SyncSite::Tick);
assert!(engine.flush().is_err());
drop(failure);
assert!(
crate::sync::lock(&engine.shards()[0].inner).is_unsynced(),
"failed shard 0 stays unsynced"
);
assert!(
!crate::sync::lock(&engine.shards()[1].inner).is_unsynced(),
"shard 1 must still have been synced"
);
}
#[cfg(feature = "armour")]
mod durable_tick {
use super::*;
use crate::armour::{Db, WorkerIntervals};
use crate::{CollectionMeta, Config, NoHook};
use armour_core::GetType;
use std::time::Duration;
use zerocopy::{FromBytes, Immutable, IntoBytes};
#[derive(Clone, Copy, Debug, GetType, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct Row {
n: u64,
}
impl CollectionMeta for Row {
type SelfId = [u8; 8];
const NAME: &'static str = "durable_tick_rows";
}
fn workers_off() -> WorkerIntervals {
WorkerIntervals {
compaction: None,
flush: None,
ttl_sweep: None,
ttl_budget: 4096,
}
}
#[test]
fn durable_tick_worker_advances_barrier_counter() {
reset();
let dir = tempfile::tempdir().unwrap();
let db = Db::builder(dir.path())
.for_tests()
.workers(WorkerIntervals {
flush: Some(Duration::from_millis(50)),
..workers_off()
})
.open()
.unwrap();
let tree = db
.open_zero_tree::<Row, 8, _>(Config::test(), NoHook, &[])
.unwrap();
tree.put(&1u64.to_be_bytes(), &Row { n: 1 }).unwrap();
let start = std::time::Instant::now();
while data_successes() == 0 && start.elapsed() < Duration::from_secs(2) {
std::thread::sleep(Duration::from_millis(20));
}
assert!(
data_successes() >= 1,
"flush: Some(50ms) must barrier a dirty Bitcask shard"
);
}
#[test]
fn durable_tick_worker_advances_barrier_counter_zero_map() {
reset();
let dir = tempfile::tempdir().unwrap();
let db = Db::builder(dir.path())
.for_tests()
.workers(WorkerIntervals {
flush: Some(Duration::from_millis(50)),
..workers_off()
})
.open()
.unwrap();
let map = db
.open_zero_map::<Row, 8, _>(Config::test(), NoHook, &[])
.unwrap();
map.put(&1u64.to_be_bytes(), &Row { n: 1 }).unwrap();
let start = std::time::Instant::now();
while data_successes() == 0 && start.elapsed() < Duration::from_secs(2) {
std::thread::sleep(Duration::from_millis(20));
}
assert!(
data_successes() >= 1,
"flush: Some(50ms) must barrier a dirty Bitcask ZeroMap shard"
);
}
#[test]
fn durable_tick_worker_none_does_not_barrier() {
reset();
let dir = tempfile::tempdir().unwrap();
let db = Db::builder(dir.path())
.for_tests()
.workers(workers_off())
.open()
.unwrap();
let tree = db
.open_zero_tree::<Row, 8, _>(Config::test(), NoHook, &[])
.unwrap();
tree.put(&1u64.to_be_bytes(), &Row { n: 1 }).unwrap();
std::thread::sleep(Duration::from_millis(200));
assert_eq!(
data_successes(),
0,
"flush: None must not run the tick (explicit flush / checkpoint still would)"
);
}
#[test]
fn durable_tick_checkpoint_quiet_attempts_no_barriers() {
reset();
let dir = tempfile::tempdir().unwrap();
let db = Db::builder(dir.path())
.for_tests()
.workers(workers_off())
.open()
.unwrap();
let tree = db
.open_zero_tree::<Row, 8, _>(Config::test(), NoHook, &[])
.unwrap();
db.tx()
.put(&tree, &1u64.to_be_bytes(), &Row { n: 1 })
.unwrap()
.commit()
.unwrap();
crate::txn::checkpoint::durable_flush_all(&db.collections).unwrap();
reset();
crate::txn::checkpoint::run(&db).unwrap();
assert_eq!(
data_attempts(),
0,
"checkpoint over already-clean collections attempts no barriers"
);
assert!(
db.txlog().pending().is_empty(),
"quiet checkpoint must still retire applied records"
);
}
#[test]
fn durable_tick_failed_barrier_blocks_checkpoint_retire() {
reset();
let dir = tempfile::tempdir().unwrap();
let db = Db::builder(dir.path())
.for_tests()
.workers(workers_off())
.open()
.unwrap();
let tree = db
.open_zero_tree::<Row, 8, _>(Config::test(), NoHook, &[])
.unwrap();
db.tx()
.put(&tree, &1u64.to_be_bytes(), &Row { n: 1 })
.unwrap()
.commit()
.unwrap();
FAIL_NEXT_DATA_SYNC.store(true, Ordering::SeqCst);
assert!(crate::txn::checkpoint::run(&db).is_err());
assert!(
!db.txlog().pending().is_empty(),
"a failed collection barrier must not retire the log record"
);
}
#[test]
fn checkpoint_barrier_leaves_later_append_dirty() {
use std::sync::Arc;
reset();
let gate = hold_data_sync();
let dir = tempfile::tempdir().unwrap();
let db = Arc::new(
Db::builder(dir.path())
.for_tests()
.workers(workers_off())
.open()
.unwrap(),
);
let tree = db
.open_zero_tree::<Row, 8, _>(Config::test(), NoHook, &[])
.unwrap();
db.tx()
.put(&tree, &1u64.to_be_bytes(), &Row { n: 1 })
.unwrap()
.commit()
.unwrap();
let checkpoint_db = Arc::clone(&db);
let checkpoint = std::thread::spawn(move || checkpoint_db.checkpoint_now());
gate.wait_until_blocked();
let append_result = tree.put(&2u64.to_be_bytes(), &Row { n: 2 });
gate.release();
let checkpoint_result = checkpoint.join().expect("checkpoint thread panicked");
append_result.unwrap();
checkpoint_result.unwrap();
let before_later_barrier = data_attempts();
crate::txn::checkpoint::durable_flush_all(&db.collections).unwrap();
assert_eq!(
data_attempts(),
before_later_barrier + 1,
"the append after checkpoint preparation needs its own barrier"
);
}
#[test]
fn durable_tick_enable_fsync_warns_once_on_bitcask_open() {
use std::sync::{Arc, Mutex};
use tracing::Subscriber;
use tracing::event::Event;
use tracing::metadata::Metadata;
use tracing::span::{Attributes, Id, Record};
#[derive(Default)]
struct WarnCollector(Arc<Mutex<Vec<String>>>);
impl Subscriber for WarnCollector {
fn enabled(&self, m: &Metadata<'_>) -> bool {
*m.level() <= tracing::Level::WARN
}
fn new_span(&self, _: &Attributes<'_>) -> Id {
Id::from_u64(1)
}
fn record(&self, _: &Id, _: &Record<'_>) {}
fn record_follows_from(&self, _: &Id, _: &Id) {}
fn event(&self, ev: &Event<'_>) {
struct Vis(String);
impl tracing::field::Visit for Vis {
fn record_debug(&mut self, f: &tracing::field::Field, v: &dyn std::fmt::Debug) {
self.0.push_str(&format!("{}={v:?} ", f.name()));
}
}
let mut vis = Vis(String::new());
ev.record(&mut vis);
if let Ok(mut g) = self.0.lock() {
g.push(format!("{} {}", ev.metadata().level(), vis.0));
}
}
fn enter(&self, _: &Id) {}
fn exit(&self, _: &Id) {}
}
let dir = tempfile::tempdir().unwrap();
let mut cfg = Config::test();
cfg.enable_fsync = true;
let collector = WarnCollector::default();
let events = Arc::clone(&collector.0);
tracing::subscriber::with_default(collector, || {
let _engine = crate::engine::Engine::open(dir.path(), cfg).unwrap();
});
let logged = events.lock().expect("events").clone();
assert!(
logged
.iter()
.any(|e| e.contains("WARN") && e.contains("enable_fsync") && e.contains("Bitcask")),
"explicit enable_fsync=true must warn once at Engine::open. Got: {logged:?}"
);
}
#[test]
fn durable_tick_workload_durable_does_not_warn() {
use std::sync::{Arc, Mutex};
use tracing::Subscriber;
use tracing::event::Event;
use tracing::metadata::Metadata;
use tracing::span::{Attributes, Id, Record};
#[derive(Default)]
struct WarnCollector(Arc<Mutex<Vec<String>>>);
impl Subscriber for WarnCollector {
fn enabled(&self, m: &Metadata<'_>) -> bool {
*m.level() <= tracing::Level::WARN
}
fn new_span(&self, _: &Attributes<'_>) -> Id {
Id::from_u64(1)
}
fn record(&self, _: &Id, _: &Record<'_>) {}
fn record_follows_from(&self, _: &Id, _: &Id) {}
fn event(&self, ev: &Event<'_>) {
struct Vis(String);
impl tracing::field::Visit for Vis {
fn record_debug(&mut self, f: &tracing::field::Field, v: &dyn std::fmt::Debug) {
self.0.push_str(&format!("{}={v:?} ", f.name()));
}
}
let mut vis = Vis(String::new());
ev.record(&mut vis);
if let Ok(mut g) = self.0.lock() {
g.push(format!("{} {}", ev.metadata().level(), vis.0));
}
}
fn enter(&self, _: &Id) {}
fn exit(&self, _: &Id) {}
}
let dir = tempfile::tempdir().unwrap();
let cfg = Config::durable().build();
assert!(!cfg.enable_fsync, "Durable must no longer set enable_fsync");
let collector = WarnCollector::default();
let events = Arc::clone(&collector.0);
tracing::subscriber::with_default(collector, || {
let _engine = crate::engine::Engine::open(dir.path(), cfg).unwrap();
});
let logged = events.lock().expect("events").clone();
assert!(
logged.iter().all(|e| !e.contains("enable_fsync")),
"Durable preset must not warn. Got: {logged:?}"
);
}
}