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 2cacea5b1 fix(partitions): sync checkpoints through original writers
(#4253)
2cacea5b1 is described below
commit 2cacea5b1dd6d7e9a9b72c9a120f648af2553be3
Author: Gunther Xing <[email protected]>
AuthorDate: Tue Sep 22 17:14:49 2026 +0800
fix(partitions): sync checkpoints through original writers (#4253)
---
core/partitions/Cargo.toml | 9 ++--
core/partitions/src/iggy_index_writer.rs | 4 ++
core/partitions/src/iggy_partition.rs | 28 +++++-----
core/partitions/src/lib.rs | 2 +
core/partitions/src/persistence.rs | 49 ++++++++++++++++-
core/simulator/src/storage/tests.rs | 93 ++++++++++++++++++++++++++++++--
6 files changed, 162 insertions(+), 23 deletions(-)
diff --git a/core/partitions/Cargo.toml b/core/partitions/Cargo.toml
index 8b1efd778..d14692c08 100644
--- a/core/partitions/Cargo.toml
+++ b/core/partitions/Cargo.toml
@@ -36,11 +36,10 @@ publish = false
# on the read path.
poll-diagnostics = []
-# Simulator-only detector hook (`IggyPartitions::hold_borrow_across_await`):
-# deliberately holds a `with_partition` borrow across an `.await` so the
-# dispatch shell can prove its borrow-across-await detector. A
-# `-p iggy-server` build excludes it; `cargo build --workspace` unifies
-# features so the shared `partitions` unit compiles it in when the simulator
+# Simulator-only hooks: the dispatch shell's borrow-across-await detector and
+# original-writer checkpoint barriers for deterministic storage tests. A
+# `-p iggy-server` build excludes them; `cargo build --workspace` unifies
+# features so the shared `partitions` unit compiles them in when the simulator
# requests it. No production caller.
simulator = []
diff --git a/core/partitions/src/iggy_index_writer.rs
b/core/partitions/src/iggy_index_writer.rs
index ddd6f294d..c91cea223 100644
--- a/core/partitions/src/iggy_index_writer.rs
+++ b/core/partitions/src/iggy_index_writer.rs
@@ -144,6 +144,10 @@ impl IggyIndexWriter {
self.index_size_bytes.fetch_add(bytes, Ordering::Release);
}
+ pub(crate) fn path(&self) -> &str {
+ &self.file_path
+ }
+
/// Flushes buffered index file contents to disk.
///
/// Uses `fdatasync` (data only): index files are append-only and the
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index d3ae0e6c7..2c9d77d80 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -28,7 +28,9 @@ use crate::offset_storage::{
persist_offset, persist_offset_max, persist_purge_generation_with_storage,
read_purge_generation,
};
-use crate::persistence::{PartitionPersistence, PersistenceCompletion,
PersistenceNotifier};
+use crate::persistence::{
+ CheckpointBarrier, PartitionPersistence, PersistenceCompletion,
PersistenceNotifier,
+};
use crate::poll_plan::{
DiskReadPlan, DiskSegment, PartitionDirResolution, PollContext, PollPlan,
PollReadResult,
PollTier, ResidentTailSnapshot,
@@ -1033,19 +1035,21 @@ where
return;
}
}
- if let Some(writer) =
self.log.index_writers().last().and_then(Option::as_ref)
- && let Err(error) = writer.fsync().await
- {
- error!(%error, namespace_raw = self.namespace().inner(),
"partition checkpoint index sync failed");
- self.fatal = Some(FatalCommit {
- namespace_raw: self.namespace().inner(),
- op: through_op,
- operation: Operation::SendMessages,
- });
- return;
+ let mut barriers = Vec::new();
+ if let Some(writer) =
self.log.index_writers().last().and_then(Option::as_ref) {
+ if let Err(error) = writer.fsync().await {
+ error!(%error, namespace_raw = self.namespace().inner(),
"partition checkpoint index sync failed");
+ self.fatal = Some(FatalCommit {
+ namespace_raw: self.namespace().inner(),
+ op: through_op,
+ operation: Operation::SendMessages,
+ });
+ return;
+ }
+ barriers.push(CheckpointBarrier::already_synced(writer.path()));
}
let (files, directories) = self.persistence_checkpoint_files(config);
- persistence.checkpoint_files(through_op, files, directories);
+ persistence.checkpoint_files(through_op, files, directories, barriers);
self.start_persistence();
}
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 200892f8b..7556ea67e 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -30,6 +30,8 @@ mod messages_writer;
pub mod offset_storage;
mod persistence;
mod poll_plan;
+#[cfg(feature = "simulator")]
+pub use persistence::CheckpointBarrier;
pub use persistence::{
PartitionPersistence, PersistenceCompletion, PersistenceMetrics,
PersistenceNotifier,
};
diff --git a/core/partitions/src/persistence.rs
b/core/partitions/src/persistence.rs
index 19d77d564..d543fe981 100644
--- a/core/partitions/src/persistence.rs
+++ b/core/partitions/src/persistence.rs
@@ -25,8 +25,10 @@ use server_common::iobuf::Frozen;
use smallvec::SmallVec;
use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
+use std::future::Future;
use std::io;
use std::path::{Path, PathBuf};
+use std::pin::Pin;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, LazyLock, Mutex, Weak};
@@ -127,6 +129,40 @@ pub struct PersistenceMetrics {
pub type PersistenceNotifier = Rc<dyn Fn(PersistenceCompletion)>;
+/// A file durability barrier that keeps the original writer alive until sync.
+#[must_use]
+pub struct CheckpointBarrier {
+ path: PathBuf,
+ sync: Pin<Box<dyn Future<Output = io::Result<()>> + 'static>>,
+}
+
+impl CheckpointBarrier {
+ /// Retain a simulator file's original writer until checkpoint
synchronizes it.
+ #[cfg(feature = "simulator")]
+ pub fn from_file<F: DurableFile + 'static>(path: impl Into<PathBuf>, file:
F) -> Self {
+ Self::from_future(path, async move { file.sync().await })
+ }
+
+ fn from_future(
+ path: impl Into<PathBuf>,
+ sync: impl Future<Output = io::Result<()>> + 'static,
+ ) -> Self {
+ Self {
+ path: path.into(),
+ sync: Box::pin(sync),
+ }
+ }
+
+ pub(crate) fn already_synced(path: impl Into<PathBuf>) -> Self {
+ Self::from_future(path, async { Ok(()) })
+ }
+
+ async fn run(self) -> io::Result<PathBuf> {
+ self.sync.await?;
+ Ok(self.path)
+ }
+}
+
pub struct PartitionPersistence<S: DurableStorage = DiskStorage> {
group: u64,
instance: u64,
@@ -429,6 +465,7 @@ enum Mutation<S: DurableStorage> {
through_op: u64,
files: Vec<PathBuf>,
directories: Vec<PathBuf>,
+ barriers: Vec<CheckpointBarrier>,
offset_files: Vec<RetainedOffsetFile<S::File>>,
synced_files: BTreeSet<PathBuf>,
},
@@ -832,7 +869,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
}
pub fn checkpoint(&self, through_op: u64) {
- self.checkpoint_files(through_op, Vec::new(), Vec::new());
+ self.checkpoint_files(through_op, Vec::new(), Vec::new(), Vec::new());
}
pub fn checkpoint_files(
@@ -840,6 +877,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
through_op: u64,
files: Vec<PathBuf>,
directories: Vec<PathBuf>,
+ barriers: Vec<CheckpointBarrier>,
) {
if through_op <= self.checkpoint_requested.get() {
return;
@@ -859,6 +897,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
through_op,
files,
directories,
+ barriers,
offset_files,
synced_files,
});
@@ -1264,8 +1303,9 @@ impl<S: DurableStorage> PartitionPersistence<S> {
through_op,
files,
directories,
+ barriers,
offset_files,
- synced_files,
+ mut synced_files,
..
} => {
self.checkpoint_running.set(true);
@@ -1273,6 +1313,11 @@ impl<S: DurableStorage> PartitionPersistence<S> {
futures::stream::iter(offset_files.iter().map(Ok::<_,
io::Error>))
.try_for_each_concurrent(16, |retained|
retained.file.sync())
.await?;
+ let barrier_paths = futures::future::try_join_all(
+ barriers.into_iter().map(CheckpointBarrier::run),
+ )
+ .await?;
+ synced_files.extend(barrier_paths);
journal
.checkpoint_files(through_op, &files, &directories,
&synced_files)
.await
diff --git a/core/simulator/src/storage/tests.rs
b/core/simulator/src/storage/tests.rs
index 13947f668..636486619 100644
--- a/core/simulator/src/storage/tests.rs
+++ b/core/simulator/src/storage/tests.rs
@@ -27,7 +27,7 @@ use journal::partition_journal::{
PARTITION_WAL_BLOCK_SIZE, SegmentPosition, SegmentReference, record_length,
};
use journal::{DurableAppend, PartitionPrepareJournal};
-use partitions::{PartitionPersistence, PersistenceMetrics, install_backup};
+use partitions::{CheckpointBarrier, PartitionPersistence, PersistenceMetrics,
install_backup};
use server_common::send_messages::{
BATCH_MESSAGE_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages,
SendMessagesOwned,
};
@@ -652,6 +652,7 @@ fn
replacing_a_retained_offset_writer_keeps_both_inodes_until_checkpoint() {
1,
vec![path.to_path_buf()],
vec![Path::new(DIRECTORY).to_path_buf()],
+ Vec::new(),
);
assert!(persistence.start());
Rc::clone(&persistence).run().await;
@@ -731,6 +732,7 @@ fn
checkpoint_skips_duplicate_offset_sync_but_still_refuses_a_missing_path() {
1,
vec![path.to_path_buf()],
vec![Path::new(DIRECTORY).to_path_buf()],
+ Vec::new(),
);
storage.clear_trace();
assert!(persistence.start());
@@ -797,6 +799,39 @@ async fn interrupted_install() -> SimStorage {
storage
}
+/// A hard link preserves the inode, not the writer's error cursor. Opening the
+/// backup name after writeback failed must not authorize destructive install.
+#[test]
+#[ignore = "`install_backup::link_tree` synchronizes hard links through
handles opened after the writeback failure"]
+fn
given_a_failed_writeback_when_beginning_an_install_backup_should_refuse_publication()
{
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let path = Path::new("/partition/materialized");
+ let mut writer = storage.open(path, OpenMode::Create).await.unwrap();
+ writer.write(0, b"pending".to_vec()).await.unwrap();
+ storage.sync_directory(Path::new(DIRECTORY)).await.unwrap();
+
+ storage.fail_writeback(path).unwrap();
+ let result = install_backup::begin_with_storage(Path::new(DIRECTORY),
&storage).await;
+
+ assert!(
+ writer.sync().await.is_err(),
+ "the original writer did not observe the injected writeback
failure"
+ );
+ assert!(
+ result.is_err(),
+ "install backup published after synchronizing a fresh hard-link
handle past the writeback error"
+ );
+ assert!(
+ !storage
+ .exists(Path::new("/partition/.install-backup"))
+ .await
+ .unwrap(),
+ "a failed backup was published"
+ );
+ });
+}
+
#[test]
fn lost_frontier_cannot_turn_a_durable_journal_into_an_empty_one() {
block_on(async {
@@ -1114,7 +1149,12 @@ fn
checkpoint_syncs_the_retained_writer_before_reclaiming_its_history() {
);
storage.remove_file(path).await.unwrap();
persistence.retire_offset_file(path.to_str().unwrap());
- persistence.checkpoint_files(4, Vec::new(),
vec![Path::new(DIRECTORY).to_path_buf()]);
+ persistence.checkpoint_files(
+ 4,
+ Vec::new(),
+ vec![Path::new(DIRECTORY).to_path_buf()],
+ Vec::new(),
+ );
storage.fail_at(0, FaultMode::Before);
assert!(persistence.start());
Rc::clone(&persistence).run().await;
@@ -1142,10 +1182,12 @@ fn
checkpoint_barriers_complete_before_wal_reclamation() {
.unwrap();
let mut file = storage.open(path, OpenMode::Create).await.unwrap();
file.write(0, b"committed".to_vec()).await.unwrap();
+ let barrier = CheckpointBarrier::from_file(path, file);
persistence.checkpoint_files(
4,
vec![path.to_path_buf()],
vec![Path::new(DIRECTORY).to_path_buf()],
+ vec![barrier],
);
assert!(persistence.checkpoint_pending());
assert!(!persistence.needs_checkpoint());
@@ -1185,7 +1227,7 @@ fn
failed_materialization_keeps_wal_coverage_and_fences_completion() {
} else {
(missing, Vec::new())
};
- persistence.checkpoint_files(4, files, directories);
+ persistence.checkpoint_files(4, files, directories, Vec::new());
assert!(persistence.start());
Rc::clone(&persistence).run().await;
assert_eq!(
@@ -2758,7 +2800,6 @@ fn
given_a_silent_short_write_when_recovering_then_the_record_should_be_refused(
}
#[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;
@@ -2767,6 +2808,7 @@ fn
given_a_failed_writeback_when_checkpointing_then_wal_history_should_not_be_re
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();
+ let barrier = CheckpointBarrier::from_file(path, writer);
// The device drops the dirty pages before the checkpoint's barrier.
The
// writer that issued them is the only handle told; the descriptor
@@ -2777,6 +2819,7 @@ fn
given_a_failed_writeback_when_checkpointing_then_wal_history_should_not_be_re
4,
vec![path.to_path_buf()],
vec![Path::new(DIRECTORY).to_path_buf()],
+ vec![barrier],
);
assert!(persistence.start());
Rc::clone(&persistence).run().await;
@@ -2800,6 +2843,48 @@ fn
given_a_failed_writeback_when_checkpointing_then_wal_history_should_not_be_re
});
}
+#[test]
+fn
given_multiple_writers_for_one_checkpoint_file_when_one_has_not_observed_the_writeback_failure_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 first_writer = storage.open(path,
OpenMode::Create).await.unwrap();
+ first_writer.write(0, b"first".to_vec()).await.unwrap();
+ let mut second_writer = storage.open(path,
OpenMode::ReadWrite).await.unwrap();
+ second_writer.write(5, b"second".to_vec()).await.unwrap();
+
+ storage.fail_writeback(path).unwrap();
+ // Consuming the inode error through one file description must not let
+ // checkpoint skip another writer that still has the error pending.
+ assert!(first_writer.sync().await.is_err());
+ let barriers = vec![
+ CheckpointBarrier::from_file(path, first_writer),
+ CheckpointBarrier::from_file(path, second_writer),
+ ];
+ persistence.checkpoint_files(
+ 4,
+ vec![path.to_path_buf()],
+ vec![Path::new(DIRECTORY).to_path_buf()],
+ barriers,
+ );
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+
+ assert!(persistence.failure().is_some());
+ 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);
+ assert_eq!(recovered.head(), 4);
+ });
+}
+
#[test]
fn
given_sim_storage_when_opening_persistence_then_the_writer_lease_should_be_taken()
{
block_on(async {
