This is an automated email from the ASF dual-hosted git repository.
slbotbm 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 1e1278299 fix(partitions): fence install backup on writeback failure
(#4278)
1e1278299 is described below
commit 1e1278299bfc26bbfa7ed826a4096bc7bbcdd74b
Author: Gunther Xing <[email protected]>
AuthorDate: Wed Oct 14 13:21:12 2026 +0800
fix(partitions): fence install backup on writeback failure (#4278)
---
core/partitions/Cargo.toml | 2 +-
core/partitions/src/iggy_index_writer.rs | 13 +-
core/partitions/src/iggy_partition.rs | 141 ++++++++++++++++++++-
core/partitions/src/install_backup.rs | 84 ++++++++++---
core/partitions/src/io.rs | 9 +-
core/partitions/src/lib.rs | 6 +-
core/partitions/src/messages_writer.rs | 13 +-
core/partitions/src/persistence.rs | 67 +++++++---
core/partitions/src/state_transfer.rs | 25 ++--
core/simulator/src/storage/tests.rs | 209 ++++++++++++++++++++++++++++---
10 files changed, 499 insertions(+), 70 deletions(-)
diff --git a/core/partitions/Cargo.toml b/core/partitions/Cargo.toml
index bb4b42056..35ff7f38e 100644
--- a/core/partitions/Cargo.toml
+++ b/core/partitions/Cargo.toml
@@ -37,7 +37,7 @@ publish = false
poll-diagnostics = []
# Simulator-only hooks: the dispatch shell's borrow-across-await detector and
-# original-writer checkpoint barriers for deterministic storage tests. A
+# original-writer file sync 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.
diff --git a/core/partitions/src/iggy_index_writer.rs
b/core/partitions/src/iggy_index_writer.rs
index 7a86dc064..337238cb6 100644
--- a/core/partitions/src/iggy_index_writer.rs
+++ b/core/partitions/src/iggy_index_writer.rs
@@ -171,10 +171,15 @@ impl IggyIndexWriter {
///
/// Returns an error if the file cannot be synchronized.
pub async fn fsync(&self) -> Result<(), IggyError> {
- self.file
- .sync_data()
- .await
- .map_err(|_| IggyError::CannotWriteToFile)?;
+ self.file.sync_data().await.map_err(|err| {
+ error!(
+ target: "iggy.partitions.storage",
+ file = self.file_path.as_str(),
+ %err,
+ "failed to sync index file"
+ );
+ IggyError::CannotWriteToFile
+ })?;
Ok(())
}
}
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index dd85ade26..5a7785788 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -24,7 +24,7 @@ use crate::journal::{
use crate::log::JournalInfo;
use crate::log::SegmentedLog;
use crate::persistence::{
- CheckpointBarrier, PartitionPersistence, PersistenceCompletion,
PersistenceNotifier,
+ FileSyncBarrier, PartitionPersistence, PersistenceCompletion,
PersistenceNotifier,
};
use crate::poll_plan::{
DiskReadPlan, DiskSegment, PartitionDirResolution, PollContext, PollPlan,
PollReadResult,
@@ -91,10 +91,11 @@ use server_common::{
sharding::IggyNamespace,
};
use std::cell::{Cell, RefCell};
-use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
use std::fmt;
use std::hash::Hash;
use std::num::{NonZeroU32, NonZeroU64};
+use std::path::PathBuf;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
@@ -1760,7 +1761,10 @@ where
});
return;
}
- barriers.push(CheckpointBarrier::already_synced(writer.path()));
+ // [`IggyIndexWriter::fsync`] used the original descriptor above.
+ // The marker tells [`PartitionPersistence::checkpoint_files`] not
+ // to replace that proof with a fresh-handle sync.
+ barriers.push(FileSyncBarrier::already_synced(writer.path()));
}
if let Err(error) = self.persist_retry_checkpoint(through_op).await {
error!(%error, namespace_raw = self.namespace().inner(),
"partition receipt checkpoint failed");
@@ -1778,6 +1782,30 @@ where
self.start_persistence();
}
+ /// Queue original-writer syncs before the install backup is published.
+ /// This runs even when the checkpoint frontier cannot advance.
+ pub(crate) fn barrier_install_files_locked(&self) -> BTreeSet<PathBuf> {
+ let Some(persistence) = &self.persistence else {
+ return BTreeSet::new();
+ };
+ let mut barriers = Vec::with_capacity(2);
+ if let Some(writer) =
self.log.messages_writers().last().and_then(Option::as_ref) {
+ let path = writer.path();
+ let writer = Rc::clone(writer);
+ barriers.push(FileSyncBarrier::from_future(path, async move {
+ writer.fsync().await.map_err(std::io::Error::other)
+ }));
+ }
+ if let Some(writer) =
self.log.index_writers().last().and_then(Option::as_ref) {
+ let path = writer.path().to_owned();
+ let writer = Rc::clone(writer);
+ barriers.push(FileSyncBarrier::from_future(path, async move {
+ writer.fsync().await.map_err(std::io::Error::other)
+ }));
+ }
+ persistence.barrier_files(barriers)
+ }
+
fn persistence_checkpoint_files(
&self,
config: &PartitionsConfig,
@@ -1892,7 +1920,7 @@ where
PendingCheckpoint::Enqueue(through, ref synced_index) => {
let barriers = synced_index
.iter()
- .map(CheckpointBarrier::already_synced)
+ .map(FileSyncBarrier::already_synced)
.collect();
let (files, directories) =
self.persistence_checkpoint_files(config);
if let Some(persistence) = &self.persistence {
@@ -12968,6 +12996,77 @@ mod tests {
}
}
+ #[cfg(target_os = "linux")]
+ #[compio::test]
+ async fn
given_an_index_sync_failure_when_installing_state_transfer_should_refuse_backup_publication()
+ {
+ let directory = tempfile::tempdir().unwrap();
+ let (mut partition, _) = recording_partition_at(0, 3);
+ let partition_dir = directory.path().to_string_lossy().into_owned();
+ partition.set_partition_dir(partition_dir.clone());
+ partition.runtime_options.durability =
iggy_common::Durability::Persisted;
+ for kind in ["consumers", "groups", "external_groups"] {
+
std::fs::create_dir_all(directory.path().join("offsets").join(kind)).unwrap();
+ }
+ partition.consumer_offset_dirs[ConsumerKind::Consumer.index()] =
+ Some(format!("{partition_dir}/offsets/consumers"));
+ partition.consumer_offset_dirs[ConsumerKind::ConsumerGroup.index()] =
+ Some(format!("{partition_dir}/offsets/groups"));
+ partition.consumer_offset_dirs[ConsumerKind::ExternalGroup.index()] =
+ Some(format!("{partition_dir}/offsets/external_groups"));
+ partition.open_persistence().await.unwrap();
+ let persistence = Rc::clone(partition.persistence.as_ref().unwrap());
+ let prepare = checksummed_segment_prepare(1, 0, 0, b"durable");
+ persistence
+ .append(prepare.clone().into_frozen(), true)
+ .unwrap();
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ partition.consensus.restore_commit_state(1, 1);
+ persistence.checkpoint(1);
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ assert_eq!(persistence.checkpoint_op(), 1);
+
+ let writer = IggyIndexWriter::new("/dev/null",
Rc::new(AtomicU64::new(0)), true, false)
+ .await
+ .unwrap();
+ assert_eq!(
+ writer
+ .save_indexes_buffered_at(vec![1; 32], 0)
+ .await
+ .unwrap(),
+ 32
+ );
+ let active = partition.log.index_writers().len() - 1;
+ partition.log.index_writers_mut()[active] = Some(Rc::new(writer));
+ let offsets = crate::state_transfer::ConsumerOffsetsWire {
+ dedup_capacity: partition.dedup().capacity(),
+ required_metadata_frontier: 0,
+ prepare_checksum: Some(prepare.header().checksum),
+ checkpoint_prepare: prepare.as_slice().to_vec(),
+ next_offset: 1,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ external_groups: Vec::new(),
+ dedup: Vec::new(),
+ owners: Vec::new(),
+ history_transition: None,
+ owner_retirement: None,
+ };
+
+ let result = partition
+ .install_state_transfer(&repair_config(), 1, Vec::new(),
&offsets.encode())
+ .await;
+
+ assert!(matches!(
+ result,
+ Err(crate::state_transfer::PartitionInstallError::SwapIo { .. })
+ ));
+ assert!(partition.fatal().is_some());
+ assert!(!directory.path().join(".install-backup").exists());
+ }
+
#[compio::test]
async fn
pending_wal_prefix_keeps_pipeline_replies_and_does_not_partially_flush() {
for durability in [
@@ -26964,6 +27063,40 @@ mod tests {
#[cfg(target_os = "linux")]
const DEV_FULL: &str = "/dev/full";
+ #[cfg(target_os = "linux")]
+ #[compio::test]
+ async fn
given_an_unsyncable_messages_writer_when_installing_should_refuse_the_barrier()
{
+ let directory = tempfile::tempdir().unwrap();
+ let wal = directory.path().join("wal");
+ let (persistence, _) = PartitionPersistence::open(&wal, 42,
7).await.unwrap();
+ let messages_writer =
+ MessagesWriter::new("/dev/null", Rc::new(AtomicU64::new(0)),
false, false, None)
+ .await
+ .unwrap();
+ let index_path = directory.path().join("segment.index");
+ let index_writer = IggyIndexWriter::new(
+ index_path.to_str().unwrap(),
+ Rc::new(AtomicU64::new(0)),
+ false,
+ false,
+ )
+ .await
+ .unwrap();
+ let mut partition = test_partition();
+ partition.log.add_persisted_segment(
+ Segment::new(0, IggyByteSize::from(1024 * 1024_u64)),
+ SegmentStorage::default(),
+ Some(Rc::new(messages_writer)),
+ Some(Rc::new(index_writer)),
+ );
+ partition.persistence = Some(Rc::clone(&persistence));
+
+ partition.barrier_install_files_locked();
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ assert!(persistence.failure().is_some());
+ }
+
const FIRST_PAYLOAD: &[u8] = b"first-chunk";
const SECOND_PAYLOAD: &[u8] = b"second-chunk-is-longer";
diff --git a/core/partitions/src/install_backup.rs
b/core/partitions/src/install_backup.rs
index 537682529..4c669b8ce 100644
--- a/core/partitions/src/install_backup.rs
+++ b/core/partitions/src/install_backup.rs
@@ -15,12 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use crate::CREATED_REVISION_FILE;
+use std::collections::BTreeSet;
+use std::io;
+use std::path::{Path, PathBuf};
+
+use futures::TryStreamExt;
use journal::durable_storage::{DiskStorage, DurableFile, DurableStorage,
OpenMode};
use journal::partition_journal::FRONTIER_FILE_NAME;
use server_common::fatal::NoteDescriptorExhaustion;
-use std::io;
-use std::path::Path;
+
+use crate::{CREATED_REVISION_FILE, FILE_SYNC_CONCURRENCY};
const BACKUP: &str = ".install-backup";
const BUILDING: &str = ".install-building";
@@ -58,26 +62,33 @@ pub async fn recover_with_storage<S: DurableStorage>(
}
storage.remove_tree(&directory.join(&entry.name)).await?;
}
- link_tree(&backup, directory, false, storage).await?;
+ link_tree(&backup, directory, false, &BTreeSet::new(), storage).await?;
finish_with_storage(directory, storage).await
}
/// Freeze the old materialization and WAL together before a destructive
install.
+///
/// The caller drains persistence and holds the partition write lock until
finish.
+/// Each path in `synced_files` must still name the inode synced by its
original
+/// writer, with no further writes to that inode before `begin` returns. These
+/// paths skip a fresh sync of the backup hard link.
///
/// # Errors
/// Returns an error if the rollback state cannot be made durable.
/// After any failure the caller must stop serving until recovery.
-pub async fn begin(directory: &Path) -> io::Result<()> {
- begin_with_storage(directory, &DiskStorage)
+pub async fn begin(directory: &Path, synced_files: &BTreeSet<PathBuf>) ->
io::Result<()> {
+ begin_with_storage(directory, synced_files, &DiskStorage)
.await
.note_descriptor_exhaustion(|| format!("starting an install in {}",
directory.display()))
}
+/// Storage-generic install entry point with the same `synced_files` contract
as [`begin`].
+///
/// # Errors
/// Returns an error when the install transaction cannot complete durably.
pub async fn begin_with_storage<S: DurableStorage>(
directory: &Path,
+ synced_files: &BTreeSet<PathBuf>,
storage: &S,
) -> io::Result<()> {
if storage.exists(&directory.join(BACKUP)).await? {
@@ -87,7 +98,7 @@ pub async fn begin_with_storage<S: DurableStorage>(
storage.remove_tree(&building).await?;
storage.remove_tree(&directory.join(RETIRED)).await?;
storage.create_directories(&building).await?;
- link_tree(directory, &building, true, storage).await?;
+ link_tree(directory, &building, true, synced_files, storage).await?;
storage.rename(&building, &directory.join(BACKUP)).await?;
storage.sync_directory(directory).await
}
@@ -122,10 +133,12 @@ async fn link_tree<S: DurableStorage>(
source: &Path,
target: &Path,
skip_scratch: bool,
+ synced_files: &BTreeSet<PathBuf>,
storage: &S,
) -> io::Result<()> {
let mut pending = vec![(source.to_path_buf(), target.to_path_buf())];
let mut directories = Vec::new();
+ let mut files_to_sync = Vec::new();
while let Some((source, target)) = pending.pop() {
directories.push(target.clone());
for entry in storage.entries(&source).await? {
@@ -150,21 +163,36 @@ async fn link_tree<S: DurableStorage>(
} else {
// Transfer unlinks or atomically replaces these frozen files.
// Hard links retain the old bytes without copying segment
data.
- storage.hard_link(&source.join(&name), &destination).await?;
- storage
- .open(&destination, OpenMode::Read)
- .await?
- .sync()
- .await?;
+ let source_file = source.join(&name);
+ storage.hard_link(&source_file, &destination).await?;
+ if !synced_files.contains(&source_file) {
+ files_to_sync.push(destination);
+ if files_to_sync.len() == FILE_SYNC_CONCURRENCY {
+ sync_files(&mut files_to_sync, storage).await?;
+ }
+ }
}
}
}
+ sync_files(&mut files_to_sync, storage).await?;
for directory in directories.into_iter().rev() {
storage.sync_directory(&directory).await?;
}
Ok(())
}
+async fn sync_files<S: DurableStorage>(files: &mut Vec<PathBuf>, storage: &S)
-> io::Result<()> {
+ futures::stream::iter(std::mem::take(files).into_iter().map(Ok::<_,
io::Error>))
+ .try_for_each_concurrent(FILE_SYNC_CONCURRENCY, |destination| async
move {
+ storage
+ .open(&destination, OpenMode::Read)
+ .await?
+ .sync()
+ .await
+ })
+ .await
+}
+
async fn copy_file<S: DurableStorage>(
source: &Path,
destination: &Path,
@@ -207,7 +235,7 @@ mod tests {
] {
std::fs::write(root.join(name), name.as_bytes()).unwrap();
}
- begin(root).await.unwrap();
+ begin(root, &BTreeSet::new()).await.unwrap();
// Model the install's unlink and atomic replacement operations.
for name in ["0.log", "offsets/1", "prepares-1/frontier"] {
std::fs::remove_file(root.join(name)).unwrap();
@@ -229,12 +257,36 @@ mod tests {
recover(root).await.unwrap();
}
+ #[compio::test]
+ async fn
given_more_files_than_one_sync_batch_when_recovering_should_restore_them_all() {
+ let directory = tempfile::tempdir().unwrap();
+ let root = directory.path();
+ for index in 0..=FILE_SYNC_CONCURRENCY {
+ std::fs::write(root.join(format!("{index}.log")), b"old").unwrap();
+ }
+
+ begin(root, &BTreeSet::new()).await.unwrap();
+ for index in 0..=FILE_SYNC_CONCURRENCY {
+ let path = root.join(format!("{index}.log"));
+ std::fs::remove_file(&path).unwrap();
+ std::fs::write(path, b"new").unwrap();
+ }
+ recover(root).await.unwrap();
+
+ for index in 0..=FILE_SYNC_CONCURRENCY {
+ assert_eq!(
+ std::fs::read(root.join(format!("{index}.log"))).unwrap(),
+ b"old"
+ );
+ }
+ }
+
#[compio::test]
async fn completed_install_never_restores_the_old_materialization() {
let directory = tempfile::tempdir().unwrap();
let root = directory.path();
std::fs::write(root.join("0.log"), b"old").unwrap();
- begin(root).await.unwrap();
+ begin(root, &BTreeSet::new()).await.unwrap();
std::fs::remove_file(root.join("0.log")).unwrap();
std::fs::write(root.join("0.log"), b"new").unwrap();
File::open(root.join("0.log"))
@@ -259,7 +311,7 @@ mod tests {
crate::write_created_revision(partition_dir, CREATED_REVISION)
.await
.unwrap();
- begin(root).await.unwrap();
+ begin(root, &BTreeSet::new()).await.unwrap();
// The delete walk removes the backup's files in no fixed order, the
// marker last only at the top level.
for entry in std::fs::read_dir(root.join(BACKUP)).unwrap() {
diff --git a/core/partitions/src/io.rs b/core/partitions/src/io.rs
index e2fa4eaca..df26976bd 100644
--- a/core/partitions/src/io.rs
+++ b/core/partitions/src/io.rs
@@ -386,6 +386,7 @@ pub enum TransferFileJob {
Backup {
directory: String,
begin: bool,
+ synced_files: std::collections::BTreeSet<std::path::PathBuf>,
},
Sweep {
directory: String,
@@ -876,10 +877,14 @@ impl TransferFileJob {
crate::state_transfer::discard_offset_writes(&writes).await;
TransferFileResult::Finished(Ok(()))
}
- Self::Backup { directory, begin } => {
+ Self::Backup {
+ directory,
+ begin,
+ synced_files,
+ } => {
let path = std::path::Path::new(&directory);
let outcome = if begin {
- crate::install_backup::begin(path).await
+ crate::install_backup::begin(path, &synced_files).await
} else {
crate::install_backup::finish(path).await
};
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 6ca10b735..2b2bc00bc 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -33,8 +33,7 @@ pub mod offset_storage;
mod partition_storage;
mod persistence;
mod poll_plan;
-#[cfg(feature = "simulator")]
-pub use persistence::CheckpointBarrier;
+pub use persistence::FileSyncBarrier;
pub(crate) use persistence::PersistenceDrain;
pub use persistence::{
PartitionPersistence, PersistenceCompletion, PersistenceMetrics,
PersistenceNotifier,
@@ -78,6 +77,9 @@ pub const DEFAULT_OFFSET_RESERVATION_LEASE: u32 = 64 * 1024;
/// Shipped per-kind durable consumer-offset limit for one partition.
pub const DEFAULT_CONSUMER_OFFSETS_MAX: usize = 4096;
+
+pub(crate) const FILE_SYNC_CONCURRENCY: usize = 16;
+
pub use iggy_partition::{PollCompletion, PollReplication};
pub use messages_writer::MessagesWriter;
pub use offset_storage::delete_persisted_offset;
diff --git a/core/partitions/src/messages_writer.rs
b/core/partitions/src/messages_writer.rs
index 7b1db97cc..126778451 100644
--- a/core/partitions/src/messages_writer.rs
+++ b/core/partitions/src/messages_writer.rs
@@ -164,10 +164,15 @@ impl MessagesWriter {
///
/// Returns an error if the file cannot be synchronized.
pub async fn fsync(&self) -> Result<(), IggyError> {
- self.file
- .sync_data()
- .await
- .map_err(|_| IggyError::CannotWriteToFile)?;
+ self.file.sync_data().await.map_err(|err| {
+ error!(
+ target: "iggy.partitions.storage",
+ file = self.file_path.as_str(),
+ %err,
+ "failed to sync segment file"
+ );
+ IggyError::CannotWriteToFile
+ })?;
Ok(())
}
}
diff --git a/core/partitions/src/persistence.rs
b/core/partitions/src/persistence.rs
index e11f17eb6..ce8079a65 100644
--- a/core/partitions/src/persistence.rs
+++ b/core/partitions/src/persistence.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::FILE_SYNC_CONCURRENCY;
use crate::offset_storage::{OffsetFilePermit, RetainedOffsetFile,
RetainedOffsetFiles};
use futures::TryStreamExt;
use iggy_binary_protocol::{Operation, PrepareHeader};
@@ -91,19 +92,19 @@ pub type PersistenceNotifier = Rc<dyn
Fn(PersistenceCompletion)>;
/// A file durability barrier that keeps the original writer alive until sync.
#[must_use]
-pub struct CheckpointBarrier {
+pub struct FileSyncBarrier {
path: PathBuf,
sync: Pin<Box<dyn Future<Output = io::Result<()>> + 'static>>,
}
-impl CheckpointBarrier {
- /// Retain a simulator file's original writer until checkpoint
synchronizes it.
+impl FileSyncBarrier {
+ /// Retains a simulator file's original writer until its sync runs.
#[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(
+ pub(crate) fn from_future(
path: impl Into<PathBuf>,
sync: impl Future<Output = io::Result<()>> + 'static,
) -> Self {
@@ -118,7 +119,9 @@ impl CheckpointBarrier {
}
async fn run(self) -> io::Result<PathBuf> {
- self.sync.await?;
+ self.sync.await.map_err(|error| {
+ io::Error::new(error.kind(), format!("{}: {error}",
self.path.display()))
+ })?;
Ok(self.path)
}
}
@@ -430,12 +433,17 @@ enum Mutation<S: DurableStorage> {
epoch: u64,
from_op: u64,
},
+ Barrier {
+ epoch: u64,
+ barriers: Vec<FileSyncBarrier>,
+ offset_files: Vec<RetainedOffsetFile<S::File>>,
+ },
Checkpoint {
epoch: u64,
through_op: u64,
files: Vec<PathBuf>,
directories: Vec<PathBuf>,
- barriers: Vec<CheckpointBarrier>,
+ barriers: Vec<FileSyncBarrier>,
offset_files: Vec<RetainedOffsetFile<S::File>>,
synced_files: BTreeSet<PathBuf>,
},
@@ -882,6 +890,8 @@ impl<S: DurableStorage> PartitionPersistence<S> {
}
pub fn checkpoint(&self, through_op: u64) {
+ // This shorthand has no materialized paths, so
+ // [`FileSyncBarrier`] has no original writer to retain.
self.checkpoint_files(through_op, Vec::new(), Vec::new(), Vec::new());
}
@@ -890,7 +900,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
through_op: u64,
files: Vec<PathBuf>,
directories: Vec<PathBuf>,
- barriers: Vec<CheckpointBarrier>,
+ barriers: Vec<FileSyncBarrier>,
) {
if through_op <= self.checkpoint_requested.get() {
return;
@@ -909,6 +919,20 @@ impl<S: DurableStorage> PartitionPersistence<S> {
});
}
+ /// Queue a durability barrier through retained original writers without
+ /// advancing [`PartitionPersistence::checkpoint_op`]. The returned paths
+ /// count as synced only after the queued barrier succeeds.
+ pub fn barrier_files(&self, barriers: Vec<FileSyncBarrier>) ->
BTreeSet<PathBuf> {
+ let (offset_files, mut synced_files) =
self.offset_files.take_checkpoint();
+ synced_files.extend(barriers.iter().map(|barrier|
barrier.path.clone()));
+ self.queue.borrow_mut().push_back(Mutation::Barrier {
+ epoch: self.epoch.get(),
+ barriers,
+ offset_files,
+ });
+ synced_files
+ }
+
pub const fn checkpoint_pending(&self) -> bool {
self.checkpoint_running.get() || self.checkpoint_requested.get() >
self.checkpoint.get()
}
@@ -958,6 +982,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
.set(self.checkpoint_requested.get().max(*through_op));
true
}
+ Mutation::Barrier { .. } => true,
_ => false,
});
self.accepted.borrow_mut().truncate_from(from_op);
@@ -1248,6 +1273,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
| Mutation::EnableSegments { epoch, .. }
| Mutation::ReanchorSegments { epoch, .. }
| Mutation::Truncate { epoch, .. }
+ | Mutation::Barrier { epoch, .. }
| Mutation::Checkpoint { epoch, .. }
| Mutation::Reset { epoch, .. } => (*epoch, 0),
};
@@ -1359,6 +1385,14 @@ impl<S: DurableStorage> PartitionPersistence<S> {
}
Mutation::Sync { .. } => journal.sync().await,
Mutation::Truncate { from_op, .. } =>
journal.truncate_from(from_op).await,
+ Mutation::Barrier {
+ barriers,
+ offset_files,
+ ..
+ } => {
+ Self::sync_files(offset_files, barriers).await?;
+ journal.sync().await
+ }
Mutation::Checkpoint {
through_op,
files,
@@ -1370,14 +1404,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
} => {
self.checkpoint_running.set(true);
let result = async {
- 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);
+ synced_files.extend(Self::sync_files(offset_files,
barriers).await?);
journal
.checkpoint_files(through_op, &files, &directories,
&synced_files)
.await
@@ -1410,6 +1437,16 @@ impl<S: DurableStorage> PartitionPersistence<S> {
}
}
+ async fn sync_files(
+ offset_files: Vec<RetainedOffsetFile<S::File>>,
+ barriers: Vec<FileSyncBarrier>,
+ ) -> io::Result<Vec<PathBuf>> {
+ futures::stream::iter(offset_files.iter().map(Ok::<_, io::Error>))
+ .try_for_each_concurrent(FILE_SYNC_CONCURRENCY, |retained|
retained.file.sync())
+ .await?;
+
futures::future::try_join_all(barriers.into_iter().map(FileSyncBarrier::run)).await
+ }
+
async fn append_batch(
&self,
journal: &mut PartitionPrepareJournal<S>,
diff --git a/core/partitions/src/state_transfer.rs
b/core/partitions/src/state_transfer.rs
index 00fcb39d5..097afc15b 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -58,7 +58,7 @@ use server_common::iobuf::Owned;
use server_common::send_messages::{decode_batch_slice, decode_prepare_slice};
use server_common::sharding::IggyNamespace;
use server_common::yield_to_reactor;
-use std::collections::HashSet;
+use std::collections::{BTreeSet, HashSet};
use std::fmt;
use std::future::Future;
use std::mem::size_of;
@@ -2040,11 +2040,13 @@ pub enum PartitionInstallError {
previous_end: u64,
next_start: u64,
},
- /// Filesystem failure at/after the swap; disk holds a contiguous prefix
- /// of the new state and a crash-restart recovers it (see the module
- /// crash-window notes). The IN-MEMORY partition is converged to an
- /// empty, honestly-lagging state before this returns, so the live
- /// process stays serviceable and the normal triggers re-transfer.
+ /// A writer sync or backup preparation failure returns before the swap;
+ /// the old in-memory partition remains in place and may be fenced.
+ /// At/after the swap, disk holds a contiguous prefix of the new state
+ /// and a crash-restart recovers it (see the module crash-window notes).
+ /// On the storeless post-swap failure path, the IN-MEMORY partition is
+ /// converged to an empty, honestly-lagging state before this returns,
+ /// so the live process stays serviceable and the normal triggers
re-transfer.
SwapIo {
path: String,
source: std::io::Error,
@@ -2411,6 +2413,7 @@ pub(crate) struct PendingInstall {
next_offset: u64,
phase: InstallPhase,
drain: Option<crate::PersistenceDrain>,
+ synced_files: BTreeSet<PathBuf>,
directory_handle: Option<compio::fs::File>,
offset_dirs_changed: [bool; ConsumerKind::COUNT],
/// Published with the offsets, replacing the local receipt table.
@@ -3331,13 +3334,18 @@ where
}
InstallPhase::Drain => {
if let Some(persistence) = &self.persistence {
+ if install.drain.is_none() {
+ // Checkpoint syncs may be skipped when the frontier
is unchanged.
+ // The backup still needs proof from each original
writer.
+ install.synced_files =
self.barrier_install_files_locked();
+ install.disposition = InstallDisposition::Backup;
+ }
self.start_persistence();
let drain = install
.drain
.get_or_insert_with(|| persistence.begin_drain());
match persistence.observe_drain(drain) {
Ok(true) => {
- install.disposition = InstallDisposition::Backup;
install.phase = InstallPhase::Backup;
}
Ok(false) => step = crate::PartitionIoStep::Pending,
@@ -3615,6 +3623,7 @@ where
file_job(crate::io::TransferFileJob::Backup {
directory: install.partition_dir.clone(),
begin: true,
+ synced_files: install.synced_files.clone(),
}),
InstallFilePhase::Backup,
),
@@ -3762,6 +3771,7 @@ where
file_job(crate::io::TransferFileJob::Backup {
directory: install.partition_dir.clone(),
begin: false,
+ synced_files: BTreeSet::new(),
}),
InstallFilePhase::FinishBackup,
),
@@ -4320,6 +4330,7 @@ where
next_offset,
phase: InstallPhase::StageOffsets(0),
drain: None,
+ synced_files: BTreeSet::new(),
directory_handle: None,
offset_dirs_changed: [false; ConsumerKind::COUNT],
retry_table: Some(retry_table),
diff --git a/core/simulator/src/storage/tests.rs
b/core/simulator/src/storage/tests.rs
index 625866083..958a0e950 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::{CheckpointBarrier, PartitionPersistence, PersistenceMetrics,
install_backup};
+use partitions::{FileSyncBarrier, PartitionPersistence, PersistenceMetrics,
install_backup};
use server_common::send_messages::{
BATCH_MESSAGE_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages,
SendMessagesOwned,
};
@@ -784,7 +784,7 @@ fn synchronized_corruption_in_any_record_block_is_refused()
{
async fn interrupted_install() -> SimStorage {
let (storage, mut journal) = baseline().await;
- install_backup::begin_with_storage(Path::new(DIRECTORY), &storage)
+ install_backup::begin_with_storage(Path::new(DIRECTORY), &BTreeSet::new(),
&storage)
.await
.unwrap();
replace(&storage, Path::new("/partition/state"), b"new")
@@ -796,36 +796,215 @@ async fn interrupted_install() -> SimStorage {
storage
}
+#[test]
+fn
given_synced_install_file_when_power_loss_interrupts_install_should_restore_backup()
{
+ 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"old".to_vec()).await.unwrap();
+ writer.sync().await.unwrap();
+ storage.sync_directory(Path::new(DIRECTORY)).await.unwrap();
+
+ let synced_files = BTreeSet::from([path.to_path_buf()]);
+ storage.clear_trace();
+ install_backup::begin_with_storage(Path::new(DIRECTORY),
&synced_files, &storage)
+ .await
+ .unwrap();
+ assert!(
+ !storage.trace().contains(&StorageOperation::FileSync),
+ "the backup synced a hard link whose original writer had passed
its barrier"
+ );
+ replace(&storage, path, b"new").await.unwrap();
+
+ storage.crash(Crash::PowerLoss);
+ install_backup::recover_with_storage(Path::new(DIRECTORY), &storage)
+ .await
+ .unwrap();
+ let file = storage.open(path, OpenMode::Read).await.unwrap();
+ assert_eq!(file.read(0, 3).await.unwrap(), b"old");
+ });
+}
+
/// A hard link preserves the inode, not the writer's error cursor. Opening the
-/// backup name after writeback failed must not authorize destructive install.
+/// backup name after writeback failed must not authorize destructive install;
+/// [`FileSyncBarrier::from_file`] must retain the original error cursor.
#[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 (persistence, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
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;
+ let barrier = FileSyncBarrier::from_file(path, writer);
+ let synced_files = persistence.barrier_files(vec![barrier]);
+ assert!(synced_files.contains(path));
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ let result = async {
+ persistence.drain().await?;
+ install_backup::begin_with_storage(Path::new(DIRECTORY),
&synced_files, &storage).await
+ }
+ .await;
+ let error = result.unwrap_err();
assert!(
- writer.sync().await.is_err(),
- "the original writer did not observe the injected writeback
failure"
+ error
+ .to_string()
+ .contains("/partition/materialized: writeback failed before
this handle synced"),
+ "unexpected barrier failure: {error}"
+ );
+ assert!(
+ !storage
+ .exists(Path::new("/partition/.install-backup"))
+ .await
+ .unwrap(),
+ "a failed backup was published"
);
+ });
+}
+
+#[test]
+fn
given_a_failed_offset_writeback_when_barriering_install_files_should_refuse_publication()
{
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let (persistence, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ let path = Path::new("/partition/offsets/consumers/1");
+ storage
+ .create_directories(path.parent().unwrap())
+ .await
+ .unwrap();
+ let mut writer = storage.open(path, OpenMode::Create).await.unwrap();
+ writer.write(0, b"pending".to_vec()).await.unwrap();
+ storage
+ .sync_directory(path.parent().unwrap())
+ .await
+ .unwrap();
+ persistence
+ .retain_offset_file(path.to_str().unwrap().to_owned(), writer)
+ .await
+ .unwrap();
+
+ storage.fail_writeback(path).unwrap();
+ let synced_files = persistence.barrier_files(Vec::new());
+ assert!(synced_files.contains(path));
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ let result = async {
+ persistence.drain().await?;
+ install_backup::begin_with_storage(Path::new(DIRECTORY),
&synced_files, &storage).await
+ }
+ .await;
+
assert!(
result.is_err(),
- "install backup published after synchronizing a fresh hard-link
handle past the writeback error"
+ "install backup published after offset writeback failed"
);
+ assert!(persistence.failure().is_some());
assert!(
!storage
.exists(Path::new("/partition/.install-backup"))
.await
- .unwrap(),
- "a failed backup was published"
+ .unwrap()
+ );
+ });
+}
+
+#[test]
+fn
given_a_truncated_install_barrier_when_checkpointing_should_report_retained_offset_writeback_failure()
+ {
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let (persistence, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ let first = prepare(1, 0);
+ let second = prepare(2, first.header().checksum);
+ persistence.append(first.into_frozen(), true).unwrap();
+ persistence.append(second.into_frozen(), true).unwrap();
+
+ let path = Path::new("/partition/offsets/consumers/1");
+ storage
+ .create_directories(path.parent().unwrap())
+ .await
+ .unwrap();
+ let mut writer = storage.open(path, OpenMode::Create).await.unwrap();
+ writer.write(0, b"pending".to_vec()).await.unwrap();
+ storage
+ .sync_directory(path.parent().unwrap())
+ .await
+ .unwrap();
+ persistence
+ .retain_offset_file(path.to_str().unwrap().to_owned(), writer)
+ .await
+ .unwrap();
+
+ storage.fail_writeback(path).unwrap();
+ persistence.barrier_files(Vec::new());
+ persistence.truncate_from(2);
+ // Any later mutation must wait for the old barrier; a checkpoint also
+ // makes an accidental durability advance observable.
+ persistence.checkpoint(1);
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ let error = persistence.drain().await.unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains("writeback failed before this handle synced"),
+ "unexpected barrier failure: {error}"
);
+ assert!(persistence.failure().is_some());
+ assert_eq!(persistence.checkpoint_op(), 0);
+ });
+}
+
+#[test]
+fn
given_a_truncated_install_barrier_when_rebarriering_should_make_retained_offset_durable()
{
+ block_on(async {
+ let storage = storage_for_partition().await;
+ let (persistence, _) =
+ PartitionPersistence::open_with_storage(Path::new(WAL), 42, 7,
storage.clone())
+ .await
+ .unwrap();
+ let first = prepare(1, 0);
+ let second = prepare(2, first.header().checksum);
+ persistence.append(first.into_frozen(), true).unwrap();
+ persistence.append(second.into_frozen(), true).unwrap();
+
+ let path = Path::new("/partition/offset");
+ let mut writer = storage.open(path, OpenMode::Create).await.unwrap();
+ writer.write(0, b"retained".to_vec()).await.unwrap();
+ storage.sync_directory(Path::new(DIRECTORY)).await.unwrap();
+ persistence
+ .retain_offset_file(path.to_str().unwrap().to_owned(), writer)
+ .await
+ .unwrap();
+
+ persistence.barrier_files(Vec::new());
+ persistence.truncate_from(2);
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ persistence.drain().await.unwrap();
+
+ persistence.barrier_files(Vec::new());
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ persistence.drain().await.unwrap();
+ storage.crash(Crash::PowerLoss);
+ let file = storage.open(path, OpenMode::Read).await.unwrap();
+ assert_eq!(file.read(0, 8).await.unwrap(), b"retained");
});
}
@@ -1179,7 +1358,7 @@ 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);
+ let barrier = FileSyncBarrier::from_file(path, file);
persistence.checkpoint_files(
4,
vec![path.to_path_buf()],
@@ -2523,7 +2702,7 @@ async fn install(
storage: &SimStorage,
journal: &mut PartitionPrepareJournal<SimStorage>,
) -> io::Result<()> {
- install_backup::begin_with_storage(Path::new(DIRECTORY), storage).await?;
+ install_backup::begin_with_storage(Path::new(DIRECTORY), &BTreeSet::new(),
storage).await?;
replace(storage, Path::new("/partition/state"), b"new").await?;
for path in MATERIALIZED_FILES {
replace(storage, Path::new(path), b"new").await?;
@@ -2778,7 +2957,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);
+ let barrier = FileSyncBarrier::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
@@ -2831,8 +3010,8 @@ fn
given_multiple_writers_for_one_checkpoint_file_when_one_has_not_observed_the_
// 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),
+ FileSyncBarrier::from_file(path, first_writer),
+ FileSyncBarrier::from_file(path, second_writer),
];
persistence.checkpoint_files(
4,
