This is an automated email from the ASF dual-hosted git repository.
hubcio 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 7bd69670f refactor(partitions): make purge recovery testable with
SimStorage (#4240)
7bd69670f is described below
commit 7bd69670f9ad9946f6490359443ec25bf7e2a089
Author: diegomrsantos <[email protected]>
AuthorDate: Mon Sep 21 18:37:42 2026 +0200
refactor(partitions): make purge recovery testable with SimStorage (#4240)
---
Cargo.lock | 1 +
core/journal/Cargo.toml | 1 +
core/journal/src/durable_storage.rs | 192 +++++++++++++-
core/partitions/src/iggy_partition.rs | 137 ++++++++--
core/partitions/src/offset_storage.rs | 247 ++++++++++++------
core/server/src/lib.rs | 5 +-
core/server/src/offset_recovery.rs | 179 ++++++-------
core/server/src/partition_helpers.rs | 105 ++++++--
core/simulator/src/storage.rs | 10 +-
core/simulator/src/storage/purge.rs | 469 ++++++++++++++++++++++++++++++++++
core/simulator/src/storage/tests.rs | 2 +-
11 files changed, 1119 insertions(+), 229 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 27491fac9..43b1eeb33 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8062,6 +8062,7 @@ dependencies = [
"nix",
"server_common",
"tempfile",
+ "tokio",
"tracing",
"twox-hash",
]
diff --git a/core/journal/Cargo.toml b/core/journal/Cargo.toml
index 01443d492..bfbe72c15 100644
--- a/core/journal/Cargo.toml
+++ b/core/journal/Cargo.toml
@@ -35,6 +35,7 @@ futures = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
server_common = { workspace = true }
+tokio = { workspace = true }
tracing = { workspace = true }
twox-hash = { workspace = true }
diff --git a/core/journal/src/durable_storage.rs
b/core/journal/src/durable_storage.rs
index 3ac3cbcf4..fc6f148ed 100644
--- a/core/journal/src/durable_storage.rs
+++ b/core/journal/src/durable_storage.rs
@@ -22,16 +22,29 @@ use compio::fs::{File, OpenOptions};
use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
use futures::channel::oneshot;
use futures::lock::Mutex;
+use futures::{Stream, stream};
use server_common::iobuf::{Frozen, Owned};
use std::ffi::OsString;
use std::io;
-use std::path::Path;
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+use tokio::sync::{Semaphore, mpsc};
+use tracing::warn;
+
+const FILE_DIRECTORY_BUFFER: usize = 64;
+static FILE_DIRECTORY_READERS: Semaphore = Semaphore::const_new(4);
+
+/// Paths to regular files from one directory scan, followed by any scan error.
+/// Dropping the stream may cancel enumeration without waiting for its worker.
+pub type RegularFiles = Pin<Box<dyn Stream<Item = io::Result<PathBuf>>>>;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OpenMode {
Read,
ReadWrite,
Create,
+ /// Create or truncate a file without requiring read permission.
+ CreateWriteOnly,
/// Create a file if absent, preserving the inode and bytes if it exists.
CreateOrOpen,
}
@@ -43,6 +56,11 @@ pub struct StorageEntry {
/// Filesystem operations whose completion and persistence order affect
recovery.
/// Implementations must preserve open file identity across rename and unlink.
+///
+/// Completing a write or filesystem mutation does not by itself guarantee
+/// survival across a crash. File sync makes file contents durable; directory
+/// sync makes changes to that directory's entries durable. Syncing a parent
+/// directory does not sync changes inside its child directories.
pub trait DurableStorage {
type File: DurableFile;
@@ -75,9 +93,38 @@ pub trait DurableStorage {
/// # Errors
/// Returns the underlying filesystem error.
fn exists(&self, path: &Path) -> impl Future<Output = io::Result<bool>>;
+ /// Check whether a path resolves to an existing target, following
symbolic links.
+ /// Backends without symbolic links can use the default existence check.
+ ///
+ /// # Errors
+ /// Returns the underlying filesystem error. Missing targets return
`false`.
+ fn exists_following_links(&self, path: &Path) -> impl Future<Output =
io::Result<bool>> {
+ self.exists(path)
+ }
/// # Errors
/// Returns an error for unreadable directories or unsupported file types.
fn entries(&self, path: &Path) -> impl Future<Output =
io::Result<Vec<StorageEntry>>>;
+ /// Enumerate regular files without including directories.
+ ///
+ /// Disk scans skip unreadable entries and unsupported file types, and
bound
+ /// both worker concurrency and buffered paths. Other backends may use
their
+ /// existing directory enumeration through the default implementation.
+ ///
+ /// # Errors
+ /// Returns an error, or yields one in the stream, if the directory cannot
be
+ /// enumerated. Dropping a stream does not make any filesystem changes
durable.
+ fn regular_files(&self, path: &Path) -> impl Future<Output =
io::Result<RegularFiles>> {
+ async move {
+ let entries = self.entries(path).await?;
+ let path = path.to_path_buf();
+ Ok(Box::pin(stream::iter(
+ entries
+ .into_iter()
+ .filter(|entry| !entry.directory)
+ .map(move |entry| Ok(path.join(entry.name))),
+ )) as RegularFiles)
+ }
+ }
/// # Errors
/// Returns the underlying filesystem error. Missing paths are accepted.
fn remove_tree(&self, path: &Path) -> impl Future<Output = io::Result<()>>;
@@ -191,9 +238,16 @@ impl DurableStorage for DiskStorage {
async fn open(&self, path: &Path, mode: OpenMode) -> io::Result<File> {
let mut options = OpenOptions::new();
- options.read(true).write(mode != OpenMode::Read);
- if matches!(mode, OpenMode::Create | OpenMode::CreateOrOpen) {
- options.create(true).truncate(mode == OpenMode::Create);
+ options
+ .read(mode != OpenMode::CreateWriteOnly)
+ .write(mode != OpenMode::Read);
+ if matches!(
+ mode,
+ OpenMode::Create | OpenMode::CreateWriteOnly |
OpenMode::CreateOrOpen
+ ) {
+ options
+ .create(true)
+ .truncate(matches!(mode, OpenMode::Create |
OpenMode::CreateWriteOnly));
}
options.open(path).await
}
@@ -226,12 +280,76 @@ impl DurableStorage for DiskStorage {
}
}
+ async fn exists_following_links(&self, path: &Path) -> io::Result<bool> {
+ match compio::fs::metadata(path).await {
+ Ok(_) => Ok(true),
+ Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
+ Err(error) => Err(error),
+ }
+ }
+
async fn entries(&self, path: &Path) -> io::Result<Vec<StorageEntry>> {
// getdents has no io_uring operation, and shard fallback pools are
disabled.
let path = path.to_path_buf();
run_blocking("iggy-directory-scan", move ||
directory_entries(&path)).await
}
+ async fn regular_files(&self, path: &Path) -> io::Result<RegularFiles> {
+ // Compio has no asynchronous directory iterator and the shard's
blocking
+ // pool is disabled. The worker owns the permit so cancellation cannot
+ // exceed the concurrency bound.
+ let permit = FILE_DIRECTORY_READERS
+ .acquire()
+ .await
+ .map_err(io::Error::other)?;
+ let (sender, receiver) = mpsc::channel(FILE_DIRECTORY_BUFFER);
+ let directory = path.to_path_buf();
+ std::thread::Builder::new()
+ .name("iggy-file-scan".to_owned())
+ .spawn(move || {
+ let _permit = permit;
+ let result = (|| {
+ // An unreadable entry must not hide the remaining files.
+ // Only opening the directory fails the scan.
+ for entry in std::fs::read_dir(&directory)? {
+ let entry = match entry {
+ Ok(entry) => entry,
+ Err(error) => {
+ warn!(path = %directory.display(), %error,
"failed to read directory entry");
+ continue;
+ }
+ };
+ let is_file = match entry.file_type() {
+ Ok(file_type) => file_type.is_file(),
+ Err(error) => {
+ warn!(path = %directory.display(), %error,
"failed to read entry type");
+ continue;
+ }
+ };
+ if is_file &&
sender.blocking_send(Ok(Some(entry.path()))).is_err() {
+ return Ok(());
+ }
+ }
+ Ok(())
+ })();
+ // Explicit completion distinguishes an empty directory from an
+ // interrupted worker. Closed receivers abandon enumeration.
+ let _ = sender.blocking_send(result.map(|()| None));
+ })?;
+ Ok(Box::pin(stream::unfold(
+ Some(receiver),
+ |receiver| async move {
+ let mut receiver = receiver?;
+ match receiver.recv().await {
+ Some(Ok(Some(path))) => Some((Ok(path), Some(receiver))),
+ Some(Ok(None)) => None,
+ Some(Err(error)) => Some((Err(error), None)),
+ None => Some((Err(io::Error::other("directory reader
stopped")), None)),
+ }
+ },
+ )))
+ }
+
async fn remove_tree(&self, path: &Path) -> io::Result<()> {
let mut pending = vec![(path.to_path_buf(), false)];
while let Some((path, visited)) = pending.pop() {
@@ -378,7 +496,10 @@ fn directory_entries(path: &Path) ->
io::Result<Vec<StorageEntry>> {
#[cfg(test)]
mod tests {
- use super::{DiskStorage, DurableFile, DurableStorage, OpenMode,
run_blocking};
+ use super::{
+ DiskStorage, DurableFile, DurableStorage, FILE_DIRECTORY_BUFFER,
OpenMode, run_blocking,
+ };
+ use futures::TryStreamExt;
use futures::channel::oneshot;
use futures::future::{Either, select};
use std::io;
@@ -387,6 +508,67 @@ mod tests {
const WORKER_TIMEOUT: Duration = Duration::from_secs(5);
+ #[compio::test]
+ async fn
given_symlink_when_target_disappears_should_report_target_absent() {
+ let directory = tempfile::tempdir().unwrap();
+ let target = directory.path().join("target");
+ let link = directory.path().join("link");
+ std::fs::write(&target, []).unwrap();
+ std::os::unix::fs::symlink(&target, &link).unwrap();
+
+ assert!(DiskStorage.exists_following_links(&target).await.unwrap());
+ assert!(DiskStorage.exists_following_links(&link).await.unwrap());
+
+ std::fs::remove_file(&target).unwrap();
+
+ assert!(DiskStorage.exists(&link).await.unwrap());
+ assert!(!DiskStorage.exists_following_links(&link).await.unwrap());
+ }
+
+ #[compio::test]
+ async fn
given_buffered_file_scan_when_cancelled_should_release_worker_capacity() {
+ let directory = tempfile::tempdir().unwrap();
+ let entry_count = FILE_DIRECTORY_BUFFER * 2;
+ for consumer_id in 0..entry_count {
+ std::fs::write(directory.path().join(consumer_id.to_string()),
[]).unwrap();
+ }
+
+ // More cancellations than available workers expose permits retained by
+ // a worker whose receiver has gone away.
+ for _ in 0..8 {
+ let entries =
DiskStorage.regular_files(directory.path()).await.unwrap();
+ drop(entries);
+ }
+
+ let paths: Vec<_> = DiskStorage
+ .regular_files(directory.path())
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(paths.len(), entry_count);
+ }
+
+ #[compio::test]
+ async fn given_non_regular_entries_when_scanning_files_should_skip_them() {
+ let directory = tempfile::tempdir().unwrap();
+ let regular_file = directory.path().join("1");
+ std::fs::write(®ular_file, []).unwrap();
+ std::fs::create_dir(directory.path().join("2")).unwrap();
+ std::os::unix::fs::symlink(®ular_file,
directory.path().join("3")).unwrap();
+
+ let paths: Vec<_> = DiskStorage
+ .regular_files(directory.path())
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_eq!(paths, [regular_file]);
+ }
+
#[compio::test]
async fn cancelled_blocking_operation_keeps_its_permit_until_completion() {
let executor_thread = std::thread::current().id();
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 5e39d0c3e..8f7f4f100 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -24,8 +24,9 @@ use crate::log::JournalInfo;
use crate::log::SegmentedLog;
use crate::messages_writer::MessagesWriter;
use crate::offset_storage::{
- PURGE_GENERATION_FILE, delete_persisted_offset, persist_offset,
persist_offset_max,
- persist_purge_generation, read_purge_generation,
+ PURGE_GENERATION_FILE, delete_persisted_offset,
delete_persisted_offset_with_storage,
+ persist_offset, persist_offset_max, persist_purge_generation_with_storage,
+ read_purge_generation,
};
use crate::persistence::{PartitionPersistence, PersistenceCompletion,
PersistenceNotifier};
use crate::poll_plan::{
@@ -50,6 +51,7 @@ use consensus::{
replicate_frozen_to_next_in_chain, replicate_preflight,
report_uncommittable_head,
restamp_prepare_view, send_prepare_ok as send_prepare_ok_common,
verify_prepare_integrity,
};
+use futures::{StreamExt, TryStreamExt};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffsetRequest, StoreConsumerOffsetRequest,
@@ -67,6 +69,7 @@ use iggy_common::{
TopicRuntimeOptions,
};
use journal::Journal as _;
+use journal::durable_storage::{DiskStorage, DurableStorage};
use journal::local_gate::LocalGate;
use journal::superblock::{
PingPongSuperblock, SUPERBLOCK_RETRY_BACKOFF_BASE_MICROS,
SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS,
@@ -88,6 +91,7 @@ use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt;
use std::hash::Hash;
use std::num::NonZeroU32;
+use std::path::Path;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
@@ -695,10 +699,26 @@ where
/// sentinel 0 instead would make the reconciler silently re-purge and
/// destroy post-purge messages, so the boot fails loud.
pub async fn hydrate_applied_purge_generation(&mut self) -> Result<(),
IggyError> {
+ self.hydrate_applied_purge_generation_with_storage(&DiskStorage)
+ .await
+ }
+
+ /// Restore the applied generation from the filesystem used for purge
cleanup.
+ ///
+ /// Set the partition directory and its creation revision before calling
this.
+ /// A simulator must call this on a new partition after discarding its
volatile
+ /// state, so completion is recovered from storage instead of retained in
memory.
+ ///
+ /// # Errors
+ /// Propagates failures reading the marker, as the disk recovery entry
point does.
+ pub async fn hydrate_applied_purge_generation_with_storage<S:
DurableStorage>(
+ &mut self,
+ storage: &S,
+ ) -> Result<(), IggyError> {
if let Some(dir) = self.partition_dir() {
let path = format!("{dir}/{PURGE_GENERATION_FILE}");
self.applied_purge_generation =
- read_purge_generation(&path, self.created_revision).await?;
+ read_purge_generation(storage, &path,
self.created_revision).await?;
}
Ok(())
}
@@ -7814,6 +7834,46 @@ where
self.installed_frontier = None;
self.segment_checksum_cache.borrow_mut().clear();
+ self.complete_purge_with_storage(&DiskStorage, generation)
+ .await
+ }
+
+ /// Clear consumer progress and record completion after resetting message
history.
+ ///
+ /// Completion proceeds through three stages:
+ /// 1. Clear live consumer and group bookmarks, delete their files, and
sync
+ /// each offset directory so the deletions can survive power loss.
+ /// 2. Reset offset bookkeeping and prevent old journal entries from being
+ /// applied or served as messages again.
+ /// 3. Persist the purge generation before advancing the live generation,
+ /// then invalidate cached state transfer offers and restamp the
frontier.
+ ///
+ /// Message history must already be reset, as [`Self::purge`] does before
+ /// entering this phase. The caller must exclude concurrent writes
throughout.
+ /// Retry behavior remains in [`Self::purge`], including its deferral and
+ /// message reset decisions. Storage controls the offset files and
completion
+ /// marker; journal and superblock operations still use the implementations
+ /// attached to this partition.
+ ///
+ /// Offset deletion and directory sync failures are logged and completion
+ /// continues. Keeping that decision here makes a storage harness exercise
the
+ /// same failure behavior as the server. Consequently, success does not
prove
+ /// that all bookmark deletions are durable: syncing the generation
marker's
+ /// parent does not sync the separate consumer and group directories.
+ ///
+ /// # Errors
+ /// Returns [`PurgeError::GenerationNotRecorded`] if the completion marker
+ /// cannot be persisted. Cleanup is not rolled back, the applied generation
+ /// remains unchanged, and the flag that defers prepare acknowledgements is
+ /// set. The normal purge path manages that flag when retrying.
+ #[allow(clippy::too_many_lines)]
+ pub async fn complete_purge_with_storage<S: DurableStorage>(
+ &mut self,
+ storage: &S,
+ generation: u64,
+ ) -> Result<(), PurgeError> {
+ let namespace = self.namespace();
+
// Clear consumer + consumer-group offsets (memory + disk). Collect the
// file paths before deleting so the map guard is not held across an
// await.
@@ -7849,28 +7909,25 @@ where
// full reset, and an offset file the live map never held -- a
pre-purge
// op re-persisted by journal repair on a restarted replica -- would
// otherwise survive for boot to hydrate back.
- let strayed_consumers =
-
crate::state_transfer::strayed_offset_files(self.consumer_offsets_path.as_deref())
- .into_iter()
- .filter_map(|path| {
- crate::state_transfer::numeric_offset_id(&path)
- .map(|id| (ConsumerKind::Consumer, id, path))
- });
- let strayed_groups = crate::state_transfer::strayed_offset_files(
+ let strayed_consumers = purge_offset_files(
+ storage,
+ self.consumer_offsets_path.as_deref(),
+ ConsumerKind::Consumer,
+ )
+ .await;
+ let strayed_groups = purge_offset_files(
+ storage,
self.consumer_group_offsets_path.as_deref(),
+ ConsumerKind::ConsumerGroup,
)
- .into_iter()
- .filter_map(|path| {
- crate::state_transfer::numeric_offset_id(&path)
- .map(|id| (ConsumerKind::ConsumerGroup, id, path))
- });
+ .await;
for (kind, consumer_id, path) in consumer_paths
.into_iter()
.chain(group_paths)
.chain(strayed_consumers)
.chain(strayed_groups)
{
- if let Err(error) = delete_persisted_offset(&path).await {
+ if let Err(error) = delete_persisted_offset_with_storage(storage,
&path).await {
self.consumer_offset_capacity_for(kind)
.record_stranded(consumer_id);
warn!(
@@ -7904,7 +7961,7 @@ where
.into_iter()
.chain(self.consumer_group_offsets_path.clone())
{
- if let Err(error) = crate::state_transfer::fsync_dir(&dir).await {
+ if let Err(error) = storage.sync_directory(Path::new(&dir)).await {
warn!(
target: "iggy.partitions.diag",
plane = "partitions",
@@ -7980,8 +8037,13 @@ where
// recorded the generation keep it.
if let Some(dir) = self.partition_dir() {
let path = format!("{dir}/{PURGE_GENERATION_FILE}");
- if let Err(error) =
- persist_purge_generation(&path, generation,
self.created_revision).await
+ if let Err(error) = persist_purge_generation_with_storage(
+ storage,
+ &path,
+ generation,
+ self.created_revision,
+ )
+ .await
{
self.purge_deferred = true;
warn!(
@@ -8416,6 +8478,41 @@ where
}
}
+async fn purge_offset_files<S: DurableStorage>(
+ storage: &S,
+ directory: Option<&str>,
+ kind: ConsumerKind,
+) -> Vec<(ConsumerKind, u32, String)> {
+ let Some(directory) = directory else {
+ return Vec::new();
+ };
+ let entries =
futures::stream::once(storage.regular_files(Path::new(directory))).try_flatten();
+ futures::pin_mut!(entries);
+ let mut offsets = Vec::new();
+ while let Some(entry) = entries.next().await {
+ let path = match entry {
+ Ok(path) => path,
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ path = directory,
+ %error,
+ "failed to scan consumer offset directory during purge"
+ );
+ continue;
+ }
+ };
+ let Some(path) = path.to_str() else {
+ continue;
+ };
+ if let Some(consumer_id) =
crate::state_transfer::numeric_offset_id(path) {
+ offsets.push((kind, consumer_id, path.to_owned()));
+ }
+ }
+ offsets
+}
+
/// Automatic commits remain monotone because an earlier poll can commit after
/// a later one. Explicit stores may intentionally rewind the cursor.
fn upsert_committed_offset<K>(
diff --git a/core/partitions/src/offset_storage.rs
b/core/partitions/src/offset_storage.rs
index aaf341412..266f5a280 100644
--- a/core/partitions/src/offset_storage.rs
+++ b/core/partitions/src/offset_storage.rs
@@ -15,12 +15,25 @@
// specific language governing permissions and limitations
// under the License.
+//! Store consumer bookmarks and the partition's applied purge generation.
+//!
+//! A bookmark records the last consumed offset. The purge marker instead
records
+//! which reset was applied to this incarnation of the partition. Recovery uses
+//! them to restore progress and decide whether a purge must be repeated.
+//!
+//! Functions ending in `_with_storage` share the persistence sequence between
+//! real disk and simulated storage. File sync makes record contents durable;
+//! directory sync makes creation, replacement, or deletion durable. Offset
callers
+//! own that directory sync, while purge marker writes include it before
returning.
+
+use std::{io, path::Path};
+
use compio::{
- fs::{OpenOptions, create_dir_all, remove_file, rename},
- io::{AsyncReadAt, AsyncReadAtExt, AsyncWriteAtExt},
+ fs::{OpenOptions, remove_file, rename},
+ io::{AsyncReadAt, AsyncWriteAtExt},
};
use iggy_common::{IggyError, calculate_checksum};
-use std::{io, path::Path};
+use journal::durable_storage::{DiskStorage, DurableFile, DurableStorage,
OpenMode};
use tracing::warn;
const OFFSET_SIZE: usize = core::mem::size_of::<u64>();
@@ -122,11 +135,28 @@ pub fn decode_offset_record(bytes: &[u8]) -> OffsetRecord
{
/// # Errors
/// [`IggyError`] when the directory, file, or write cannot be created or
completed.
pub async fn persist_offset(path: &str, offset: u64, persisted: bool) ->
Result<(), IggyError> {
+ persist_offset_with_storage(&DiskStorage, path, offset, persisted).await
+}
+
+/// Persist a consumer offset through the supplied storage backend.
+///
+/// Uses the same record and replacement policy as [`persist_offset`]. When
+/// `persisted` is true, the caller must still sync the parent directory before
+/// treating the replacement name as durable. Calls for one path must be
serialized.
+///
+/// # Errors
+/// Returns the directory, open, or write error from [`persist_offset`].
+pub async fn persist_offset_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+ offset: u64,
+ persisted: bool,
+) -> Result<(), IggyError> {
let record = encode_offset_record(offset);
if persisted {
- replace_file(path, record, true, false).await
+ replace_file(storage, path, record, true, false).await
} else {
- write_in_place(path, record).await
+ write_in_place(storage, path, record).await
}
}
@@ -160,28 +190,33 @@ pub async fn persist_offset_retained(
Ok((result, file))
}
-async fn write_in_place<const N: usize>(path: &str, record: [u8; N]) ->
Result<(), IggyError> {
- create_parent_dir(path).await?;
- let mut file = OpenOptions::new()
- .write(true)
- .create(true)
- .truncate(true)
- .open(path)
+async fn write_in_place<S: DurableStorage, const N: usize>(
+ storage: &S,
+ path: &str,
+ record: [u8; N],
+) -> Result<(), IggyError> {
+ create_parent_dir_with_storage(storage, path).await?;
+ let mut file = storage
+ .open(Path::new(path), OpenMode::CreateWriteOnly)
.await
.map_err(|_|
IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
- file.write_all_at(record, 0)
+ file.write(0, record.to_vec())
.await
- .0
.map_err(|_| IggyError::CannotWriteToFile)
}
async fn create_parent_dir(path: &str) -> Result<(), IggyError> {
- // No `exists()` probe first: that is a BLOCKING `std::path` stat on the
pump
- // in front of every write, which serialises a batched fan-out on stats
- // before it can submit any I/O. `create_dir_all` is already a no-op on an
- // existing directory.
+ create_parent_dir_with_storage(&DiskStorage, path).await
+}
+
+async fn create_parent_dir_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+) -> Result<(), IggyError> {
+ // Probing with `Path::exists()` blocks the pump before it can submit
writes.
+ // Creating an existing directory already succeeds without changing it.
if let Some(parent) = Path::new(path).parent() {
- create_dir_all(parent).await.map_err(|_| {
+ storage.create_directories(parent).await.map_err(|_| {
IggyError::CannotCreateConsumerOffsetsDirectory(parent.display().to_string())
})?;
}
@@ -190,8 +225,8 @@ async fn create_parent_dir(path: &str) -> Result<(),
IggyError> {
pub(crate) async fn stage_offset_replacement(path: &str, offset: u64) ->
Result<(), IggyError> {
// Install can remove old files before publishing replacements. Staging
- // must survive a crash regardless of the normal consumer-offset
durability policy.
- write_replacement(path, encode_offset_record(offset), true)
+ // must survive a crash regardless of the normal consumer offset
durability policy.
+ write_replacement(&DiskStorage, path, encode_offset_record(offset), true)
.await
.map(|_| ())
}
@@ -206,23 +241,25 @@ pub(crate) async fn discard_offset_replacement(path:
&str) {
let _ = remove_file(replacement_path(path)).await;
}
-async fn replace_file<const N: usize>(
+async fn replace_file<S: DurableStorage, const N: usize>(
+ storage: &S,
path: &str,
record: [u8; N],
persisted: bool,
sync_parent: bool,
) -> Result<(), IggyError> {
- let temporary = write_replacement(path, record, persisted).await?;
- if rename(&temporary, path).await.is_err() {
- let _ = remove_file(&temporary).await;
+ let temporary = write_replacement(storage, path, record, persisted).await?;
+ if storage
+ .rename(Path::new(&temporary), Path::new(path))
+ .await
+ .is_err()
+ {
+ let _ = storage.remove_file(Path::new(&temporary)).await;
return Err(IggyError::CannotWriteToFile);
}
if sync_parent && let Some(parent) = Path::new(path).parent() {
- let parent = compio::fs::File::open(parent)
- .await
- .map_err(|_| IggyError::CannotSyncFile)?;
- parent
- .sync_all()
+ storage
+ .sync_directory(parent)
.await
.map_err(|_| IggyError::CannotSyncFile)?;
}
@@ -230,32 +267,30 @@ async fn replace_file<const N: usize>(
Ok(())
}
-async fn write_replacement<const N: usize>(
+async fn write_replacement<S: DurableStorage, const N: usize>(
+ storage: &S,
path: &str,
record: [u8; N],
persisted: bool,
) -> Result<String, IggyError> {
- create_parent_dir(path).await?;
+ create_parent_dir_with_storage(storage, path).await?;
// Keep the previous cursor intact until the complete replacement exists.
- // A failed truncate-and-write otherwise turns a valid cursor into a torn
+ // A failed write after truncation otherwise turns a valid cursor into a
torn
// file that boot discards. The fixed sibling is safe because writes to one
// consumer key are serialized by the partition pump.
let temporary = replacement_path(path);
- let mut file = OpenOptions::new()
- .write(true)
- .create(true)
- .truncate(true)
- .open(&temporary)
+ let mut file = storage
+ .open(Path::new(&temporary), OpenMode::CreateWriteOnly)
.await
.map_err(|_|
IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
- if file.write_all_at(record, 0).await.0.is_err() {
- let _ = remove_file(&temporary).await;
+ if file.write(0, record.to_vec()).await.is_err() {
+ let _ = storage.remove_file(Path::new(&temporary)).await;
return Err(IggyError::CannotWriteToFile);
}
- if persisted && file.sync_data().await.is_err() {
- let _ = remove_file(&temporary).await;
+ if persisted && file.sync().await.is_err() {
+ let _ = storage.remove_file(Path::new(&temporary)).await;
return Err(IggyError::CannotWriteToFile);
}
drop(file);
@@ -346,61 +381,90 @@ pub async fn read_offset_max(path: &str, offset: u64) ->
Result<PersistedOffset,
/// Durably record the purge generation a partition has locally applied, keyed
/// to the incarnation (`created_revision`) it was applied for.
///
-/// Atomic replacement like [`persist_offset`] but ALWAYS data-synced,
regardless of
-/// the consumer-offset durability policy: purges are rare, the record is 16
bytes, and
+/// Atomic replacement like [`persist_offset`] but always synced, regardless of
+/// the consumer offset durability policy: purges are rare, the record is 16
bytes, and
/// a generation lost from the page cache in a crash makes the reconciler
-/// re-purge on restart, wiping messages appended after the purge. A failure
-/// leaves the previous record on disk so the caller keeps its in-memory
-/// applied generation old and retries.
+/// repeat the purge on restart, wiping messages appended after the purge.
+/// The parent directory is synced after the replacement is renamed into place.
+/// A failure before rename preserves the previous record. If the directory
sync
+/// fails, the replacement is visible but its survival across a crash is
uncertain.
///
/// # Errors
-/// Propagates the underlying open/write/sync failure.
+/// Propagates the underlying open, write, or sync failure.
pub async fn persist_purge_generation(
path: &str,
generation: u64,
created_revision: u64,
+) -> Result<(), IggyError> {
+ persist_purge_generation_with_storage(&DiskStorage, path, generation,
created_revision).await
+}
+
+/// Persist a purge completion marker through the supplied storage backend.
+///
+/// The record includes the partition incarnation. Both the replacement file
and
+/// its parent directory are synced as in [`persist_purge_generation`]. Calls
for
+/// one path must be serialized because they share a temporary filename.
+///
+/// # Errors
+/// Returns the directory, open, write, or sync error from
[`persist_purge_generation`].
+pub async fn persist_purge_generation_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+ generation: u64,
+ created_revision: u64,
) -> Result<(), IggyError> {
let mut record = [0u8; PURGE_GENERATION_RECORD_SIZE];
record[..OFFSET_SIZE].copy_from_slice(&generation.to_le_bytes());
record[OFFSET_SIZE..].copy_from_slice(&created_revision.to_le_bytes());
- replace_file(path, record, true, true).await
+ replace_file(storage, path, record, true, true).await
}
/// Read the purge generation this replica applied for the `created_revision`
-/// incarnation of the partition.
+/// incarnation of the partition through the supplied storage backend.
///
-/// Absent and torn files map to `Ok(0)`: both imply a purge died mid-write,
and
-/// `0` makes the reconciler re-apply the purge, the correct self-healing
-/// recovery for an idempotent wipe.
+/// Absent and torn files map to `Ok(0)`, which makes the reconciler apply any
+/// committed purge again. A failed existence probe is logged and treated as
absence.
///
-/// A record written for a DIFFERENT incarnation maps to `Ok(0)` too. A failed
+/// A record written for a different incarnation maps to `Ok(0)` too. A failed
/// `delete_partitions_from_disk` leaves the directory (and this file) behind;
/// the recreated topic's generations restart at 0, so hydrating the dead
/// incarnation's generation would swallow every purge of the new topic until
/// the committed counter climbed past it.
///
-/// A real I/O error propagates instead: collapsing it to `0` would re-purge a
+/// An open or read error propagates. Collapsing it to `0` would purge a
/// partition whose durable generation is intact but momentarily unreadable,
/// destroying every message appended after that purge.
///
/// # Errors
-/// Propagates a real open/read failure (anything but absent or short).
-pub async fn read_purge_generation(path: &str, created_revision: u64) ->
Result<u64, IggyError> {
- if !Path::new(path).exists() {
- return Ok(0);
+/// Propagates an open or read failure after the existence probe, except a
short read.
+pub async fn read_purge_generation<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+ created_revision: u64,
+) -> Result<u64, IggyError> {
+ match storage.exists_following_links(Path::new(path)).await {
+ Ok(true) => {}
+ Ok(false) => return Ok(0),
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ path,
+ %error,
+ "failed to check purge generation file, treating it as absent"
+ );
+ return Ok(0);
+ }
}
- let file = OpenOptions::new()
- .read(true)
- .open(path)
+ let file = storage
+ .open(Path::new(path), OpenMode::Read)
.await
.map_err(|_|
IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
- let buf = vec![0u8; PURGE_GENERATION_RECORD_SIZE];
- let compio::BufResult(read, buf) = file.read_exact_at(buf, 0).await;
- match read {
- Ok(()) => {}
+ let buf = match file.read(0, PURGE_GENERATION_RECORD_SIZE).await {
+ Ok(buf) => buf,
Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof =>
return Ok(0),
Err(_) => return
Err(IggyError::CannotReadConsumerOffsets(path.to_owned())),
- }
+ };
let (generation_bytes, revision_bytes) = buf.split_at(OFFSET_SIZE);
let generation = u64::from_le_bytes(
generation_bytes
@@ -458,10 +522,24 @@ async fn read_offset_record(path: &str) ->
Result<Option<OffsetRecord>, IggyErro
/// # Errors
/// Returns [`IggyError::CannotDeleteConsumerOffsetFile`] if the unlink fails.
pub async fn delete_persisted_offset(path: &str) -> Result<bool, IggyError> {
+ delete_persisted_offset_with_storage(&DiskStorage, path).await
+}
+
+/// Unlink a consumer offset through the supplied storage backend.
+///
+/// Returns `false` when already absent. No directory sync runs here, so the
caller
+/// must sync the parent directory before treating a successful removal as
durable.
+///
+/// # Errors
+/// Returns [`IggyError::CannotDeleteConsumerOffsetFile`] if unlinking fails.
+pub async fn delete_persisted_offset_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+) -> Result<bool, IggyError> {
// NotFound is tolerated on the result instead of probed for: the probe was
// a blocking stat on the pump before every unlink, and "already gone" is
// exactly the outcome this wants anyway.
- match remove_file(path).await {
+ match storage.remove_file(Path::new(path)).await {
Ok(()) => Ok(true),
Err(error) if error.kind() == std::io::ErrorKind::NotFound =>
Ok(false),
Err(_) =>
Err(IggyError::CannotDeleteConsumerOffsetFile(path.to_owned())),
@@ -677,7 +755,9 @@ mod tests {
.into_owned();
assert_eq!(
- read_purge_generation(&path, 11).await.expect("absent file"),
+ read_purge_generation(&DiskStorage, &path, 11)
+ .await
+ .expect("absent file"),
0,
"absent file is 0"
);
@@ -686,14 +766,18 @@ mod tests {
.await
.expect("persist generation");
assert_eq!(
- read_purge_generation(&path, 11).await.expect("valid file"),
+ read_purge_generation(&DiskStorage, &path, 11)
+ .await
+ .expect("valid file"),
3,
"round-trip"
);
std::fs::write(&path, [0xAB, 0xCD]).expect("write torn file");
assert_eq!(
- read_purge_generation(&path, 11).await.expect("torn file"),
+ read_purge_generation(&DiskStorage, &path, 11)
+ .await
+ .expect("torn file"),
0,
"torn file degrades to 0 so the reconciler re-applies the purge"
);
@@ -701,7 +785,7 @@ mod tests {
// A directory path is a real I/O error, not a short read: it must
// surface, not collapse to the re-purge sentinel (a silent re-purge
// would destroy post-purge messages).
- let result = read_purge_generation(&dir.to_string_lossy(), 11).await;
+ let result = read_purge_generation(&DiskStorage,
&dir.to_string_lossy(), 11).await;
assert!(
matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))),
"real I/O error must propagate, got {result:?}",
@@ -728,12 +812,16 @@ mod tests {
.expect("persist generation");
assert_eq!(
- read_purge_generation(&path, 41).await.expect("same dir"),
+ read_purge_generation(&DiskStorage, &path, 41)
+ .await
+ .expect("same dir"),
9,
"the incarnation that wrote it still hydrates it"
);
assert_eq!(
- read_purge_generation(&path, 42).await.expect("stale file"),
+ read_purge_generation(&DiskStorage, &path, 42)
+ .await
+ .expect("stale file"),
0,
"a record from a dead incarnation must not fence the new one"
);
@@ -742,9 +830,16 @@ mod tests {
persist_purge_generation(&path, 1, 42)
.await
.expect("persist generation");
- assert_eq!(read_purge_generation(&path, 42).await.expect("rekeyed"),
1);
assert_eq!(
- read_purge_generation(&path, 41).await.expect("now stale"),
+ read_purge_generation(&DiskStorage, &path, 42)
+ .await
+ .expect("rekeyed"),
+ 1
+ );
+ assert_eq!(
+ read_purge_generation(&DiskStorage, &path, 41)
+ .await
+ .expect("now stale"),
0
);
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index b6ff9b467..a73da7b44 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -33,7 +33,8 @@ pub const SEMANTIC_VERSION: SemanticVersion =
SemanticVersion::parse_const(VERSI
// Visibility rule: `pub` = named external consumer. main.rs consumes `boot`
// (including `boot::systemd`) and `server_error`; the simulator consumes
// `shell`, `boot::wire_shell_handlers`, and (through `ShellHandlers.sessions`)
-// `session_manager`. Everything else is crate-internal.
+// `session_manager`, plus the storage abstraction for offset recovery
reexported
+// below. Everything else is internal to the crate.
// boot: process entry, shard threads, recovery orchestration.
pub mod boot;
@@ -71,3 +72,5 @@ pub(crate) mod partition_helpers;
pub(crate) mod segment_recovery;
pub mod server_error;
pub(crate) mod sysinfo_probe;
+
+pub use partition_helpers::configure_consumer_offsets_with_storage;
diff --git a/core/server/src/offset_recovery.rs
b/core/server/src/offset_recovery.rs
index 40dca3ff8..25c4ca5d2 100644
--- a/core/server/src/offset_recovery.rs
+++ b/core/server/src/offset_recovery.rs
@@ -26,17 +26,18 @@
//! ways: it reads the first eight bytes and stops, and a file it wrote itself
decodes
//! here as unchecksummed.
+use std::path::Path;
+use std::sync::atomic::AtomicU64;
+
+use futures::StreamExt;
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyError};
+#[cfg(test)]
+use journal::durable_storage::DiskStorage;
+use journal::durable_storage::{DurableFile, DurableStorage, OpenMode};
use partitions::offset_storage::{OffsetRecord, decode_offset_record,
offset_replacement_id};
-use std::path::PathBuf;
-use std::sync::atomic::AtomicU64;
-use tokio::sync::{Semaphore, mpsc};
use tracing::{error, trace, warn};
const COMPONENT: &str = "STREAMING_PARTITIONS";
-const OFFSET_DIRECTORY_BUFFER: usize = 64;
-static OFFSET_DIRECTORY_READERS: Semaphore = Semaphore::const_new(4);
-type OffsetDirectoryEntries = mpsc::Receiver<std::io::Result<Option<PathBuf>>>;
pub struct RecoveredOffsets<T> {
pub entries: Vec<T>,
@@ -58,40 +59,75 @@ impl<T> Default for RecoveredOffsets<T> {
}
}
+#[cfg(test)]
pub async fn load_consumer_offsets(
path: &str,
) -> Result<RecoveredOffsets<ConsumerOffset>, IggyError> {
- let mut recovered = load_offsets(path, ConsumerKind::Consumer, |offset|
offset).await?;
+ load_consumer_offsets_with_storage(&DiskStorage, path).await
+}
+
+/// Recover consumer records, ordered by consumer ID, from a storage backend.
+/// Invalid records are removed when possible. Unreadable records or removals
+/// that cannot be made durable retain their IDs in `stranded_ids`.
+///
+/// # Errors
+/// Returns [`IggyError::CannotReadConsumerOffsets`] if the directory cannot be
+/// enumerated, including when it is missing.
+pub async fn load_consumer_offsets_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+) -> Result<RecoveredOffsets<ConsumerOffset>, IggyError> {
+ let mut recovered =
+ load_offsets(storage, path, ConsumerKind::Consumer, |offset|
offset).await?;
recovered.entries.sort_by_key(|offset| offset.consumer_id);
Ok(recovered)
}
+#[cfg(test)]
pub async fn load_consumer_group_offsets(
path: &str,
) -> Result<RecoveredOffsets<(ConsumerGroupId, ConsumerOffset)>, IggyError> {
- load_offsets(path, ConsumerKind::ConsumerGroup, |offset| {
+ load_consumer_group_offsets_with_storage(&DiskStorage, path).await
+}
+
+/// Recover group records with the same cleanup and stranded file handling as
+/// [`load_consumer_offsets_with_storage`].
+///
+/// # Errors
+/// Returns [`IggyError::CannotReadConsumerOffsets`] if the directory cannot be
+/// enumerated, including when it is missing.
+pub async fn load_consumer_group_offsets_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+) -> Result<RecoveredOffsets<(ConsumerGroupId, ConsumerOffset)>, IggyError> {
+ load_offsets(storage, path, ConsumerKind::ConsumerGroup, |offset| {
(ConsumerGroupId(offset.consumer_id as usize), offset)
})
.await
}
-async fn load_offsets<T>(
+async fn load_offsets<S: DurableStorage, T>(
+ storage: &S,
path: &str,
kind: ConsumerKind,
construct: impl Fn(ConsumerOffset) -> T,
) -> Result<RecoveredOffsets<T>, IggyError> {
trace!(?kind, path, "loading consumer offsets");
- let mut dir_entries = offset_directory_entries(path).await?;
+ let mut dir_entries = storage
+ .regular_files(Path::new(path))
+ .await
+ .map_err(|error| {
+ warn!(?kind, path, %error, "failed to enumerate offset directory");
+ IggyError::CannotReadConsumerOffsets(path.to_owned())
+ })?;
let mut recovered = RecoveredOffsets::default();
- loop {
- let entry_path = match dir_entries.recv().await {
- Some(Ok(Some(path))) => path,
- Some(Ok(None)) => break,
- Some(Err(error)) => {
+ while let Some(entry) = dir_entries.next().await {
+ let entry_path = match entry {
+ Ok(path) => path,
+ Err(error) => {
warn!(?kind, path, %error, "failed to enumerate offset
directory");
return
Err(IggyError::CannotReadConsumerOffsets(path.to_owned()));
}
- None => return
Err(IggyError::CannotReadConsumerOffsets(path.to_owned())),
};
let name = entry_path
.file_name()
@@ -99,7 +135,7 @@ async fn load_offsets<T>(
.to_string_lossy()
.into_owned();
if offset_replacement_id(&name).is_some() {
- remove_stale_replacement(&entry_path, &name).await;
+ remove_stale_replacement(storage, &entry_path, &name).await;
continue;
}
let Ok(consumer_id) = name.parse::<u32>() else {
@@ -113,7 +149,7 @@ async fn load_offsets<T>(
error!(?kind, name, "invalid consumer offset path");
continue;
};
- let offset = match read_offset_file(&path,
offset_kind_label(kind)).await {
+ let offset = match read_offset_file(storage, &path,
offset_kind_label(kind)).await {
OffsetFileLoad::Loaded(offset) => offset,
OffsetFileLoad::Removed => continue,
OffsetFileLoad::Stranded => {
@@ -131,62 +167,11 @@ async fn load_offsets<T>(
Ok(recovered)
}
-async fn offset_directory_entries(path: &str) ->
Result<OffsetDirectoryEntries, IggyError> {
- // Compio has no asynchronous directory iterator and the shard's blocking
- // pool is disabled. Bound both OS threads and buffered paths. The worker
- // owns the permit so cancellation cannot exceed the concurrency bound.
- let permit = OFFSET_DIRECTORY_READERS
- .acquire()
- .await
- .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?;
- let (sender, receiver) = mpsc::channel(OFFSET_DIRECTORY_BUFFER);
- let directory = path.to_owned();
- std::thread::Builder::new()
- .name("iggy-offset-recovery".to_owned())
- .spawn(move || {
- let _permit = permit;
- let result = (|| {
- // Only the directory read itself is fatal. One unreadable
- // entry is skipped with a warning, as the on-reactor loader
- // did, so a single bad dirent cannot keep a partition from
- // booting.
- for entry in std::fs::read_dir(&directory)? {
- let entry = match entry {
- Ok(entry) => entry,
- Err(error) => {
- warn!(path = directory, %error, "failed to read
offset directory entry");
- continue;
- }
- };
- let is_file = match entry.file_type() {
- Ok(file_type) => file_type.is_file(),
- Err(error) => {
- warn!(path = directory, %error, "failed to read
offset entry type");
- continue;
- }
- };
- if is_file &&
sender.blocking_send(Ok(Some(entry.path()))).is_err() {
- return Ok(());
- }
- }
- Ok(())
- })();
- // Explicit completion distinguishes an empty directory from an
- // interrupted worker. Closed receivers simply abandon enumeration.
- let _ = sender.blocking_send(result.map(|()| None));
- })
- .map_err(|error| {
- error!(path, %error, "failed to start offset directory reader");
- IggyError::CannotReadConsumerOffsets(path.to_owned())
- })?;
- Ok(receiver)
-}
-
/// A crashed atomic replacement leaves its sibling behind. The rename never
/// landed, so the sibling is never authoritative. Removal needs no directory
/// sync because a resurrected sibling is still ignored on the next load.
-async fn remove_stale_replacement(path: &std::path::Path, name: &str) {
- match compio::fs::remove_file(path).await {
+async fn remove_stale_replacement<S: DurableStorage>(storage: &S, path: &Path,
name: &str) {
+ match storage.remove_file(path).await {
Ok(()) => trace!("Removed stale offset replacement file: '{name}'."),
Err(e) => warn!(
"{COMPONENT} (error: {e}) - could not remove stale offset
replacement \
@@ -202,8 +187,18 @@ const fn offset_kind_label(kind: ConsumerKind) -> &'static
str {
}
}
-async fn read_offset_file(path: &str, offset_kind: &'static str) ->
OffsetFileLoad {
- let bytes = match compio::fs::read(path).await {
+async fn read_offset_file<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+ offset_kind: &'static str,
+) -> OffsetFileLoad {
+ let bytes = match async {
+ let file = storage.open(Path::new(path), OpenMode::Read).await?;
+ let length =
usize::try_from(file.length().await?).map_err(std::io::Error::other)?;
+ file.read(0, length).await
+ }
+ .await
+ {
Ok(bytes) => bytes,
Err(e) => {
warn!(
@@ -220,7 +215,7 @@ async fn read_offset_file(path: &str, offset_kind: &'static
str) -> OffsetFileLo
"{COMPONENT} - failed to read {offset_kind} from file
(truncated), \
path: {path}, removing invalid file."
);
- remove_invalid_offset_file(path, offset_kind).await
+ remove_invalid_offset_file(storage, path, offset_kind).await
}
// Skipped rather than loaded: resuming from a cursor provably not the
one
// written reads as ordinary redelivery or a gap, never as corruption.
@@ -238,28 +233,27 @@ async fn read_offset_file(path: &str, offset_kind:
&'static str) -> OffsetFileLo
(offset: {offset}, expected: {expected}, found: {found}), \
path: {path}, removing it and resuming this consumer from the
start."
);
- remove_invalid_offset_file(path, offset_kind).await
+ remove_invalid_offset_file(storage, path, offset_kind).await
}
}
}
-async fn remove_invalid_offset_file(path: &str, offset_kind: &'static str) ->
OffsetFileLoad {
- if let Err(error) = compio::fs::remove_file(path).await {
+async fn remove_invalid_offset_file<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+ offset_kind: &'static str,
+) -> OffsetFileLoad {
+ if let Err(error) = storage.remove_file(Path::new(path)).await {
error!(
"{COMPONENT} (error: {error}) - could not remove the invalid \
{offset_kind} file, path: {path}; remove it manually."
);
return OffsetFileLoad::Stranded;
}
- let Some(parent) = std::path::Path::new(path).parent() else {
+ let Some(parent) = Path::new(path).parent() else {
return OffsetFileLoad::Removed;
};
- match async {
- let directory = compio::fs::File::open(parent).await?;
- directory.sync_all().await
- }
- .await
- {
+ match storage.sync_directory(parent).await {
Ok(()) => OffsetFileLoad::Removed,
Err(error) => {
error!(
@@ -285,27 +279,13 @@ mod tests {
));
}
- #[compio::test]
- async fn
given_full_directory_buffer_when_loader_is_cancelled_should_release_worker_capacity()
{
- let dir = tempfile::tempdir().unwrap();
- for id in 0..OFFSET_DIRECTORY_BUFFER * 2 {
- std::fs::write(dir.path().join(id.to_string()),
0_u64.to_le_bytes()).unwrap();
- }
- let path = dir.path().to_str().unwrap();
- for _ in 0..8 {
- let entries = offset_directory_entries(path).await.unwrap();
- drop(entries);
- }
- let loaded = load_consumer_offsets(path).await.unwrap();
- assert_eq!(loaded.entries.len(), OFFSET_DIRECTORY_BUFFER * 2);
- }
-
#[compio::test]
async fn
given_numeric_directory_and_torn_file_when_loading_should_remove_only_invalid_file()
{
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join("7")).unwrap();
std::fs::write(dir.path().join("8"), [1, 2]).unwrap();
std::fs::write(dir.path().join("9"), 12_u64.to_le_bytes()).unwrap();
+ std::fs::write(dir.path().join("10"), []).unwrap();
std::fs::write(dir.path().join("9.tmp"), [0_u8; 4]).unwrap();
std::fs::write(dir.path().join("notes.tmp"), b"unrelated").unwrap();
let path = dir.path().to_str().unwrap();
@@ -316,6 +296,7 @@ mod tests {
assert_eq!(consumers.entries[0].consumer_id, 9);
assert!(consumers.stranded_ids.is_empty());
assert!(!dir.path().join("8").exists());
+ assert!(!dir.path().join("10").exists());
let groups = load_consumer_group_offsets(path).await.unwrap();
assert_eq!(groups.entries.len(), 1);
assert_eq!(groups.entries[0].0, ConsumerGroupId(9));
diff --git a/core/server/src/partition_helpers.rs
b/core/server/src/partition_helpers.rs
index 121df8de7..d7ecfd798 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -30,7 +30,7 @@
//! partition yet.
use crate::offset_recovery::{
- RecoveredOffsets, load_consumer_group_offsets, load_consumer_offsets,
+ RecoveredOffsets, load_consumer_group_offsets_with_storage,
load_consumer_offsets_with_storage,
};
use crate::segment_recovery::{
RecoveredSegment, load_persisted_segments,
load_persisted_segments_with_checkpoint,
@@ -140,9 +140,9 @@ pub async fn create_partition_file_hierarchy(
Ok(())
}
-/// Populate `partition` with consumer-offset / consumer-group-offset storage.
+/// Populate `partition` with consumer and consumer group offset storage from
disk.
///
-/// Hydrates from on-disk state if files exist (recovery path) or
+/// Hydrates from state on disk if files exist (recovery path) or
/// configures empty maps (fresh partition path). Recovered offsets are bounded
/// so a partition that lost its tail does not surface consumer offsets ahead
of
/// an offset it never handed out, and `current_offset` is where a bounded one
@@ -150,15 +150,50 @@ pub async fn create_partition_file_hierarchy(
///
/// # Errors
///
-/// Returns [`ServerError::ConsumerOffsetsLoad`] when the on-disk files
-/// exist but fail to decode. A stored offset past the offset space is clamped
+/// Returns [`ServerError::ConsumerOffsetsLoad`] when an existing offset
+/// directory cannot be enumerated. A stored offset past the offset space is
clamped
/// to `current_offset` (with a warning), not an error.
-#[allow(clippy::too_many_lines)]
pub async fn configure_consumer_offsets(
partition: &mut IggyPartition<Rc<IggyMessageBus>>,
config: &ServerConfig,
namespace: IggyNamespace,
current_offset: u64,
+) -> Result<(), ServerError> {
+ configure_consumer_offsets_with_storage(
+ &DiskStorage,
+ partition,
+ config,
+ namespace,
+ current_offset,
+ )
+ .await
+}
+
+/// Recover consumer and group offsets from `storage` into a new partition.
+///
+/// Restore the partition's message offset and reservation frontier before
+/// calling this, and pass its restored offset counter as `current_offset`.
+/// These values bound which saved consumer positions are plausible.
+///
+/// Missing directories produce empty maps. Valid records seed the visible
+/// offsets and their persistence state. Unreadable records and invalid records
+/// whose removal cannot be made durable retain their admission capacity slots.
+/// Offsets beyond the space reserved by the partition are clamped to
+/// `current_offset`, as during server boot. Clamping changes the visible
position
+/// without rewriting its file; persistence tracking retains the original value
+/// read from storage.
+///
+/// # Errors
+/// Returns [`ServerError::ConsumerOffsetsLoad`] if an existing offset
directory
+/// cannot be enumerated. Consumer recovery may already have seeded the
partition
+/// when group recovery fails, so callers must discard a failed recovery.
+#[allow(clippy::too_many_lines)]
+pub async fn configure_consumer_offsets_with_storage<S: DurableStorage>(
+ storage: &S,
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerConfig,
+ namespace: IggyNamespace,
+ current_offset: u64,
) -> Result<(), ServerError> {
let stream_id = namespace.stream_id();
let topic_id = namespace.topic_id();
@@ -179,6 +214,7 @@ pub async fn configure_consumer_offsets(
let offset_space_ceiling =
current_offset.max(partition.mint_frontier().saturating_sub(1));
let recovered_consumers = load_partition_consumer_offsets(
+ storage,
&consumer_offsets_path,
"consumer",
stream_id,
@@ -221,6 +257,7 @@ pub async fn configure_consumer_offsets(
}
let recovered_groups = load_partition_consumer_group_offsets(
+ storage,
&consumer_group_offsets_path,
stream_id,
topic_id,
@@ -249,7 +286,7 @@ pub async fn configure_consumer_offsets(
let committed_offset = offset.offset.load(Ordering::Relaxed);
partition.seed_recovered_consumer_offset(
ConsumerKind::ConsumerGroup,
- u32::try_from(group_id.0).expect("recovered group id
originated as u32"),
+ offset.consumer_id,
committed_offset,
recovered_offset,
);
@@ -295,35 +332,45 @@ pub async fn configure_consumer_offsets(
Ok(())
}
-async fn load_partition_consumer_offsets(
+async fn load_partition_consumer_offsets<S: DurableStorage>(
+ storage: &S,
path: &str,
consumer_kind: &'static str,
stream_id: usize,
topic_id: usize,
partition_id: usize,
) -> Result<RecoveredOffsets<iggy_common::ConsumerOffset>, ServerError> {
- if !Path::new(path).exists() {
+ if !storage
+ .exists_following_links(Path::new(path))
+ .await
+ .unwrap_or(false)
+ {
return Ok(RecoveredOffsets::default());
}
- load_consumer_offsets(path).await.or_else(|source| {
- if matches!(&source,
IggyError::CannotReadConsumerOffsets(missing_path) if
!Path::new(missing_path).exists())
+ match load_consumer_offsets_with_storage(storage, path).await {
+ Ok(offsets) => Ok(offsets),
+ Err(IggyError::CannotReadConsumerOffsets(_))
+ if !storage
+ .exists_following_links(Path::new(path))
+ .await
+ .unwrap_or(false) =>
{
- return Ok(RecoveredOffsets::default());
+ Ok(RecoveredOffsets::default())
}
-
- Err(ServerError::ConsumerOffsetsLoad {
+ Err(source) => Err(ServerError::ConsumerOffsetsLoad {
consumer_kind,
stream_id,
topic_id,
partition_id,
path: path.to_string(),
source: Box::new(source),
- })
- })
+ }),
+ }
}
-async fn load_partition_consumer_group_offsets(
+async fn load_partition_consumer_group_offsets<S: DurableStorage>(
+ storage: &S,
path: &str,
stream_id: usize,
topic_id: usize,
@@ -332,25 +379,33 @@ async fn load_partition_consumer_group_offsets(
RecoveredOffsets<(iggy_common::ConsumerGroupId,
iggy_common::ConsumerOffset)>,
ServerError,
> {
- if !Path::new(path).exists() {
+ if !storage
+ .exists_following_links(Path::new(path))
+ .await
+ .unwrap_or(false)
+ {
return Ok(RecoveredOffsets::default());
}
- load_consumer_group_offsets(path).await.or_else(|source| {
- if matches!(&source,
IggyError::CannotReadConsumerOffsets(missing_path) if
!Path::new(missing_path).exists())
+ match load_consumer_group_offsets_with_storage(storage, path).await {
+ Ok(offsets) => Ok(offsets),
+ Err(IggyError::CannotReadConsumerOffsets(_))
+ if !storage
+ .exists_following_links(Path::new(path))
+ .await
+ .unwrap_or(false) =>
{
- return Ok(RecoveredOffsets::default());
+ Ok(RecoveredOffsets::default())
}
-
- Err(ServerError::ConsumerOffsetsLoad {
+ Err(source) => Err(ServerError::ConsumerOffsetsLoad {
consumer_kind: "consumer group",
stream_id,
topic_id,
partition_id,
path: path.to_string(),
source: Box::new(source),
- })
- })
+ }),
+ }
}
/// Provision an initial segment + writers for a partition that has none.
diff --git a/core/simulator/src/storage.rs b/core/simulator/src/storage.rs
index e9a47a8d8..441134971 100644
--- a/core/simulator/src/storage.rs
+++ b/core/simulator/src/storage.rs
@@ -277,7 +277,10 @@ impl DurableStorage for SimStorage {
}
async fn open(&self, path: &Path, mode: OpenMode) -> io::Result<SimFile> {
- let creates = matches!(mode, OpenMode::Create |
OpenMode::CreateOrOpen);
+ let creates = matches!(
+ mode,
+ OpenMode::Create | OpenMode::CreateWriteOnly |
OpenMode::CreateOrOpen
+ );
let operation = if creates {
StorageOperation::Create
} else {
@@ -290,7 +293,7 @@ impl DurableStorage for SimStorage {
if let Some(&inode) = state.directory(parent)?.get(&name) {
match &mut state.inodes[inode] {
Inode::File { buffered, .. } => {
- if mode == OpenMode::Create {
+ if matches!(mode, OpenMode::Create |
OpenMode::CreateWriteOnly) {
buffered.clear();
}
}
@@ -669,3 +672,6 @@ fn missing() -> io::Error {
#[cfg(test)]
mod tests;
+
+#[cfg(test)]
+mod purge;
diff --git a/core/simulator/src/storage/purge.rs
b/core/simulator/src/storage/purge.rs
new file mode 100644
index 000000000..adb97dc60
--- /dev/null
+++ b/core/simulator/src/storage/purge.rs
@@ -0,0 +1,469 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Exercise purge completion and consumer recovery across simulated power
loss.
+//!
+//! A consumer offset is a bookmark: storing 2 means `Next` starts at message
3.
+//! Purge must remove those bookmarks as well as the old messages, so consumers
+//! can read a replacement history from offset 0. The purge generation marker
+//! records which purge was applied; it is separate from consumer progress.
+//!
+//! The two controls make that distinction observable. Without a purge,
recovery
+//! must preserve bookmark 2 and return messages 3 and 4. After a completed
purge,
+//! recovery must find no bookmarks and return all five messages, 0 through 4.
+//! Both controls cover individual consumers and groups under both offset
policies.
+//!
+//! The harness enters after message history has been reset. It uses production
+//! purge completion and consumer recovery with `SimStorage`, then polls
through
+//! the real `Next` path. Message recovery is narrower than server boot: a
helper
+//! replays a durable journal into a new partition. No partition memory
survives
+//! recovery. These controls do not inject sync failures or exercise purge
retries.
+
+use super::tests::owned_prepare;
+use super::{Crash, SimStorage};
+use configs::server::ServerConfig;
+use consensus::{LocalPipeline, Sequencer, VsrConsensus};
+use futures::executor::block_on;
+use iggy_common::{
+ ConsumerKind, Durability, IggyByteSize, PartitionStats, PollingStrategy,
TopicRuntimeOptions,
+};
+use journal::durable_storage::{DurableFile, DurableStorage, OpenMode};
+use journal::{DurableAppend, PartitionPrepareJournal};
+use message_bus::IggyMessageBus;
+use partitions::offset_storage::{
+ persist_offset_with_storage, persist_purge_generation_with_storage,
+};
+use partitions::{
+ IggyPartition, IggyPartitions, Partition, PartitionPathLayout,
PartitionsConfig, PollingArgs,
+ PollingConsumer,
+};
+use server::configure_consumer_offsets_with_storage;
+use server_common::send_messages::decode_batch_slice;
+use server_common::sharding::{IggyNamespace, ShardId};
+use std::path::{Path, PathBuf};
+use std::rc::Rc;
+use std::sync::Arc;
+
+const CREATED_REVISION: u64 = 7;
+const OLD_GENERATION: u64 = 4;
+const NEW_GENERATION: u64 = 5;
+const STORED_OFFSET: u64 = 2;
+const CONSUMER_ID: usize = 7;
+const GROUP_ID: usize = 9;
+const STRAY_ID: usize = 99;
+const FRESH_MESSAGE_COUNT: u64 = 5;
+
+type TestPartition = IggyPartition<Rc<IggyMessageBus>>;
+
+/// Own the simulated filesystem and configuration, but no live partition
state.
+///
+/// Offset files and the purge marker use the server's directory layout. The
+/// separate message journal supplies history for rebuilding each new
partition.
+struct PurgeStorageHarness {
+ storage: SimStorage,
+ config: ServerConfig,
+ namespace: IggyNamespace,
+ /// Policy for consumer offsets; message durability is established by the
fixture.
+ policy: Durability,
+}
+
+impl PurgeStorageHarness {
+ /// Store bookmark 2 for a consumer and a group, plus the earlier purge
marker.
+ /// All files and their directory entries are durable before the test
starts,
+ /// including when the selected offset policy does not require an
immediate sync.
+ async fn with_stored_progress(policy: Durability) -> Self {
+ let harness = Self {
+ storage: SimStorage::default(),
+ config: ServerConfig {
+ path: "/purge".to_owned(),
+ ..ServerConfig::default()
+ },
+ namespace: IggyNamespace::new(0, 0, 42),
+ policy,
+ };
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ let directory = harness.offset_directory(kind);
+ harness
+ .storage
+ .create_directories(&directory)
+ .await
+ .unwrap();
+ for ancestor in directory.ancestors() {
+ harness.storage.sync_directory(ancestor).await.unwrap();
+ }
+ }
+ harness
+ .persist_bookmark(ConsumerKind::Consumer, CONSUMER_ID)
+ .await;
+ harness
+ .persist_bookmark(ConsumerKind::ConsumerGroup, GROUP_ID)
+ .await;
+ persist_purge_generation_with_storage(
+ &harness.storage,
+ &format!("{}/purge.gen", harness.partition_directory()),
+ OLD_GENERATION,
+ CREATED_REVISION,
+ )
+ .await
+ .unwrap();
+ harness
+ }
+
+ /// Persist bookmark 2 without adding it to any partition's live offset
maps.
+ /// This also creates stray files that the purge must discover by scanning.
+ async fn persist_bookmark(&self, kind: ConsumerKind, consumer_id: usize) {
+ let directory = self.offset_directory(kind);
+ let path = directory.join(consumer_id.to_string());
+ persist_offset_with_storage(
+ &self.storage,
+ path.to_str().unwrap(),
+ STORED_OFFSET,
+ self.policy.is_persisted(),
+ )
+ .await
+ .unwrap();
+ // Establish the same durable input under both policies. The replicated
+ // policy does not itself require the cursor write to sync immediately.
+ self.storage
+ .open(&path, OpenMode::Read)
+ .await
+ .unwrap()
+ .sync()
+ .await
+ .unwrap();
+ self.storage.sync_directory(&directory).await.unwrap();
+ }
+
+ /// Journal five messages at offsets 0 through 4 without touching offset
directories.
+ /// Syncing the message journal must not accidentally make bookmark
cleanup durable.
+ async fn persist_fresh_history(&self) {
+ let mut journal = PartitionPrepareJournal::open_with_storage(
+ &self.journal_directory(),
+ self.namespace.inner(),
+ CREATED_REVISION,
+ self.storage.clone(),
+ )
+ .await
+ .unwrap();
+ let mut parent_checksum = 0;
+ for offset in 0..FRESH_MESSAGE_COUNT {
+ let operation_number = offset + 1;
+ // The helper fills the payload with the operation number, allowing
+ // polls to verify message contents as well as their offsets.
+ let prepare = owned_prepare(operation_number, parent_checksum,
offset);
+ parent_checksum = prepare.header().checksum;
+ journal.append(prepare.into_frozen()).await.unwrap();
+ }
+ journal.sync().await.unwrap();
+ // Never write back the whole simulated filesystem here: that would
also
+ // make an offset deletion durable after its own directory sync failed.
+ self.storage
+ .sync_directory(Path::new(&self.partition_directory()))
+ .await
+ .unwrap();
+ }
+
+ /// Rebuild messages, the applied purge marker, and consumer progress from
storage.
+ ///
+ /// Journal entries are appended and committed into a fresh in-memory
partition
+ /// before the shared server offset loader runs. This supplies real
readable
+ /// messages for `Next` without claiming to exercise the full server boot
path.
+ async fn recover_partition(&self) -> TestPartition {
+ let journal = PartitionPrepareJournal::open_with_storage(
+ &self.journal_directory(),
+ self.namespace.inner(),
+ CREATED_REVISION,
+ self.storage.clone(),
+ )
+ .await
+ .unwrap();
+ let mut partition = self.empty_partition();
+ for prepare in journal.prepares().await.unwrap() {
+ let operation_number = prepare.header().op;
+ partition.append_messages(prepare).await.unwrap();
+ partition
+ .consensus()
+ .sequencer()
+ .set_sequence(operation_number);
+ partition.consensus().advance_commit_max(operation_number);
+ partition.commit_journal(&partition_config()).await;
+ assert!(partition.fatal().is_none());
+ }
+ let current_offset = partition.offsets().commit_offset;
+ assert_eq!(current_offset, FRESH_MESSAGE_COUNT - 1);
+ self.recover_progress(&mut partition, current_offset).await;
+ partition
+ }
+
+ /// Use the shared marker and offset loaders, including the server's
offset clamping.
+ /// `current_offset` is the message bound against which saved progress is
checked.
+ async fn recover_progress(&self, partition: &mut TestPartition,
current_offset: u64) {
+ partition
+ .hydrate_applied_purge_generation_with_storage(&self.storage)
+ .await
+ .unwrap();
+ configure_consumer_offsets_with_storage(
+ &self.storage,
+ partition,
+ &self.config,
+ self.namespace,
+ current_offset,
+ )
+ .await
+ .unwrap();
+ }
+
+ /// Poll `Next` for the consumer and group, checking actual offsets and
payloads.
+ /// This executes and completes real polls with automatic offset commits
disabled;
+ /// it consumes the partition because completing polls can update live
tracking.
+ async fn poll_next_and_assert_messages(
+ &self,
+ partition: TestPartition,
+ expected_offsets: &[u64],
+ ) {
+ let partitions = IggyPartitions::new(ShardId::new(0),
partition_config());
+ partitions.insert(self.namespace, partition);
+ for consumer in consumers() {
+ let plan = partitions
+ .build_poll_snapshot(
+ &self.namespace,
+ consumer,
+ &PollingArgs {
+ strategy: PollingStrategy::next(),
+ count: 10,
+ auto_commit: false,
+ },
+ )
+ .unwrap();
+ assert!(!plan.needs_off_pump_io());
+ let completion = partitions
+ .complete_poll(&self.namespace, plan.execute().await)
+ .unwrap();
+ assert!(completion.replication.is_none());
+ let actual_offsets: Vec<_> = completion
+ .fragments
+ .iter()
+ .map(|fragment| {
+ let batch =
decode_batch_slice(fragment.as_slice()).unwrap();
+ assert_eq!(batch.message_count(), 1);
+ let message = batch.iter().next().unwrap();
+ let expected_byte = u8::try_from(batch.header.base_offset
+ 1).unwrap();
+ assert!(message.payload.iter().all(|byte| *byte ==
expected_byte));
+ batch.header.base_offset
+ })
+ .collect();
+ assert_eq!(
+ actual_offsets, expected_offsets,
+ "{:?}, {consumer:?}",
+ self.policy
+ );
+ }
+ }
+
+ /// Create a partition with no recovered messages or consumer progress.
+ /// Its identity matches the durable records so recovery can accept those
records.
+ fn empty_partition(&self) -> TestPartition {
+ let consensus = VsrConsensus::new(
+ 1,
+ 0,
+ 1,
+ self.namespace.inner(),
+ Rc::new(IggyMessageBus::new(0)),
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ let mut partition = IggyPartition::with_in_memory_storage(
+ Arc::new(PartitionStats::default()),
+ consensus,
+ IggyByteSize::from(1024 * 1024),
+ );
+ partition.set_partition_dir(self.partition_directory());
+ partition.set_created_revision(CREATED_REVISION);
+ partition.set_runtime_options(TopicRuntimeOptions {
+ durability: Durability::Replicated,
+ consumer_offset_durability: self.policy,
+ ..TopicRuntimeOptions::default()
+ });
+ partition
+ }
+
+ fn partition_directory(&self) -> String {
+ self.config.get_partition_path(0, 0, 42)
+ }
+
+ fn journal_directory(&self) -> PathBuf {
+ Path::new(&self.partition_directory()).join("fresh-history")
+ }
+
+ fn offset_directory(&self, kind: ConsumerKind) -> PathBuf {
+ match kind {
+ ConsumerKind::Consumer => self.config.get_consumer_offsets_path(0,
0, 42).into(),
+ ConsumerKind::ConsumerGroup => {
+ self.config.get_consumer_group_offsets_path(0, 0, 42).into()
+ }
+ }
+ }
+}
+
+fn consumers() -> [PollingConsumer; 2] {
+ [
+ PollingConsumer::Consumer(CONSUMER_ID, 42),
+ PollingConsumer::ConsumerGroup(GROUP_ID, 1),
+ ]
+}
+
+fn partition_config() -> PartitionsConfig {
+ PartitionsConfig {
+ messages_required_to_save: 100,
+ size_of_messages_required_to_save: IggyByteSize::from(1024 * 1024),
+ validate_checksum: true,
+ segment_size: IggyByteSize::from(1024 * 1024),
+ preallocate_segments: false,
+ encryptor: None,
+ path_layout: PartitionPathLayout::default(),
+ }
+}
+
+/// After power loss, a new partition must load the consumer and group
+/// bookmarks from storage. Both saved bookmarks are 2, so Next must
+/// return messages 3 and 4.
+#[test]
+fn
given_stored_progress_when_power_is_lost_should_recover_both_consumer_bookmarks()
{
+ block_on(async {
+ for policy in [Durability::Replicated, Durability::Persisted] {
+ let harness =
PurgeStorageHarness::with_stored_progress(policy).await;
+ harness.persist_fresh_history().await;
+
+ harness.storage.crash(Crash::PowerLoss);
+ let recovered = harness.recover_partition().await;
+
+ assert_eq!(recovered.applied_purge_generation(), OLD_GENERATION);
+ for consumer in consumers() {
+ assert_eq!(recovered.get_consumer_offset(consumer),
Some(STORED_OFFSET));
+ }
+ // Bookmark 2 means the first three messages were already consumed.
+ harness
+ .poll_next_and_assert_messages(recovered, &[3, 4])
+ .await;
+ }
+ });
+}
+
+/// Purge must remove the consumer and group bookmarks before fresh messages
arrive.
+/// After power loss, a new partition must find no saved bookmarks, so Next
returns
+/// all fresh messages, 0 through 4. Restoring either old bookmark of 2 would
skip
+/// messages 0 through 2 even though that bookmark is still within the new
history.
+#[test]
+fn given_completed_purge_when_power_is_lost_should_read_all_fresh_messages() {
+ block_on(async {
+ for policy in [Durability::Replicated, Durability::Persisted] {
+ // Start at the completion phase: message history has been reset,
+ // but the old consumer and group bookmarks still need to be
cleared.
+ let harness =
PurgeStorageHarness::with_stored_progress(policy).await;
+ let mut partition = harness.empty_partition();
+ harness
+ .recover_progress(&mut partition, STORED_OFFSET)
+ .await;
+ assert_eq!(
+ partition.applied_purge_generation(),
+ OLD_GENERATION,
+ "{policy:?}: setup must load the earlier purge marker"
+ );
+ for consumer in consumers() {
+ assert_eq!(
+ partition.get_consumer_offset(consumer),
+ Some(STORED_OFFSET),
+ "{policy:?}, {consumer:?}: setup must load the old
bookmark"
+ );
+ }
+
+ // These files arrive after recovery, so only the production
directory
+ // sweep can discover them. Both directories must be cleaned
durably.
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ harness.persist_bookmark(kind, STRAY_ID).await;
+ }
+
+ // Check the purge's effects on the live partition before
discarding it.
+ partition
+ .complete_purge_with_storage(&harness.storage, NEW_GENERATION)
+ .await
+ .expect("complete purge cleanup");
+ assert_eq!(
+ partition.applied_purge_generation(),
+ NEW_GENERATION,
+ "{policy:?}: purge must advance the applied generation"
+ );
+ for consumer in consumers() {
+ assert_eq!(
+ partition.get_consumer_offset(consumer),
+ None,
+ "{policy:?}, {consumer:?}: purge must clear the live
bookmark"
+ );
+ }
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ assert_eq!(
+ partition.durable_consumer_offset_count(kind),
+ 0,
+ "{policy:?}, {kind:?}: purge must clear durability
tracking"
+ );
+ let directory = harness.offset_directory(kind);
+ let remaining_paths: Vec<_> = harness
+ .storage
+ .entries(&directory)
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|entry| directory.join(entry.name))
+ .collect();
+ assert!(
+ remaining_paths.is_empty(),
+ "{policy:?}, {kind:?}: purge left bookmark files:
{remaining_paths:?}"
+ );
+ }
+ drop(partition);
+
+ // Save messages 0 through 4 after purge, then simulate power loss.
+ // The new partition must load its messages and bookmarks from
storage.
+ harness.persist_fresh_history().await;
+ harness.storage.crash(Crash::PowerLoss);
+ let recovered = harness.recover_partition().await;
+
+ assert_eq!(
+ recovered.applied_purge_generation(),
+ NEW_GENERATION,
+ "{policy:?}: the completed purge marker must survive power
loss"
+ );
+ for consumer in consumers() {
+ assert_eq!(
+ recovered.get_consumer_offset(consumer),
+ None,
+ "{policy:?}, {consumer:?}: a deleted bookmark must not
return after power loss"
+ );
+ }
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ assert_eq!(
+ recovered.durable_consumer_offset_count(kind),
+ 0,
+ "{policy:?}, {kind:?}: recovery must find no durable
bookmarks"
+ );
+ }
+ harness
+ .poll_next_and_assert_messages(recovered, &[0, 1, 2, 3, 4])
+ .await;
+ }
+ });
+}
diff --git a/core/simulator/src/storage/tests.rs
b/core/simulator/src/storage/tests.rs
index 7e0eade54..13947f668 100644
--- a/core/simulator/src/storage/tests.rs
+++ b/core/simulator/src/storage/tests.rs
@@ -2235,7 +2235,7 @@ async fn assert_owned_segments(
);
}
-fn owned_prepare(op: u64, parent: u128, offset: u64) -> Message<PrepareHeader>
{
+pub(super) fn owned_prepare(op: u64, parent: u128, offset: u64) ->
Message<PrepareHeader> {
let payload = vec