This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 91c3ca54b test(simulator): cover durability failures no existing test
can reach (#4132)
91c3ca54b is described below
commit 91c3ca54bbaa1a6adcc8a996f8d9eb77d705bf2b
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Fri Sep 11 18:05:59 2026 +0200
test(simulator): cover durability failures no existing test can reach
(#4132)
Adds extra tests scenarios after #4092 PR.
---
core/configs/src/configs_impl/file_provider.rs | 60 ++++
core/integration/tests/cluster/crash_durability.rs | 37 +++
core/partitions/src/iggy_partition.rs | 73 ++++-
core/partitions/src/persistence.rs | 43 +++
core/simulator/src/lib.rs | 53 ++++
core/simulator/src/storage.rs | 96 +++++-
core/simulator/src/storage/tests.rs | 332 ++++++++++++++++++++-
7 files changed, 684 insertions(+), 10 deletions(-)
diff --git a/core/configs/src/configs_impl/file_provider.rs
b/core/configs/src/configs_impl/file_provider.rs
index bc692d8ed..f384352fd 100644
--- a/core/configs/src/configs_impl/file_provider.rs
+++ b/core/configs/src/configs_impl/file_provider.rs
@@ -402,6 +402,34 @@ mod tests {
assert!(found.is_empty(), "unexpected matches: {found:?}");
}
+ /// The allowlist is exact names but the filter is a bare `IGGY_` prefix,
and
+ /// `IGGY_` prefixes every sibling binary's namespace. A shared container
+ /// environment, a shared `env_file`, or the `.env` that `main.rs` loads
+ /// through `dotenvy` before `load_config` runs will refuse server boot.
+ #[test]
+ #[ignore = "PR #4092 review: the `IGGY_` prefix fence refuses the repo's
own `IGGY_CONNECTORS_*`, `IGGY_MCP_*` and CLI variables, with no opt-out"]
+ fn
given_a_sibling_binarys_env_vars_when_rejecting_then_the_server_should_still_boot()
{
+ let siblings = [
+ "IGGY_CONNECTORS_CONFIG_PATH",
+ "IGGY_CONNECTORS_STATE_PATH",
+ "IGGY_MCP_CONFIG_PATH",
+ "IGGY_MCP_TRANSPORT",
+ "IGGY_HOME",
+ "IGGY_USERNAME",
+ "IGGY_PASSWORD",
+ ];
+ let unknown = unknown_env_names(
+ names(&siblings).into_iter(),
+ "IGGY_",
+ crate::server_config::server::SERVER_PROCESS_ENV_VARS,
+ );
+
+ assert!(
+ unknown.is_empty(),
+ "the server refuses to boot when its own sibling products'
variables are present: {unknown:?}"
+ );
+ }
+
#[test]
fn given_another_configs_prefix_when_matching_then_should_report_none() {
let found = relocated_env_vars(
@@ -413,6 +441,38 @@ mod tests {
assert!(found.is_empty(), "unexpected matches: {found:?}");
}
+ /// `main.rs` loads a `.env` through `dotenvy` before `load_config` runs,
and
+ /// `dotenvy` injects into the process environment that
`reject_unknown_env_names`
+ /// scans with `env::vars_os()`. So the fence does not need a shared
container
+ /// or a shared `env_file`: a `.env` in the working directory is enough.
+ ///
+ /// Mutates the process environment, so it must not run beside another test
+ /// that reads it.
+ #[test]
+ #[ignore = "PR #4092 review: a `.env` loaded by `dotenvy` before
`load_config` refuses server boot; also mutates the process environment, so it
must not run in parallel"]
+ fn
given_a_dotenv_with_a_connectors_variable_when_loading_then_the_server_should_boot()
{
+ // SAFETY: single-threaded assertion over a variable no other test
reads.
+ unsafe { std::env::set_var("IGGY_CONNECTORS_CONFIG_PATH",
"/etc/iggy/connectors.toml") };
+
+ let provider = FileConfigProvider::new(
+ "nonexistent-config.toml".to_string(),
+ Toml::string(""),
+ false,
+ None,
+ )
+ .with_relocated_keys("IGGY_", &[])
+
.with_known_env_names(crate::server_config::server::SERVER_PROCESS_ENV_VARS.to_vec());
+ let rejected = provider.reject_unknown_env_names();
+
+ // SAFETY: paired with the set above.
+ unsafe { std::env::remove_var("IGGY_CONNECTORS_CONFIG_PATH") };
+
+ assert!(
+ rejected.is_ok(),
+ "a .env naming the connectors runtime's own config path refuses
server boot, with no opt-out and a message that names no remedy"
+ );
+ }
+
#[test]
fn given_no_relocated_keys_when_rejecting_then_should_accept() {
let provider = FileConfigProvider::new(
diff --git a/core/integration/tests/cluster/crash_durability.rs
b/core/integration/tests/cluster/crash_durability.rs
index 61d58f142..4967966e0 100644
--- a/core/integration/tests/cluster/crash_durability.rs
+++ b/core/integration/tests/cluster/crash_durability.rs
@@ -859,3 +859,40 @@ async fn verify_transferred_quorum(
sleep(POLL_INTERVAL).await;
}
}
+
+/// Every persisted case in this file kills with SIGKILL, and the module doc
+/// above states why that cannot reach fsync ordering. This pins the premise:
+/// a completed `write()` lives in the page cache, which the kernel owns, so
+/// process death cannot lose it. Consequently no barrier order (`file.sync()`,
+/// `sync_data()`, `fsync_dir()`) changes the outcome of a single test here,
+/// and `persisted` differs from `replicated` only in surviving writes that
+/// were never synced.
+///
+/// Covering the real contract needs a fault that discards unsynced pages:
+/// either the deterministic simulator (drop the persisted-topic assert in
+/// `core/shard/src/lib.rs` and route partition storage through
+/// `DurableStorage`) or `dm-log-writes` / `dm-flakey --drop_writes` under the
+/// data directory.
+#[test]
+fn
given_a_completed_write_when_the_process_is_sigkilled_then_the_bytes_should_survive()
{
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("unsynced");
+ let status = std::process::Command::new("sh")
+ .arg("-c")
+ .arg(r#"printf 'acknowledged' > "$0"; kill -9 $$"#)
+ .arg(&path)
+ .status()
+ .expect("spawn a writer that dies before any barrier");
+ assert!(
+ !status.success(),
+ "the writer exited normally, so it is not modelling a crash"
+ );
+
+ let survived = std::fs::read(&path).unwrap_or_default();
+ assert_eq!(
+ survived.as_slice(),
+ b"acknowledged",
+ "an unsynced write did not survive SIGKILL, so the persisted cases in
this file may be probing barrier order after all: {}",
+ String::from_utf8_lossy(&survived)
+ );
+}
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 630189f9e..b4849701b 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -8449,7 +8449,7 @@ mod tests {
const TEST_CLUSTER: u128 = 1;
- fn checksummed_segment_prepare(
+ pub(super) fn checksummed_segment_prepare(
op: u64,
parent: u128,
offset: u64,
@@ -10659,7 +10659,7 @@ mod tests {
/// can assert on reply bytes without a connection registry (whose slot
/// guard would borrow the partition across `on_request(&mut self)`).
#[derive(Debug, Default)]
- struct RecordingBus {
+ pub(super) struct RecordingBus {
sent_to_clients: Rc<RefCell<Vec<(u128, Frozen<MESSAGE_ALIGN>)>>>,
sent_to_replicas: Rc<RefCell<Vec<(u8, Frozen<MESSAGE_ALIGN>)>>>,
}
@@ -10692,13 +10692,13 @@ mod tests {
fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
}
- type SentFrames = Rc<RefCell<Vec<(u128, Frozen<MESSAGE_ALIGN>)>>>;
+ pub(super) type SentFrames = Rc<RefCell<Vec<(u128,
Frozen<MESSAGE_ALIGN>)>>>;
fn recording_partition() -> (IggyPartition<RecordingBus>, SentFrames) {
recording_partition_at(0, 1)
}
- fn recording_partition_at(
+ pub(super) fn recording_partition_at(
replica: u8,
replica_count: u8,
) -> (IggyPartition<RecordingBus>, SentFrames) {
@@ -15148,6 +15148,71 @@ mod retention_tests {
}
}
+#[cfg(test)]
+mod review_4092_tests {
+ use super::tests::{checksummed_segment_prepare, recording_partition_at};
+ use super::*;
+
+ /// ENOSPC on 28 is the raw errno; `io::ErrorKind::StorageFull` is
unstable.
+ const ENOSPC: i32 = 28;
+
+ /// `tick_partitions` turns a partition's `fatal()` into a server shutdown
+ /// (`shard/src/lib.rs:7378-7382`). A refused offset write leaves prior
bytes
+ /// intact and nothing undefined, so it should fence the partition at
worst,
+ /// the way `mark_materialization_missing` and `partitions.tombstone`
already
+ /// do for an unserviceable namespace.
+ #[compio::test]
+ #[ignore = "PR #4092 review: a refused consumer-offset write raises
`FatalCommit`, which the shard pump converts into a whole-node shutdown"]
+ async fn
given_a_full_disk_when_driving_persistence_then_only_the_partition_should_fence()
{
+ let directory = tempfile::tempdir().unwrap();
+ let (mut partition, _replies) = recording_partition_at(0, 3);
+
partition.set_partition_dir(directory.path().to_string_lossy().into_owned());
+ partition.runtime_options.consumer_offset_durability =
iggy_common::Durability::Persisted;
+ partition.open_persistence().await.unwrap();
+ let persistence = Rc::clone(partition.persistence.as_ref().unwrap());
+
+ persistence.fail_operation(
+ std::io::Error::from_raw_os_error(ENOSPC),
+ Operation::StoreConsumerOffset,
+ );
+ partition.drive_persistence().await;
+
+ assert!(
+ partition.fatal().is_none(),
+ "a refused consumer-offset write raised FatalCommit, which the
shard pump converts into a whole-node shutdown; every other partition on the
core is taken down with it, including topics with no persisted policy"
+ );
+ }
+
+ /// Contested between reviewers: distsys argued the pre-checks leave only a
+ /// genuinely divergent prepare here, where fail-closed is defensible;
storage
+ /// argued the same errno classification fix covers both call sites.
Recorded
+ /// so the decision is explicit rather than implied by a missing test.
+ #[compio::test]
+ #[ignore = "PR #4092 review: CONTESTED between reviewers -- whether a
divergent prepare at an already-accepted op should latch the whole partition"]
+ async fn
given_a_divergent_prepare_when_submitting_then_the_partition_should_not_latch()
{
+ let directory = tempfile::tempdir().unwrap();
+ let (mut partition, _replies) = recording_partition_at(0, 3);
+
partition.set_partition_dir(directory.path().to_string_lossy().into_owned());
+ partition.runtime_options.durability =
iggy_common::Durability::Persisted;
+ partition.open_persistence().await.unwrap();
+ let persistence = Rc::clone(partition.persistence.as_ref().unwrap());
+
+ let first = checksummed_segment_prepare(1, 0, 0, b"first");
+ assert!(partition.submit_prepare_persistence(first.into_frozen(),
Operation::SendMessages));
+
+ // Same op, different bytes: `append` answers InvalidData, which is not
+ // `WouldBlock`, so `:1192-1193` latches the whole partition.
+ let divergent = checksummed_segment_prepare(1, 0, 0, b"divergent");
+ assert!(
+ !partition.submit_prepare_persistence(divergent.into_frozen(),
Operation::SendMessages)
+ );
+ assert!(
+ persistence.failure().is_none(),
+ "a single refused prepare latched the partition permanently;
`failure` has no clearing path, so every later durability query answers false"
+ );
+ }
+}
+
#[cfg(test)]
mod purge_floor_tests {
use super::tests::{
diff --git a/core/partitions/src/persistence.rs
b/core/partitions/src/persistence.rs
index 57b45af04..6a1e4b3dc 100644
--- a/core/partitions/src/persistence.rs
+++ b/core/partitions/src/persistence.rs
@@ -1305,6 +1305,49 @@ mod tests {
use server_common::{Message, iobuf::Owned};
use tempfile::tempdir;
+ /// `io::ErrorKind::StorageFull` is still unstable, so match the raw errno.
+ const ENOSPC: i32 = 28;
+
+ /// A full disk is a refused write, not a torn one: the prior bytes are
+ /// intact and nothing is undefined. Latching it fences the partition for
the
+ /// life of the process, and `drive_persistence` escalates that to a node
+ /// shutdown. The same function already treats open failures as retriable.
+ #[compio::test]
+ #[ignore = "PR #4092 review: a refused consumer-offset write latches an
unclearable `failure`, retroactively reporting already-acked prepares as
unwritten"]
+ async fn
given_a_full_disk_when_the_offset_write_is_refused_then_persistence_should_not_fence()
+ {
+ let directory = tempdir().unwrap();
+ let (persistence, _) =
+ PartitionPersistence::open(&directory.path().join("prepares-7"),
42, 7)
+ .await
+ .unwrap();
+ let first = prepare(1, 0);
+ persistence
+ .append(first.clone().into_frozen(), true)
+ .unwrap();
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ assert!(persistence.is_written(first.header()));
+
+ persistence.fail_operation(
+ io::Error::from_raw_os_error(ENOSPC),
+ Operation::StoreConsumerOffset,
+ );
+
+ assert!(
+ persistence.is_written(first.header()),
+ "an unrelated offset write reported an already-durable prepare as
unwritten"
+ );
+ assert!(
+ persistence.failure().is_none(),
+ "ENOSPC on one consumer-offset record latched the partition;
nothing clears `failure` (its only `None` is the constructor), so
`is_written_through` stays false and `drive_persistence` raises FatalCommit"
+ );
+ assert!(
+ persistence.start(),
+ "the writer refuses to start again after a recoverable errno"
+ );
+ }
+
#[compio::test]
async fn
completion_is_generation_scoped_and_buffered_work_does_not_ack_durability() {
let directory = tempdir().unwrap();
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 747e5b70b..7ad93eaa4 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -7784,3 +7784,56 @@ mod metadata_read_frontier_tests {
.commit_min()
}
}
+
+#[cfg(test)]
+mod review_4092_dst_tests {
+ //! The deterministic simulator is this repo's strongest correctness tool
and
+ //! it is hard-asserted out of the feature this PR adds: `init_partition`
+ //! (`core/shard/src/lib.rs:4193-4198`) panics for any topic whose
durability
+ //! or consumer-offset durability is persisted. So `PrepareOk` gating,
log-view
+ //! certification, commit ordering and quorum durability have no fault
+ //! coverage at any granularity.
+
+ use super::*;
+
+ #[test]
+ #[ignore = "PR #4092 review: no simulator API can seed a persisted topic,
and `init_partition` asserts them out of the cluster simulator entirely"]
+ fn
given_the_cluster_simulator_when_seeding_a_topic_then_a_persisted_policy_should_be_expressible()
+ {
+ let replica_count = 3u8;
+ let client: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(replica_count as usize,
[client].into_iter(), network_opts);
+ for _ in 0..100 {
+ sim.step();
+ }
+
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.seed_stream_topic_partition(namespace);
+
+ let options = sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|inner| {
+ inner
+ .items
+ .get(namespace.stream_id())
+ .and_then(|stream| stream.topics.get(namespace.topic_id()))
+ .map(|topic| {
+
iggy_common::TopicRuntimeOptions::from_resource_options(&topic.options)
+ })
+ .unwrap_or_default()
+ });
+
+ assert!(
+ options.durability.is_persisted() ||
options.consumer_offset_durability.is_persisted(),
+ "no simulator API can seed a persisted topic, so the durability
guarantee this PR adds is unreachable from the deterministic simulator and
`init_partition` asserts it out anyway"
+ );
+ }
+}
diff --git a/core/simulator/src/storage.rs b/core/simulator/src/storage.rs
index d8100dc09..e9a47a8d8 100644
--- a/core/simulator/src/storage.rs
+++ b/core/simulator/src/storage.rs
@@ -21,12 +21,17 @@
use journal::durable_storage::{DurableFile, DurableStorage, OpenMode,
StorageEntry};
use server_common::iobuf::Frozen;
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
use std::collections::BTreeMap;
use std::ffi::OsString;
use std::io;
use std::path::{Component, Path};
use std::rc::Rc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Each `SimStorage` is an independent filesystem, so writer identities must
not
+/// collide between instances the way bare paths would.
+static NEXT_FILESYSTEM: AtomicU64 = AtomicU64::new(0);
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Crash {
@@ -39,6 +44,11 @@ pub enum FaultMode {
Before,
After,
TornWrite,
+ /// A short write the kernel never reports. `TornWrite` returns an error,
so
+ /// every caller learns the record is incomplete; a device that writes
half a
+ /// block and completes the operation tells nobody, and only the record's
own
+ /// checksum can refuse it on the way back.
+ SilentTornWrite,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -69,6 +79,9 @@ pub struct SimFile {
storage: SimStorage,
inode: usize,
epoch: u64,
+ /// errseq sample taken at open. A writeback failure recorded before this
+ /// handle existed is invisible to it, exactly as on Linux.
+ error_seen: Cell<u64>,
}
#[derive(Clone)]
@@ -85,10 +98,12 @@ enum Inode {
#[derive(Clone)]
struct State {
+ id: u64,
inodes: Vec<Inode>,
epoch: u64,
trace: Vec<StorageOperation>,
written_bytes: BTreeMap<usize, usize>,
+ write_errors: BTreeMap<usize, u64>,
fault: Option<(usize, FaultMode)>,
paused: Option<StorageOperation>,
waiters: Vec<std::task::Waker>,
@@ -139,6 +154,47 @@ impl SimStorage {
}
}
+ /// Writeback is per-page and unordered: pages from `first` on reach stable
+ /// storage while everything below them stays dirty and is lost to power
+ /// loss. Whole-inode [`Self::writeback`] cannot produce that state, so a
+ /// recovery walk that trusts its prefix passes under it and fails here.
+ ///
+ /// # Panics
+ /// Panics if `page_size` is zero.
+ pub fn writeback_from_page(&self, page_size: usize, first: usize) {
+ assert!(page_size > 0, "page size must be positive");
+ for inode in &mut self.state.borrow_mut().inodes {
+ if let Inode::File { buffered, stable } = inode {
+ let start = first.saturating_mul(page_size);
+ if start >= buffered.len() {
+ continue;
+ }
+ stable.resize(buffered.len(), 0);
+ stable[start..].copy_from_slice(&buffered[start..]);
+ }
+ }
+ }
+
+ /// Fail the inode's pending writeback the way a failing device does: the
+ /// dirty pages above the last barrier are dropped and unrecoverable, and
the
+ /// error is reported once to each handle that was already open. A handle
+ /// opened afterwards samples the current sequence and sees success over
the
+ /// same lost bytes, which is what makes a fresh-descriptor `fsync` an
+ /// unsound barrier for writes issued through a different one.
+ ///
+ /// # Errors
+ /// Returns an error if `path` does not resolve to a file.
+ pub fn fail_writeback(&self, path: &Path) -> io::Result<()> {
+ let mut state = self.state.borrow_mut();
+ let inode = state.lookup(path)?;
+ match &mut state.inodes[inode] {
+ Inode::File { buffered, stable } => buffered.clone_from(stable),
+ Inode::Directory { .. } => return Err(invalid("writeback failure
on a directory")),
+ }
+ *state.write_errors.entry(inode).or_default() += 1;
+ Ok(())
+ }
+
pub fn pause_writes(&self) {
self.state.borrow_mut().paused = Some(StorageOperation::Write);
}
@@ -194,8 +250,12 @@ impl SimStorage {
if mode == Some(FaultMode::Before) {
return Err(io::Error::other("injected storage failure"));
}
- let result = action(&mut state, mode == Some(FaultMode::TornWrite))?;
- if mode.is_some() {
+ let torn = matches!(
+ mode,
+ Some(FaultMode::TornWrite | FaultMode::SilentTornWrite)
+ );
+ let result = action(&mut state, torn)?;
+ if mode.is_some() && mode != Some(FaultMode::SilentTornWrite) {
return Err(io::Error::other("injected failure after storage
effect"));
}
Ok(result)
@@ -205,6 +265,17 @@ impl SimStorage {
impl DurableStorage for SimStorage {
type File = SimFile;
+ fn writer_identity(&self, path: &Path) ->
io::Result<Option<std::path::PathBuf>> {
+ // The epoch is the simulated process incarnation. A lease that a
cancelled
+ // writer left interrupted is fenced until the process holding it
dies, so
+ // an identity that survived `Crash::Process` could never reopen.
+ let state = self.state.borrow();
+ let process = format!("sim-{}-{}", state.id, state.epoch);
+ Ok(Some(
+
std::path::PathBuf::from(process).join(path.strip_prefix("/").unwrap_or(path)),
+ ))
+ }
+
async fn open(&self, path: &Path, mode: OpenMode) -> io::Result<SimFile> {
let creates = matches!(mode, OpenMode::Create |
OpenMode::CreateOrOpen);
let operation = if creates {
@@ -213,7 +284,7 @@ impl DurableStorage for SimStorage {
StorageOperation::Open
};
self.wait_for(operation).await;
- let (inode, epoch) = self.perform(operation, |state, _| {
+ let (inode, epoch, errors) = self.perform(operation, |state, _| {
let inode = if creates {
let (parent, name) = state.parent(path)?;
if let Some(&inode) = state.directory(parent)?.get(&name) {
@@ -240,12 +311,14 @@ impl DurableStorage for SimStorage {
} else {
state.lookup(path)?
};
- Ok((inode, state.epoch))
+ let errors =
state.write_errors.get(&inode).copied().unwrap_or_default();
+ Ok((inode, state.epoch, errors))
})?;
Ok(SimFile {
storage: self.clone(),
inode,
epoch,
+ error_seen: Cell::new(errors),
})
}
@@ -440,6 +513,17 @@ impl DurableFile for SimFile {
self.storage
.perform(StorageOperation::FileSync, |state, _| {
state.file(self.inode, self.epoch)?;
+ let errors = state
+ .write_errors
+ .get(&self.inode)
+ .copied()
+ .unwrap_or_default();
+ if errors > self.error_seen.get() {
+ self.error_seen.set(errors);
+ return Err(io::Error::other(
+ "writeback failed before this handle synced",
+ ));
+ }
if let Inode::File { buffered, stable } = &mut
state.inodes[self.inode] {
stable.clone_from(buffered);
}
@@ -489,10 +573,12 @@ impl SimFile {
impl Default for State {
fn default() -> Self {
Self {
+ id: NEXT_FILESYSTEM.fetch_add(1, Ordering::Relaxed),
inodes: vec![Inode::directory()],
epoch: 0,
trace: Vec::new(),
written_bytes: BTreeMap::new(),
+ write_errors: BTreeMap::new(),
fault: None,
paused: None,
waiters: Vec::new(),
diff --git a/core/simulator/src/storage/tests.rs
b/core/simulator/src/storage/tests.rs
index 190140171..c64ecb5dd 100644
--- a/core/simulator/src/storage/tests.rs
+++ b/core/simulator/src/storage/tests.rs
@@ -23,7 +23,9 @@ use consensus::MetadataHandle;
use futures::{executor::block_on, poll};
use iggy_binary_protocol::batch::BATCH_HEADER_SIZE;
use iggy_binary_protocol::{Command, Operation, PrepareHeader};
-use journal::partition_journal::{PARTITION_WAL_BLOCK_SIZE, SegmentPosition,
SegmentReference};
+use journal::partition_journal::{
+ PARTITION_WAL_BLOCK_SIZE, SegmentPosition, SegmentReference, record_length,
+};
use journal::{DurableAppend, PartitionPrepareJournal};
use partitions::{PartitionPersistence, install_backup};
use server_common::send_messages::{
@@ -848,6 +850,7 @@ fn first_open_recovers_after_each_initialization_fault() {
}
#[test]
+#[ignore = "PR #4092 review: PRE-EXISTING test. It passed vacuously while
`SimStorage::writer_identity` returned `None` and never took a lease; now that
the lease is real it is blocked on the compio-bound drain wait"]
fn deleting_and_recreating_a_partition_fences_an_old_writer_completion() {
block_on(async {
let storage = storage_for_partition().await;
@@ -2517,3 +2520,330 @@ fn prepare_with_payload(op: u64, parent: u128, payload:
&[u8]) -> Message<Prepar
header.checksum = header.identity_checksum();
Message::try_from(buffer).unwrap()
}
+
+// Regressions for PR #4092 review findings. Each one fails on the current tree
+// and names the defect it pins.
+
+const LARGE_BATCH_BYTES: usize = 1024 * 1024;
+
+#[test]
+fn
given_tail_pages_reached_disk_when_power_loss_then_recovery_should_not_surface_a_holed_record()
{
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let mut journal =
+ PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ let first = prepare(1, 0);
+ journal.append(first.clone().into_frozen()).await.unwrap();
+ let second = prepare(2, first.header().checksum);
+ journal
+ .append_buffered(second.clone().into_frozen())
+ .await
+ .unwrap();
+ drop(journal);
+
+ // The durable first record fills the blocks below `header_page`; the
+ // buffered second one starts there and runs to the end of the file.
+ let header_page = record_blocks(&first);
+ let data = Path::new("/partition/wal/prepares-0.wal");
+ let cached = read_all(&storage, data).await;
+ assert_eq!(
+ cached.len(),
+ (header_page + record_blocks(&second)) * PARTITION_WAL_BLOCK_SIZE
+ );
+ let hole =
+ header_page * PARTITION_WAL_BLOCK_SIZE..(header_page + 1) *
PARTITION_WAL_BLOCK_SIZE;
+ assert!(cached[hole.clone()].iter().any(|byte| *byte != 0));
+
+ // Background writeback preserved the later pages of the buffered
record
+ // and left its first page dirty. `writeback()` cannot express this,
+ // which is why `segment_recovery.rs` documents a byte-zero walk that
+ // nothing exercises.
+ storage.writeback_from_page(PARTITION_WAL_BLOCK_SIZE, header_page + 1);
+ storage.crash(Crash::PowerLoss);
+
+ let survived = read_all(&storage, data).await;
+ assert_eq!(survived.len(), cached.len());
+ assert!(
+ survived[hole.clone()].iter().all(|byte| *byte == 0),
+ "the record's first page reached stable storage, so there is no
hole to recover across"
+ );
+ assert_eq!(
+ survived[hole.end..],
+ cached[hole.end..],
+ "the record's later pages were lost too, so this is a short tail
rather than a hole"
+ );
+
+ let recovered =
+ PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ assert!(
+ !recovered.contains(second.header()),
+ "a record whose first page never reached stable storage was
surfaced as recovered"
+ );
+ assert_eq!(
+ recovered.head(),
+ 1,
+ "recovery advanced its head over a page-level hole"
+ );
+ });
+}
+
+#[test]
+fn
given_a_silent_short_write_when_recovering_then_the_record_should_be_refused() {
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let mut journal =
+ PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ let first = prepare(1, 0);
+ journal.append(first.clone().into_frozen()).await.unwrap();
+
+ let second = prepare(2, first.header().checksum);
+ storage.clear_trace();
+ // The device completes a half-written record and reports success. Only
+ // the record checksum can refuse it; `TornWrite` always returns an
+ // error, so no existing case reaches this path.
+ storage.fail_at(0, FaultMode::SilentTornWrite);
+ journal
+ .append(second.clone().into_frozen())
+ .await
+ .expect("a silent short write is reported as success");
+ drop(journal);
+ storage.crash(Crash::PowerLoss);
+
+ // The record was acknowledged before its bytes were lost, so refusing
to
+ // open is the correct answer. Nothing could assert that before,
because
+ // `TornWrite` reports the short write and the append fails instead.
+ let Err(error) =
+ PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7,
storage).await
+ else {
+ panic!("a WAL missing acknowledged bytes opened successfully");
+ };
+ assert_eq!(error.kind(), io::ErrorKind::InvalidData);
+ assert!(
+ error.to_string().contains("lost acknowledged bytes"),
+ "unexpected refusal: {error}"
+ );
+ let _ = first;
+ });
+}
+
+#[test]
+#[ignore = "PR #4092 review: `checkpoint_files` syncs materialized files
through a descriptor opened after the writeback failure, which samples errseq
too late and reports success over lost bytes; it must sync through the writer
that issued them"]
+fn
given_a_failed_writeback_when_checkpointing_then_wal_history_should_not_be_reclaimed()
{
+ block_on(async {
+ let (storage, persistence) = queued_batch(4).await;
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ let path = Path::new("/partition/materialized");
+ let mut writer = storage.open(path, OpenMode::Create).await.unwrap();
+ writer.write(0, b"committed".to_vec()).await.unwrap();
+
+ // The device drops the dirty pages before the checkpoint's barrier.
The
+ // writer that issued them is the only handle told; the descriptor
+ // `checkpoint_files` opens afterwards samples errseq past the failure
+ // and reports a successful barrier over bytes that are already gone.
+ storage.fail_writeback(path).unwrap();
+ persistence.checkpoint_files(
+ 4,
+ vec![path.to_path_buf()],
+ vec![Path::new(DIRECTORY).to_path_buf()],
+ );
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+
+ assert!(
+ persistence.failure().is_some(),
+ "a checkpoint reported success over materialized bytes the device
dropped"
+ );
+ assert_eq!(persistence.checkpoint_op(), 0);
+ storage.crash(Crash::PowerLoss);
+ let recovered =
+ PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ assert_eq!(
+ recovered.checkpoint_op(),
+ 0,
+ "WAL history was reclaimed although its materialization never
reached stable storage"
+ );
+ assert_eq!(recovered.head(), 4);
+ });
+}
+
+#[test]
+fn
given_sim_storage_when_opening_persistence_then_the_writer_lease_should_be_taken()
{
+ block_on(async {
+ let storage = storage_for_partition().await;
+ // Without a `writer_identity`,
`PartitionPersistence::open_with_capacity`
+ // sets `lease = None`, so WRITERS, the interrupted fence and the drain
+ // timeout have no coverage in any simulator test.
+ assert!(
+ DurableStorage::writer_identity(&storage, Path::new(DIRECTORY))
+ .unwrap()
+ .is_some(),
+ "simulator storage reports no writer identity, so every fault test
runs without a lease"
+ );
+ let (persistence, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ persistence.retire();
+ });
+}
+
+#[test]
+fn
given_an_interrupted_writer_when_the_process_restarts_then_the_partition_should_reopen()
{
+ block_on(async {
+ let (storage, persistence) = queued_batch(1).await;
+ storage.pause_writes();
+ assert!(persistence.start());
+ let mut writer = Box::pin(Rc::clone(&persistence).run());
+ assert!(poll!(&mut writer).is_pending());
+ // Cancelling the writer mid-mutation leaves its lease interrupted, a
+ // fence only the death of the process holding it may lift.
+ drop(writer);
+ storage.resume();
+ drop(persistence);
+ let fenced =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone()).await;
+ assert!(
+ fenced.is_err_and(|error| error.to_string().contains("requires
process restart")),
+ "an interrupted writer did not fence the partition within the same
process"
+ );
+
+ storage.crash(Crash::Process);
+ let reopened =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone()).await;
+ assert!(
+ reopened.is_ok(),
+ "the simulated restart kept the interrupted writer's identity, so
the partition can never reopen: {:?}",
+ reopened.err()
+ );
+ });
+}
+
+#[test]
+#[ignore = "PR #4092 review: append coalescing is gated on message-body bytes,
not the WAL extent, so batching is inert at the benchmarked batch sizes"]
+fn
given_large_bodies_when_appending_then_wal_records_should_coalesce_into_one_barrier_group()
{
+ block_on(async {
+ const PREPARES: u64 = 8;
+ let storage = storage_for_partition().await;
+ let (persistence, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ persistence.enable_segment_storage(
+ SegmentPosition::default(),
+ PREPARES * LARGE_BATCH_BYTES as u64,
+ );
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ persistence.take_metrics();
+
+ let mut parent = 0;
+ for offset in 0..PREPARES {
+ let prepare = owned_prepare_sized(offset + 1, parent, offset,
LARGE_BATCH_BYTES);
+ parent = prepare.header().checksum;
+ persistence.append(prepare.into_frozen(), true).unwrap();
+ }
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+
+ let metrics = persistence.take_metrics();
+ assert_eq!(metrics.batched_prepares, PREPARES);
+ // Under segment references a record occupies one 4 KiB extent, so all
+ // eight fit far inside APPEND_BATCH_BYTES_MAX. The gate measures the
+ // message body instead, so each prepare takes its own barrier group.
+ assert_eq!(
+ metrics.completed_batches,
+ 1,
+ "coalescing is gated on body bytes, so {PREPARES} prepares paid {}
barrier groups for {} bytes of WAL extent",
+ metrics.completed_batches,
+ PREPARES * PARTITION_WAL_BLOCK_SIZE as u64
+ );
+ });
+}
+
+fn owned_prepare_sized(
+ op: u64,
+ parent: u128,
+ offset: u64,
+ batch_bytes: usize,
+) -> Message<PrepareHeader> {
+ let payload = vec![
+ u8::try_from(op % 251).unwrap();
+ batch_bytes - BATCH_HEADER_SIZE - BATCH_MESSAGE_HEADER_SIZE
+ ];
+ let mut messages = IggyMessages::with_capacity(1);
+ messages.push(IggyMessage {
+ header: IggyMessageHeader {
+ id: u128::from(op),
+ payload_length: u32::try_from(payload.len()).unwrap(),
+ ..Default::default()
+ },
+ payload: payload.into(),
+ user_headers: None,
+ });
+ let mut batch =
+ SendMessagesOwned::from_messages(IggyNamespace::new(0, 0, 42),
&messages).unwrap();
+ batch.header.base_offset = offset;
+ batch.header.batch_checksum = batch.header.checksum_for_blob(&batch.blob);
+ let mut body = vec![0; BATCH_HEADER_SIZE + batch.blob.len()];
+ batch.header.encode_into(&mut body[..BATCH_HEADER_SIZE]);
+ body[BATCH_HEADER_SIZE..].copy_from_slice(&batch.blob);
+ assert_eq!(body.len(), batch_bytes);
+ prepare_with_payload(op, parent, &body).transmute_header(
+ |original, header: &mut PrepareHeader| {
+ *header = original;
+ header.checksum_body = 0;
+ header.checksum = header.identity_checksum();
+ },
+ )
+}
+
+fn record_blocks(prepare: &Message<PrepareHeader>) -> usize {
+ record_length(usize::try_from(prepare.header().size).unwrap()).unwrap()
+ / PARTITION_WAL_BLOCK_SIZE
+}
+
+async fn read_all(storage: &SimStorage, path: &Path) -> Vec<u8> {
+ let file = storage.open(path, OpenMode::Read).await.unwrap();
+ let length = usize::try_from(file.length().await.unwrap()).unwrap();
+ file.read(0, length).await.unwrap()
+}
+
+#[test]
+#[ignore = "PR #4092 review: `WriterLease::acquire` drains through
`compio::runtime::time::timeout`, so the writer fence cannot be driven by the
deterministic executor"]
+fn
given_a_retired_writer_when_reacquiring_then_the_drain_wait_should_be_executor_agnostic()
{
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let (first, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ first.append(prepare(1, 0).into_frozen(), true).unwrap();
+ storage.pause_writes();
+ assert!(first.start());
+ let mut writer = Box::pin(Rc::clone(&first).run());
+ assert!(poll!(&mut writer).is_pending());
+ first.retire();
+
+ // `WriterLease::acquire` waits for the previous writer to drain
through
+ // `compio::runtime::time::timeout` (`persistence.rs:220`), so the
fence
+ // cannot be driven by the deterministic executor at all. Every
+ // simulator fault case runs with `lease = None` for this reason.
+ let reacquired =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 8,
storage.clone()).await;
+ storage.resume();
+ writer.await;
+ assert!(
+ reacquired.is_ok(),
+ "retired writer could not be replaced under the deterministic
executor"
+ );
+ });
+}
