hubcio commented on code in PR #4278:
URL: https://github.com/apache/iggy/pull/4278#discussion_r4165416928
##########
core/partitions/src/persistence.rs:
##########
@@ -903,6 +912,27 @@ impl<S: DurableStorage> PartitionPersistence<S> {
});
}
+ /// Queue a durability barrier through every retained original writer,
+ /// including offset files restored from the backup, without advancing
+ /// [`PartitionPersistence::checkpoint_op`].
+ pub(crate) fn barrier_files(&self, barriers: Vec<FileSyncBarrier>) ->
BTreeSet<PathBuf> {
+ let synced_files = self
+ .offset_files
+ .borrow()
+ .keys()
+ .map(PathBuf::from)
+ .chain(barriers.iter().map(|barrier| barrier.path.clone()))
+ .collect();
+ let mut offset_files = std::mem::take(&mut
*self.retired_offset_files.borrow_mut());
+ offset_files.extend(self.offset_files.borrow_mut().drain().map(|(_,
file)| file));
+ self.queue.borrow_mut().push_back(Mutation::Barrier {
Review Comment:
warning: if the drain times out, this barrier stays queued, and a
view-change truncate drops it at line 982, closing its offset writers unsynced.
move a dropped barrier's `offset_files` into `retired_offset_files` there -
re-stamping like `Checkpoint` republishes the pre-truncate heads.
##########
core/simulator/src/storage/tests.rs:
##########
@@ -812,12 +814,14 @@ fn
given_a_failed_writeback_when_beginning_an_install_backup_should_refuse_publi
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 result = install_backup::begin_with_storage_and_barriers(
Review Comment:
warning: this only exercises the test-only shim, so it stays green with the
production barrier stubbed out, and nothing in `partitions` or `simulator`
fails if the messages-writer barrier or the offset half of `barrier_files` goes
away. drive `barrier_files` with failed-writeback `SimStorage` writers and drop
the shim.
##########
core/partitions/src/install_backup.rs:
##########
@@ -65,17 +70,47 @@ pub async fn recover_with_storage<S: DurableStorage>(
/// # 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_synced_files(directory, synced_files, &DiskStorage)
.await
.note_descriptor_exhaustion(|| format!("starting an install in {}",
directory.display()))
}
+/// Storage-generic form of [`begin`] used by deterministic storage tests.
+///
/// # Errors
/// Returns an error when the install transaction cannot complete durably.
pub async fn begin_with_storage<S: DurableStorage>(
directory: &Path,
storage: &S,
+) -> io::Result<()> {
+ begin_with_synced_files(directory, BTreeSet::new(), storage).await
+}
+
+/// Simulator form of [`begin_with_storage`] that retains original writers in
+/// [`FileSyncBarrier`] values until their durability barriers complete.
+///
+/// # Errors
+/// Returns an error when an original-writer barrier fails or the install
+/// transaction cannot complete durably.
+#[cfg(feature = "simulator")]
+pub async fn begin_with_storage_and_barriers<S: DurableStorage>(
+ directory: &Path,
+ barriers: Vec<FileSyncBarrier>,
+ storage: &S,
+) -> io::Result<()> {
+ let synced_files =
+
futures::future::try_join_all(barriers.into_iter().map(FileSyncBarrier::run))
+ .await?
+ .into_iter()
+ .collect();
+ begin_with_synced_files(directory, synced_files, storage).await
+}
+
+async fn begin_with_synced_files<S: DurableStorage>(
+ directory: &Path,
+ synced_files: BTreeSet<PathBuf>,
Review Comment:
nit: the set is only lent to `link_tree`, so take `&BTreeSet<PathBuf>` here
and in `begin` (line 73), like `PartitionPrepareJournal::checkpoint_files` does.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1046,13 +1046,52 @@ 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 barrier_install_files_locked(&self) ->
std::io::Result<BTreeSet<PathBuf>> {
Review Comment:
nit: nothing else in the rust code uses a `_locked` suffix - other functions
state the lock precondition in their doc. drop the suffix, for example
`sync_install_writers`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1046,13 +1046,52 @@ 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 barrier_install_files_locked(&self) ->
std::io::Result<BTreeSet<PathBuf>> {
+ let Some(persistence) = &self.persistence else {
+ return Ok(BTreeSet::new());
+ };
+ let mut barriers = Vec::with_capacity(2);
+ if let Some(writer) =
self.log.messages_writers().last().and_then(Option::as_ref) {
Review Comment:
simplification: `.map_err(std::io::Error::other)` keeps the same error text
and drops the `to_string()` allocation, about 7 lines across both blocks. the
index path still needs `writer.path().to_owned()`.
##########
core/partitions/src/install_backup.rs:
##########
@@ -65,17 +70,47 @@ pub async fn recover_with_storage<S: DurableStorage>(
/// # 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_synced_files(directory, synced_files, &DiskStorage)
.await
.note_descriptor_exhaustion(|| format!("starting an install in {}",
directory.display()))
}
+/// Storage-generic form of [`begin`] used by deterministic storage tests.
+///
/// # Errors
/// Returns an error when the install transaction cannot complete durably.
pub async fn begin_with_storage<S: DurableStorage>(
directory: &Path,
storage: &S,
+) -> io::Result<()> {
+ begin_with_synced_files(directory, BTreeSet::new(), storage).await
+}
+
+/// Simulator form of [`begin_with_storage`] that retains original writers in
+/// [`FileSyncBarrier`] values until their durability barriers complete.
+///
+/// # Errors
+/// Returns an error when an original-writer barrier fails or the install
+/// transaction cannot complete durably.
+#[cfg(feature = "simulator")]
+pub async fn begin_with_storage_and_barriers<S: DurableStorage>(
Review Comment:
simplification: this shim and its `simulator`-gated import exist for one
test. once that test drives `barrier_files`, delete both - about 20 lines - and
`FileSyncBarrier::run` can go back to private.
##########
core/partitions/src/install_backup.rs:
##########
@@ -65,17 +70,47 @@ pub async fn recover_with_storage<S: DurableStorage>(
/// # 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:
nit: the doc doesn't say what `synced_files` promises, and every listed path
skips the fresh sync. spell it out: the original writer synced the path, the
path still names that inode, and nothing writes it before `begin` returns.
##########
core/partitions/src/persistence.rs:
##########
@@ -903,6 +912,27 @@ impl<S: DurableStorage> PartitionPersistence<S> {
});
}
+ /// Queue a durability barrier through every retained original writer,
+ /// including offset files restored from the backup, without advancing
Review Comment:
nit: files restored from the backup never get a retained writer, so this
barrier doesn't cover them. drop that phrase and say the returned paths only
count as synced once the queued barrier succeeds.
##########
core/partitions/src/install_backup.rs:
##########
@@ -119,6 +154,7 @@ async fn link_tree<S: DurableStorage>(
source: &Path,
target: &Path,
skip_scratch: bool,
+ synced_files: &BTreeSet<std::path::PathBuf>,
Review Comment:
nit: `PathBuf` is already imported at line 20.
##########
core/partitions/src/install_backup.rs:
##########
@@ -65,17 +70,47 @@ pub async fn recover_with_storage<S: DurableStorage>(
/// # 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_synced_files(directory, synced_files, &DiskStorage)
.await
.note_descriptor_exhaustion(|| format!("starting an install in {}",
directory.display()))
}
+/// Storage-generic form of [`begin`] used by deterministic storage tests.
Review Comment:
nit: this isn't the storage-generic form of `begin` - it passes an empty set
and fresh-syncs every link. say that instead.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2743,15 +2743,28 @@ where
let write_lock = self.write_lock.clone();
let _guard = write_lock.lock().await;
- if let Some(persistence) = &self.persistence {
- self.start_persistence();
- persistence.drain_with_timeout().await.map_err(|source| {
- PartitionInstallError::SwapIo {
- path: partition_dir.clone(),
- source,
+ if self.persistence.is_some() {
+ // [`IggyPartition::barrier_install_files_locked`] covers original
+ // writers even when no [`crate::PartitionPersistence`] checkpoint
can advance.
+ let synced_files = match self.barrier_install_files_locked().await
{
+ Ok(synced_files) => synced_files,
+ Err(source) => {
+ if self
+ .persistence
+ .as_ref()
+ .is_some_and(|persistence|
persistence.failure().is_some())
+ {
+ self.fence_install_failure(commit_op);
+ }
+ return Err(PartitionInstallError::SwapIo {
Review Comment:
nit: the `SwapIo` doc promises a converged, post-swap partition, but this
returns it before the swap. widen the doc to cover pre-swap failures.
##########
core/partitions/src/install_backup.rs:
##########
@@ -143,12 +179,15 @@ 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) {
+ storage
+ .open(&destination, OpenMode::Read)
+ .await?
+ .sync()
Review Comment:
warning: not new, but this opens and syncs every file one by one under the
write lock, which stalls the whole shard pump on partitions with thousands of
offset files. collect them and sync through `try_for_each_concurrent(16, ...)`
like `checkpoint_files` does.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1046,13 +1046,52 @@ 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 barrier_install_files_locked(&self) ->
std::io::Result<BTreeSet<PathBuf>> {
+ let Some(persistence) = &self.persistence else {
+ return Ok(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(|error| std::io::Error::other(error.to_string()))
Review Comment:
nit: both writers' `fsync` drop the `io::Error`, so this can only ever
report `Cannot write to file` without the errno. log it in `fsync` before
mapping, like `write_frozen_chunked` does.
##########
core/partitions/src/install_backup.rs:
##########
@@ -143,12 +179,15 @@ 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:
simplification: the barrier finishes before `begin`, so this fresh sync can
only add an error, never hide one - skipping it saves at most 65 syncs on a
cold path. dropping `synced_files` cuts about 25-30 lines, your call since it
undoes the earlier review ask.
##########
core/partitions/src/install_backup.rs:
##########
@@ -65,17 +70,47 @@ pub async fn recover_with_storage<S: DurableStorage>(
/// # 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_synced_files(directory, synced_files, &DiskStorage)
.await
.note_descriptor_exhaustion(|| format!("starting an install in {}",
directory.display()))
}
+/// Storage-generic form of [`begin`] used by deterministic storage tests.
+///
/// # Errors
/// Returns an error when the install transaction cannot complete durably.
pub async fn begin_with_storage<S: DurableStorage>(
Review Comment:
simplification: if you keep the skip, fold `begin_with_synced_files` into
this and have it take `synced_files`, so `begin` wraps one function like
`recover` and `finish` do. that saves about 15 lines and replaces the four
empty-set comments with one doc line.
##########
core/partitions/src/persistence.rs:
##########
@@ -903,6 +912,27 @@ impl<S: DurableStorage> PartitionPersistence<S> {
});
}
+ /// Queue a durability barrier through every retained original writer,
+ /// including offset files restored from the backup, without advancing
+ /// [`PartitionPersistence::checkpoint_op`].
+ pub(crate) fn barrier_files(&self, barriers: Vec<FileSyncBarrier>) ->
BTreeSet<PathBuf> {
+ let synced_files = self
Review Comment:
simplification: this repeats the snapshot-then-drain from `checkpoint_files`
(lines 896-903). a shared private helper keeps that order in one place, though
it only saves a couple of lines.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1046,13 +1046,52 @@ 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 barrier_install_files_locked(&self) ->
std::io::Result<BTreeSet<PathBuf>> {
+ let Some(persistence) = &self.persistence else {
Review Comment:
simplification: the only caller already checks `self.persistence.is_some()`,
so this `else` arm is dead, and the caller re-reads the same `Option` at
state_transfer.rs:2752. bind it once there and pass the persistence in.
--
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]
