diegomrsantos commented on code in PR #4240:
URL: https://github.com/apache/iggy/pull/4240#discussion_r4062305073
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8416,6 +8478,37 @@ 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 Ok(entries) = storage.regular_files(Path::new(directory)).await else {
Review Comment:
Added a warning with the directory path and error in 5481f6fa2. The existing
fallback is unchanged.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -346,61 +381,96 @@ 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.
///
-/// 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 also treated as absence.
///
/// 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 instead: 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).
+/// Propagates an open or read failure after the existence probe, except a
short read.
pub async fn read_purge_generation(path: &str, created_revision: u64) ->
Result<u64, IggyError> {
- if !Path::new(path).exists() {
+ read_purge_generation_with_storage(&DiskStorage, path,
created_revision).await
+}
+
+/// Recover a purge marker through the supplied storage backend.
+///
+/// Uses the same absent, torn, and incarnation mismatch handling as
+/// [`read_purge_generation`], including treating a failed existence probe as
+/// absence. Once the file is opened, read failures other than a short record
+/// propagate so an unreadable marker does not trigger another purge.
+///
+/// # Errors
+/// Returns an error if opening or reading a file found by the probe fails.
+pub async fn read_purge_generation_with_storage<S: DurableStorage>(
+ storage: &S,
+ path: &str,
+ created_revision: u64,
+) -> Result<u64, IggyError> {
+ if !storage
Review Comment:
The probe error is now logged before returning generation zero.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -346,61 +381,96 @@ 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.
///
-/// 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 also treated as absence.
///
/// 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 instead: 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).
+/// Propagates an open or read failure after the existence probe, except a
short read.
pub async fn read_purge_generation(path: &str, created_revision: u64) ->
Result<u64, IggyError> {
Review Comment:
Removed the wrapper, updated the tests to use `DiskStorage`, and moved its
docs to the remaining function. The doctest check passes with `-D warnings`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8416,6 +8478,37 @@ where
}
}
+async fn purge_offset_files<S: DurableStorage>(
Review Comment:
Reused `numeric_offset_id`.
##########
core/server/src/partition_helpers.rs:
##########
@@ -140,26 +140,66 @@ 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
/// lands.
///
/// # 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, B, SB>(
Review Comment:
Dropped `B` and `SB`. Only the storage backend is generic now.
##########
core/journal/src/durable_storage.rs:
##########
@@ -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-offset-recovery".to_owned())
Review Comment:
Renamed it to `iggy-file-scan` and updated the comment. All six changes are
pushed and pass local checks. Could you take another look?
--
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]
