hubcio commented on code in PR #4278:
URL: https://github.com/apache/iggy/pull/4278#discussion_r4174552263
##########
core/partitions/src/persistence.rs:
##########
@@ -949,6 +977,10 @@ impl<S: DurableStorage> PartitionPersistence<S> {
.set(self.checkpoint_requested.get().max(*through_op));
true
}
+ Mutation::Barrier { offset_files, .. } => {
Review Comment:
warning: my hand-back suggestion leaves a gap - a `Checkpoint` queued behind
this barrier only syncs fresh handles, so after a writeback error it advances
past the unsynced writer. keep the barrier with its old epoch
(`Mutation::Barrier { .. } => true`) - it syncs the original writers and
publishes nothing. the first drain in simulator `tests.rs:928` has to expect
that error.
##########
core/partitions/src/install_backup.rs:
##########
@@ -150,15 +163,23 @@ 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) {
Review Comment:
nit: real state transfers take this skip, but no test checks it - both
simulator calls with a non-empty set fail in `drain()` first. if the set stays,
add a round trip: good barrier, `begin_with_storage` with the set, power loss,
recovery.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1048,13 +1048,46 @@ 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()));
}
let (files, directories) = self.persistence_checkpoint_files(config);
persistence.checkpoint_files(through_op, files, directories, barriers);
self.start_persistence();
}
+ /// Synchronize every live original writer while the caller holds
+ /// [`IggyPartition::write_lock`].
+ ///
+ /// Unlike [`IggyPartition::checkpoint_persistence`], this barrier runs
even
+ /// when the checkpoint frontier cannot advance.
+ pub(crate) async fn sync_install_writers(
+ &self,
+ persistence: &PartitionPersistence,
+ ) -> std::io::Result<BTreeSet<PathBuf>> {
+ 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)
+ }));
+ }
+ let synced_files = persistence.barrier_files(barriers);
+ self.start_persistence();
Review Comment:
nit: this starts `self.persistence` but queues and drains the `persistence`
argument - fine today since the only caller passes the same `Rc`, but it's a
hidden precondition. keep the parameter and add
`debug_assert!(self.persistence.as_deref().is_some_and(|own| std::ptr::eq(own,
persistence)))`.
##########
core/partitions/src/install_backup.rs:
##########
@@ -150,15 +163,23 @@ 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);
+ }
}
}
}
+ futures::stream::iter(files_to_sync.into_iter().map(Ok::<_, io::Error>))
+ .try_for_each_concurrent(16, |destination| async move {
Review Comment:
nit: bare `16` sync width, same as in `sync_files` (persistence.rs:1392).
name one crate-private constant for both - not `OFFSET_PERSIST_CONCURRENCY`,
which sizes offset writes, not syncs.
##########
core/simulator/src/storage/tests.rs:
##########
@@ -800,24 +800,33 @@ async fn interrupted_install() -> SimStorage {
}
/// 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;
- assert!(
- writer.sync().await.is_err(),
- "the original writer did not observe the injected writeback
failure"
- );
assert!(
Review Comment:
nit: this passes on any error from `drain()`. assert the text like the
sibling at line 934 - `/partition/materialized: writeback failed before this
handle synced` - which also pins the original-writer fault.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -16300,6 +16389,38 @@ 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_FULL, Rc::new(AtomicU64::new(0)), false,
false, None)
Review Comment:
nit: this writes nothing - it fails on `fdatasync` (`EINVAL` on any char
device), not the `ENOSPC` that the `DEV_FULL` doc promises. open `/dev/null`
like the sync-fault tests at 9360 and 9401.
##########
core/partitions/src/persistence.rs:
##########
@@ -903,6 +911,26 @@ 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 synced_files = self
Review Comment:
simplification: this repeats the `offset_files` key snapshot from
`checkpoint_files` right before `drain_offset_files`. return the key set from
`drain_offset_files` instead - saves ~6 lines.
##########
core/partitions/src/install_backup.rs:
##########
@@ -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<()> {
Review Comment:
simplification: real transfers take the `synced_files` skip and nothing
tests it, so dropping the set is worth another look - at most 66 extra syncs of
clean files, and an extra error only fails the install closed. saves ~27 lines.
##########
core/partitions/src/persistence.rs:
##########
@@ -131,19 +131,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.
Review Comment:
nit: the `simulator` feature comment in `core/partitions/Cargo.toml` still
says "original-writer checkpoint barriers", but `FileSyncBarrier` now backs
install barriers too. make it "original-writer file sync barriers" to match
this doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
