diff --git a/core/journal/src/partition_journal.rs b/core/journal/src/partition_journal.rs index 55502834f4..92bd8a5eb2 100644 --- a/core/journal/src/partition_journal.rs +++ b/core/journal/src/partition_journal.rs @@ -1381,6 +1381,11 @@ impl PartitionPrepareJournal { checkpoint_prepare: false, ..self.state }; + if truncate == Some(0) { + // Replacement history can rewind the operation sequence. Keeping a + // higher purge floor would discard new operations after the install. + state.purge_floor = state.purge_floor.min(checkpoint); + } if let Some(segments) = &mut state.segment_storage { if checkpoint > self.state.checkpoint { segments.checkpoint = self @@ -2614,6 +2619,42 @@ mod tests { assert_eq!(journal.certified_log_view(), None); } + #[compio::test] + async fn transferred_checkpoint_clamps_the_purge_floor_after_restart() { + let directory = tempdir().unwrap(); + let mut journal = PartitionPrepareJournal::open(directory.path(), 42, 7) + .await + .unwrap(); + let mut parent_checksum = 0; + for op in 1..=9 { + let message = prepare(op, parent_checksum); + parent_checksum = message.header().checksum; + journal.append(message.into_frozen()).await.unwrap(); + } + journal.mark_purge(3, 9).await.unwrap(); + assert_eq!(journal.purge_marker(), (3, 9)); + + // The installed history restarts new operations at 5. Its purge floor + // must not classify those operations as part of the discarded history. + journal.reset(4, None).await.unwrap(); + drop(journal); + + let mut journal = PartitionPrepareJournal::open(directory.path(), 42, 7) + .await + .unwrap(); + assert_eq!(journal.checkpoint_op(), 4); + assert_eq!(journal.purge_marker(), (3, 4)); + let fresh = prepare(5, 1234); + journal.append(fresh.clone().into_frozen()).await.unwrap(); + drop(journal); + + let journal = PartitionPrepareJournal::open(directory.path(), 42, 7) + .await + .unwrap(); + assert!(journal.contains(fresh.header())); + assert!(fresh.header().op > journal.purge_marker().1); + } + #[compio::test] async fn transferred_checkpoint_retains_its_prepare_after_restart() { let directory = tempdir().unwrap(); diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 2c9d77d809..8152b5dcbb 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -24,9 +24,10 @@ use crate::log::JournalInfo; use crate::log::SegmentedLog; use crate::messages_writer::MessagesWriter; use crate::offset_storage::{ - PURGE_GENERATION_FILE, delete_persisted_offset, delete_persisted_offset_with_storage, - persist_offset, persist_offset_max, persist_purge_generation_with_storage, - read_purge_generation, + PURGE_GENERATION_FILE, PURGE_RESET_FILE, PurgeReset, delete_persisted_offset, + delete_persisted_offset_with_storage, persist_offset, persist_offset_max, + persist_purge_generation_with_storage, persist_purge_reset_with_storage, read_purge_generation, + read_purge_reset, }; use crate::persistence::{ CheckpointBarrier, PartitionPersistence, PersistenceCompletion, PersistenceNotifier, @@ -252,6 +253,12 @@ where /// generation against this and resets only when it advances, so a redundant /// reconcile pass never re-wipes a partition already at this generation. pub(crate) applied_purge_generation: u64, + /// Durable message reset boundary. When its generation exceeds the applied + /// generation, retries finish bookmark cleanup without resetting messages again. + purge_reset: Option, + /// A reset survived without its operation journal. Only a selected empty + /// history or an installed snapshot can establish its replay boundary. + pub(crate) purge_recovery: PurgeRecovery, /// `Partition::created_revision` of the metadata row this partition was /// built for (the reconciler's "epoch"). Keys the durable `purge.gen` /// record: a delete whose on-disk cleanup failed leaves the directory @@ -259,14 +266,11 @@ where /// the dead incarnation's record must not hydrate. `0` for partitions built /// without a metadata row (tests, in-memory storage). pub(crate) created_revision: u64, - /// Highest consensus op assigned when the last purge ran. INVARIANT: every - /// journal-apply path must no-op entries with `op <= purge_floor_op`. The - /// purge keeps journal entries resident (consensus history for backups, - /// repair and retransmission) while wiping the segments, so without the - /// floor a pre-purge op committing after the purge would flush purged - /// bytes back into a fresh segment or re-advance the reset offset. Not - /// persisted: the in-memory journal dies with the process, so no resident - /// pre-purge entry survives a restart. + /// Highest consensus operation invalidated by the last message reset. + /// Every journal apply path skips entries at or below this floor, preserving + /// consensus history without restoring purged messages. Cluster recovery + /// restores the floor from `purge.reset` and the durable journal marker. + /// Singleton recovery starts a new operation sequence and does not restore it. purge_floor_op: u64, /// Durable superblock for this partition's consensus group, recording /// `(view, log_view)` across a crash so this replica can never @@ -290,15 +294,9 @@ where /// terminal policy. superblock_write_failures: Cell, superblock_retry_after_micros: Cell, - /// A committed purge this replica accepted but could not apply, because it - /// could not record the frontier reset first. Withholds `PrepareOk` until - /// the purge lands: the counter still names the PRE-purge offset space, so - /// every op acked meanwhile would be stamped from a `base_offset` the peers - /// that already purged do not share. - /// - /// The superblock persist gate cannot cover this on its own -- it fires on - /// `(view, log_view)` changes, and a replica with a stable view and a full - /// disk attempts no write, observes no failure, and fences nothing. + /// Withhold prepare acknowledgements until a committed purge finishes. + /// Committed entries from peers may still arrive during this interval, so + /// retries must preserve data above the durable message reset boundary. pub(crate) purge_deferred: bool, /// The `offset_frontier` the last successful superblock write recorded, /// seeded at boot from the record that write left behind. @@ -432,13 +430,18 @@ enum Disposition { }, } -/// Why a purge did not complete, split by whether it had already mutated. +/// Whether a recovered purge boundary has been matched to the selected history. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PurgeRecovery { + Ready, + AwaitingHistory, +} + +/// How far purge progressed, determining whether to retry or fence the partition. /// -/// The two need opposite handling, and conflating them is a data-loss bug: -/// fencing a partition whose purge failed before it touched anything -/// quarantines a complete healthy chain while the live counter still names the -/// pre-purge offset space, and the fence's own frontier write then stamps that -/// stale counter as durable truth. +/// Before message reset, retry the whole operation. After a durable reset, retry +/// bookmark cleanup and completion while preserving new messages. A reset without +/// a usable chain or durable boundary requires fencing before accepting more work. #[derive(Debug)] pub enum PurgeError { /// The frontier reset could not be recorded. NOTHING was mutated: the @@ -463,20 +466,14 @@ pub enum PurgeError { /// `ENOSPC` / `EIO` is logged by the superblock writer on the first failure /// and at every power-of-two thereafter. FrontierNotRecorded, - /// The wipe ran and the fresh chain is planted, but the applied purge - /// generation could not be recorded durably (`purge.gen`), so - /// `applied_purge_generation` stays at its pre-purge value and the - /// reconciler re-issues the purge. Retry, do not fence: the partition is - /// serviceable and re-purging an already-empty chain is cheap. - /// - /// Sets `purge_deferred` for the same reason as - /// [`Self::FrontierNotRecorded`]: an op acked between this failure and the - /// retry would be wiped by that retry while every peer that recorded the - /// generation keeps it. + /// Message reset is durable, but bookmark deletion or directory sync failed. + /// Retry cleanup at the same generation without resetting message history. + OffsetCleanupNotRecorded(IggyError), + /// Message reset and bookmark cleanup completed, but `purge.gen` could not + /// be recorded. Retry completion while preserving messages above the reset floor. GenerationNotRecorded(IggyError), - /// A step after the drain failed, so the partition holds no serviceable - /// segment chain and its next append would panic on `active_segment()`. - /// The caller must fence this group for rebuild. + /// Message reset could not establish a serviceable chain or a durable reset + /// boundary. Fence the group before accepting writes that a retry could erase. Unserviceable(IggyError), } @@ -487,14 +484,18 @@ impl fmt::Display for PurgeError { f, "could not record the purge's offset-frontier reset; nothing was mutated" ), + Self::OffsetCleanupNotRecorded(source) => write!( + f, + "purge reset message history but bookmark cleanup is not durable: {source}" + ), Self::GenerationNotRecorded(source) => write!( f, "purge reset the partition but could not record its applied generation; \ - the purge will be re-issued: {source}" + completion will be retried without resetting messages: {source}" ), Self::Unserviceable(source) => write!( f, - "purge left the partition without a serviceable chain: {source}" + "purge could not establish a durable message reset: {source}" ), } } @@ -640,6 +641,8 @@ where #[cfg(test)] offset_dir_sync_count: Cell::new(0), applied_purge_generation: 0, + purge_reset: None, + purge_recovery: PurgeRecovery::Ready, created_revision: 0, purge_floor_op: 0, superblock: None, @@ -686,26 +689,23 @@ where self.created_revision = created_revision; } - /// Seed [`Self::applied_purge_generation`] from the partition dir's - /// `purge.gen` file at build time (both fresh create and recovery walk - /// this). Absent file reads 0, so a partition that never purged and a - /// repair-rebuilt dir both start below any committed generation and the - /// reconciler re-applies the purge; a crash AFTER a purge's durable - /// generation write correctly skips the re-wipe, keeping messages - /// appended since. A record left by a PREVIOUS incarnation of this - /// namespace reads 0 as well (see [`read_purge_generation`]). No-op - /// without a partition dir (in-memory storage). + /// Load the completed purge generation and durable message reset boundary. + /// + /// Recovery must set the partition directory and incarnation first, then call + /// this before replaying journal entries. A pending reset restores its original + /// floor without claiming that bookmark cleanup completed. The offset loader + /// calls [`Self::resume_purge_cleanup_with_storage`] before reading bookmarks. + /// Existing partitions without a reset record retain the legacy recovery path. /// /// # Errors - /// Propagates a real I/O failure reading `purge.gen`: booting with the - /// sentinel 0 instead would make the reconciler silently re-purge and - /// destroy post-purge messages, so the boot fails loud. + /// Propagates marker read failures and rejects damaged reset records. Treating + /// those as absence could reset history that already contains fresh messages. pub async fn hydrate_applied_purge_generation(&mut self) -> Result<(), IggyError> { self.hydrate_applied_purge_generation_with_storage(&DiskStorage) .await } - /// Restore the applied generation from the filesystem used for purge cleanup. + /// Restore the completed generation and message reset boundary from storage. /// /// Set the partition directory and its creation revision before calling this. /// A simulator must call this on a new partition after discarding its volatile @@ -721,10 +721,164 @@ where let path = format!("{dir}/{PURGE_GENERATION_FILE}"); self.applied_purge_generation = read_purge_generation(storage, &path, self.created_revision).await?; + self.purge_reset = read_purge_reset( + storage, + &format!("{dir}/{PURGE_RESET_FILE}"), + self.created_revision, + ) + .await? + .filter(|reset| reset.generation >= self.applied_purge_generation); + if let Some(reset) = self.purge_reset { + // A singleton starts a new operation sequence on restart. A cluster + // must establish the selected history before discarding its floor. + if self.consensus.replica_count() > 1 { + self.purge_floor_op = self.purge_floor_op.max(reset.floor); + } + self.purge_deferred = reset.generation > self.applied_purge_generation; + } + // A snapshot can publish a purge generation without a local reset. + // Its journal is just as volatile as one cleared by a local purge. + if self.consensus.replica_count() > 1 + && !self.durability().is_persisted() + && !self.consumer_offset_durability().is_persisted() + && (self.applied_purge_generation > 0 || self.purge_reset.is_some()) + { + self.purge_recovery = PurgeRecovery::AwaitingHistory; + } } Ok(()) } + #[must_use] + /// The reset survived a restart without a durable operation journal. + /// Until history selection or snapshot installation resolves it, this replica + /// must not serve reads, accept writes or acknowledge replicated operations. + pub const fn purge_recovery_pending(&self) -> bool { + matches!(self.purge_recovery, PurgeRecovery::AwaitingHistory) + } + + /// Resolve a reset whose operation journal was lost, after consensus selects + /// a history. Call before starting the selected view or replaying its entries. + /// The caller must establish freshness through a new election or a reply to + /// this boot's view probe before accepting an empty history. + /// An empty history starts a new sequence. A nonempty history needs a snapshot: + /// its operation numbers alone cannot distinguish fresh sends from purged ones. + /// + /// # Errors + /// A failed reset record write leaves recovery fenced for a later retry. + pub async fn recover_purge_boundary(&mut self, selected_head: u64) -> Result<(), IggyError> { + self.recover_purge_boundary_with_storage(&DiskStorage, selected_head) + .await + } + + async fn recover_purge_boundary_with_storage( + &mut self, + storage: &S, + selected_head: u64, + ) -> Result<(), IggyError> { + if !self.purge_recovery_pending() { + return Ok(()); + } + if selected_head != 0 { + if self.consensus.state_transfer_stage() == consensus::StateTransferStage::Idle { + self.consensus.begin_state_transfer_await(); + } + return Ok(()); + } + let write_lock = self.write_lock.clone(); + let _guard = write_lock.lock().await; + if let Some(reset) = self.purge_reset { + let rebased = PurgeReset { floor: 0, ..reset }; + if let Some(directory) = self.partition_dir() { + persist_purge_reset_with_storage( + storage, + &format!("{directory}/{PURGE_RESET_FILE}"), + rebased, + self.created_revision, + ) + .await?; + } + self.purge_reset = Some(rebased); + } + self.purge_floor_op = 0; + self.log.journal().inner.clear_poll_index(0); + self.transfer = None; + self.transfer_rearm = None; + if self.consensus.state_transfer_stage() != consensus::StateTransferStage::Idle { + self.consensus + .set_state_transfer_stage(consensus::StateTransferStage::Idle); + } + self.purge_recovery = PurgeRecovery::Ready; + Ok(()) + } + + /// Finish a recovered message reset before loading consumer bookmarks. + /// Configure the offset directory paths first. No message data, journal + /// boundary, or fresh offset operation is reset by this recovery step. + /// + /// # Errors + /// Returns the cleanup or completion marker error. Recovery must not serve + /// the partition while files from the old history could still be loaded. + pub async fn resume_purge_cleanup_with_storage( + &mut self, + storage: &S, + ) -> Result<(), IggyError> { + if let Some(reset) = self.purge_reset + && reset.generation > self.applied_purge_generation + { + self.complete_purge_with_storage(storage, reset.generation) + .await + .map_err(|error| match error { + PurgeError::FrontierNotRecorded => IggyError::CannotSyncFile, + PurgeError::OffsetCleanupNotRecorded(source) + | PurgeError::GenerationNotRecorded(source) + | PurgeError::Unserviceable(source) => source, + })?; + } + Ok(()) + } + + /// Keep the reset boundary within the history retained by a state transfer. + /// The caller has refused pending cleanup and any offer below the committed + /// operation, and holds the write lock through replacement of the history. + /// Lowering the durable floor requires an install backup so recovery can + /// restore the old boundary with its history if replacement is interrupted. + pub(crate) async fn rebase_purge_reset_for_install( + &mut self, + commit_op: u64, + ) -> Result<(), IggyError> { + if let Some(reset) = self.purge_reset { + let rebased = PurgeReset { + generation: reset.generation, + floor: reset.floor.min(commit_op), + }; + if rebased != reset + && let Some(directory) = self.partition_dir() + { + persist_purge_reset_with_storage( + &DiskStorage, + &format!("{directory}/{PURGE_RESET_FILE}"), + rebased, + self.created_revision, + ) + .await?; + } + self.purge_reset = Some(rebased); + } + self.purge_floor_op = self.purge_floor_op.min(commit_op); + Ok(()) + } + + pub(crate) fn purge_reset_needs_rebase(&self, commit_op: u64) -> bool { + self.purge_reset + .is_some_and(|reset| reset.floor > commit_op) + } + + pub(crate) fn purge_cleanup_pending(&self) -> bool { + self.purge_reset + .is_some_and(|reset| reset.generation > self.applied_purge_generation) + } + #[must_use] pub const fn consensus(&self) -> &VsrConsensus { &self.consensus @@ -2801,10 +2955,14 @@ where } } + #[allow(clippy::too_many_lines)] async fn persist_consumer_offset_commit( &self, pending: PendingConsumerOffsetCommit, ) -> Result<(), IggyError> { + if self.purge_cleanup_pending() { + return Err(IggyError::TransientNotAccepted); + } // For either offset policy, the WAL protects these updates until checkpoint // syncs their retained writers and directories before reclaiming history. let persisted = @@ -2934,6 +3092,7 @@ where if result.context.history != self.poll_history || self.fatal.is_some() || self.materialization_missing + || self.purge_recovery_pending() { return Err(IggyError::TransientNotAccepted); } @@ -3079,7 +3238,8 @@ where } fn auto_commit_admission_ready(&self, kind: ConsumerKind, consumer_id: u32) -> bool { - self.observed_view == self.consensus.view() + !self.purge_cleanup_pending() + && self.observed_view == self.consensus.view() && !self.offset_reservations_need_resync.get() && (!self.consumer_offset_capacity_for(kind).is_uncertain() || self.durable_consumer_offsets.contains(kind, consumer_id)) @@ -4303,6 +4463,7 @@ where // Reject it first with the only response that proves the request // was never admitted, so the caller may safely retry elsewhere. if self.materialization_missing + || self.purge_recovery_pending() || consensus.is_follower() || !consensus.is_normal() || consensus.is_transferring() @@ -4774,6 +4935,7 @@ where #[must_use] pub fn queued_requests_ready(&self) -> bool { self.fatal.is_none() + && !self.purge_recovery_pending() && self.consensus.is_primary() && self.consensus.is_normal() && !self.consensus.is_transferring() @@ -4794,6 +4956,9 @@ where /// which is unrecoverable in place. #[allow(clippy::future_not_send, clippy::too_many_lines)] pub async fn on_replicate(&mut self, message: Message) { + if self.purge_recovery_pending() { + return; + } self.resynchronize_consumer_offset_reservations(); let header = *message.header(); // Same reason as the metadata plane: `checksum` is compared as an opaque token @@ -5163,7 +5328,7 @@ where #[allow(clippy::future_not_send)] pub async fn on_ack(&mut self, message: Message, config: &PartitionsConfig) { - if self.fatal.is_some() { + if self.fatal.is_some() || self.purge_recovery_pending() { return; } self.resynchronize_consumer_offset_reservations(); @@ -5245,6 +5410,7 @@ where pub async fn commit_journal(&mut self, config: &PartitionsConfig) { if self.fatal.is_some() || self.materialization_missing + || self.purge_recovery_pending() || self.persistence_checkpoint_pending() { return; @@ -5283,11 +5449,14 @@ where } fn drain_persistable_commits(&self, config: &PartitionsConfig) -> Vec { - let Some(persistence) = &self.persistence else { + if self.persistence.is_none() && !self.purge_cleanup_pending() { return drain_committable_prefix(self.consensus()); - }; + } self.persist_repaired_prefix(); - let through = self.consensus.commit_max().min(persistence.head()); + let through = self.persistence.as_ref().map_or_else( + || self.consensus.commit_max(), + |persistence| self.consensus.commit_max().min(persistence.head()), + ); let materialize = self.should_persist_messages(config); let mut drained = Vec::new(); self.consensus.with_pipeline_mut(|pipeline| { @@ -5320,6 +5489,17 @@ where } fn prepare_body_is_ready(&self, header: &PrepareHeader, materialize: bool) -> bool { + // Hold fresh bookmark mutations until the directory sweep has finished. + // Keep them journaled so the commit walk resumes after cleanup succeeds. + if self.purge_cleanup_pending() + && header.op > self.purge_floor_op + && matches!( + header.operation, + Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset + ) + { + return false; + } header.operation != Operation::SendMessages || header.op <= self.purge_floor_op || (!self.durability().is_persisted() && !materialize) @@ -7671,34 +7851,47 @@ where Err(PurgeError::FrontierNotRecorded) } - /// Reset the partition to a single empty segment at offset 0 and clear all - /// consumer / consumer-group offsets (memory + disk). This is the local - /// effect of a committed `PurgeTopic`: it wipes message data and offsets but - /// preserves the partition and its consumer-group membership. Mirrors the - /// legacy server's `purge_all_segments` + offset-file deletion. + /// Reset message history and remove consumer and group bookmarks for a purge. /// - /// Records `generation` as the applied purge generation so the reconciler - /// does not re-wipe a partition already purged at this generation (a later - /// `PurgeTopic` advances the committed generation and triggers a fresh pass). + /// A new generation resets the partition to one empty segment at offset 0, + /// preserving consumer group membership. Once that reset is durable, retries + /// finish bookmark cleanup without deleting fresh messages or moving the floor. + /// Completion records `generation` so redundant reconciler passes do no work. /// /// # Errors - /// [`PurgeError::FrontierNotRecorded`] before anything is mutated, which - /// the caller RETRIES: the reconciler re-issues the purge while - /// `committed > applied`, and fencing a partition that still holds its whole - /// chain would quarantine live data behind a counter that still names the - /// pre-purge offset space. [`PurgeError::Unserviceable`] once the drain has - /// run, which the caller FENCES (quarantine + retire for the reconciler to - /// rebuild), exactly as the state-transfer install's `ConvergeFailed` arm - /// does, or the next append panics on `active_segment()`. - #[allow(clippy::too_many_lines)] + /// [`PurgeError::FrontierNotRecorded`] leaves message history intact and can + /// be retried. Cleanup and completion marker failures retry only those steps. + /// [`PurgeError::Unserviceable`] requires fencing because the message reset + /// lacks a usable segment chain or a durable boundary for preserving new writes. pub async fn purge( &mut self, config: &PartitionsConfig, generation: u64, + ) -> Result<(), PurgeError> { + self.purge_with_storage(config, generation, &DiskStorage) + .await + } + + #[allow(clippy::too_many_lines)] + async fn purge_with_storage( + &mut self, + config: &PartitionsConfig, + generation: u64, + storage: &S, ) -> Result<(), PurgeError> { let write_lock = self.write_lock.clone(); let _guard = write_lock.lock().await; + if generation <= self.applied_purge_generation { + return Ok(()); + } + if self + .purge_reset + .is_some_and(|reset| reset.generation == generation) + { + return self.complete_purge_with_storage(storage, generation).await; + } + let namespace = self.namespace(); if let Some(persistence) = &self.persistence { @@ -7810,11 +8003,9 @@ where self.install_empty_segment(config, start_offset) .await .map_err(PurgeError::Unserviceable)?; - // Make the unlinks AND the replanted dirent durable together: without - // this a crash can resurrect pre-purge segments until the boot re-purge - // fires. Bounded and self-healing, so a failure is logged, not fenced: - // the generation write below has not run yet, so a crash after a failed - // fsync re-purges at boot anyway. + // Make segment removals and the replacement entry durable together. + // The reset marker publication below repeats this directory barrier and + // fences the partition if it fails, before cleanup can be retried alone. if let Some(partition_dir) = self.partition_dir.clone() && let Err(error) = crate::state_transfer::fsync_dir(&partition_dir).await { @@ -7824,8 +8015,8 @@ where namespace_raw = namespace.inner(), generation, %error, - "purge could not fsync the partition dir; a crash before the \ - generation record re-purges at boot" + "purge could not sync the partition directory; reset marker \ + publication must establish durability before cleanup" ); } // The boot-time durable line marks recovered bytes that must not be @@ -7838,44 +8029,42 @@ where self.installed_frontier = None; self.segment_checksum_cache.borrow_mut().clear(); - self.complete_purge_with_storage(&DiskStorage, generation) - .await + self.complete_purge_with_storage(storage, generation).await } - /// Clear consumer progress and record completion after resetting message history. - /// - /// Completion proceeds through three stages: - /// 1. Clear live consumer and group bookmarks, delete their files, and sync - /// each offset directory so the deletions can survive power loss. - /// 2. Reset offset bookkeeping and prevent old journal entries from being - /// applied or served as messages again. - /// 3. Persist the purge generation before advancing the live generation, - /// then invalidate cached state transfer offers and restamp the frontier. + /// Record a message reset, clear consumer bookmarks, and publish completion. /// - /// Message history must already be reset, as [`Self::purge`] does before - /// entering this phase. The caller must exclude concurrent writes throughout. - /// Retry behavior remains in [`Self::purge`], including its deferral and - /// message reset decisions. Storage controls the offset files and completion - /// marker; journal and superblock operations still use the implementations - /// attached to this partition. + /// For a new generation, message storage must already be reset and the caller + /// must exclude concurrent writes. This method fences old resident entries and + /// records the reset boundary before attempting bookmark cleanup. A retry for + /// that generation only removes bookmarks, syncs their directories, and writes + /// `purge.gen`; it preserves fresh messages and the original journal floor. /// - /// Offset deletion and directory sync failures are logged and completion - /// continues. Keeping that decision here makes a storage harness exercise the - /// same failure behavior as the server. Consequently, success does not prove - /// that all bookmark deletions are durable: syncing the generation marker's - /// parent does not sync the separate consumer and group directories. + /// The reset record survives power loss independently of `purge.gen`. Recovery + /// loads it before replay and finishes cleanup before loading bookmarks. /// /// # Errors - /// Returns [`PurgeError::GenerationNotRecorded`] if the completion marker - /// cannot be persisted. Cleanup is not rolled back, the applied generation - /// remains unchanged, and the flag that defers prepare acknowledgements is - /// set. The normal purge path manages that flag when retrying. + /// [`PurgeError::Unserviceable`] requires fencing if the reset record cannot be + /// made durable. [`PurgeError::OffsetCleanupNotRecorded`] and + /// [`PurgeError::GenerationNotRecorded`] leave a durable reset to retry, with + /// prepare acknowledgements deferred until completion succeeds. #[allow(clippy::too_many_lines)] pub async fn complete_purge_with_storage( &mut self, storage: &S, generation: u64, ) -> Result<(), PurgeError> { + if generation <= self.applied_purge_generation { + return Ok(()); + } + if self + .purge_reset + .is_none_or(|reset| reset.generation != generation) + { + self.record_completed_purge_reset(storage, generation) + .await?; + } + self.purge_deferred = true; let namespace = self.namespace(); // Clear consumer + consumer-group offsets (memory + disk). Collect the @@ -7918,13 +8107,15 @@ where self.consumer_offsets_path.as_deref(), ConsumerKind::Consumer, ) - .await; + .await + .map_err(PurgeError::OffsetCleanupNotRecorded)?; let strayed_groups = purge_offset_files( storage, self.consumer_group_offsets_path.as_deref(), ConsumerKind::ConsumerGroup, ) - .await; + .await + .map_err(PurgeError::OffsetCleanupNotRecorded)?; for (kind, consumer_id, path) in consumer_paths .into_iter() .chain(group_paths) @@ -7943,22 +8134,16 @@ where %error, "purge could not remove a consumer offset file" ); - } else { - if let Some(persistence) = &self.persistence { - persistence.retire_offset_file(&path); - } - self.consumer_offset_capacity_for(kind) - .clear_stranded(consumer_id); + return Err(PurgeError::OffsetCleanupNotRecorded(error)); } + if let Some(persistence) = &self.persistence { + persistence.retire_offset_file(&path); + } + self.consumer_offset_capacity_for(kind) + .clear_stranded(consumer_id); } - // Directory fsync so those unlinks stick, mirroring the install path: a - // crash right after the purge otherwise resurrects the offset files at - // boot, and while recovery clamps a resurrected offset down to the - // rebuilt head, "consumed through 0" is not the intended "no entry at - // all" -- that consumer skips the first post-purge message. Logged on - // failure, sharper than the partition-dir fsync above: the generation - // write below still runs, so a crash would resurrect these files with - // no boot re-purge left to clear them. + // Sync each bookmark directory independently. Syncing the reset or + // completion marker's parent cannot make these deletions durable. for dir in self .consumer_offsets_path .clone() @@ -7973,11 +8158,63 @@ where generation, dir = %dir, %error, - "purge could not fsync an offsets dir; a crash may resurrect \ - deleted offset files with the purge already recorded" + "purge bookmark cleanup is not durable; completion remains pending" + ); + if error.kind() != std::io::ErrorKind::NotFound { + return Err(PurgeError::OffsetCleanupNotRecorded( + IggyError::CannotSyncFile, + )); + } + } + } + // The reset record already protects fresh messages if this publication + // fails. Only cleanup and this marker are retried for the same generation. + if let Some(dir) = self.partition_dir() { + let path = format!("{dir}/{PURGE_GENERATION_FILE}"); + if let Err(error) = persist_purge_generation_with_storage( + storage, + &path, + generation, + self.created_revision, + ) + .await + { + self.purge_deferred = true; + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = namespace.inner(), + generation, + %error, + "purge reset the partition but could not record its applied generation; \ + deferring PrepareOk until cleanup completion is recorded" ); + return Err(PurgeError::GenerationNotRecorded(error)); } } + self.applied_purge_generation = generation; + self.purge_deferred = false; + self.transfer_offer_cache.borrow_mut().take(); + // A cleanup retry may already hold fresh messages. Restamp their current + // frontier instead of reinstalling the zero from the original reset. + if !self.reset_offset_frontier().await { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = namespace.inner(), + generation, + "purge could not record the current frontier after cleanup; \ + the existing durable frontier remains in effect" + ); + } + Ok(()) + } + + async fn record_completed_purge_reset( + &mut self, + storage: &S, + generation: u64, + ) -> Result<(), PurgeError> { self.durable_consumer_offsets.clear(); self.pending_consumer_offset_commits.clear(); @@ -8030,56 +8267,23 @@ where .len(); self.evict_committed_prefix(fenced_prefix).await; - // Last durable step: record the applied generation before the - // in-memory marker advances. On a write failure the marker stays old, - // the error propagates, and the reconciler retries the whole purge - // (idempotent, the chain is already empty). The reverse order would - // ack a purge that a crash then silently undoes: restart would - // hydrate the old generation, yet the reconciler believes the purge - // applied. Deferring PrepareOk mirrors the frontier-record failure: - // an op acked now would be wiped by the retry purge while peers that - // recorded the generation keep it. + let reset = PurgeReset { + generation, + floor: self.purge_floor_op, + }; + self.purge_deferred = true; + self.transfer_offer_cache.borrow_mut().take(); if let Some(dir) = self.partition_dir() { - let path = format!("{dir}/{PURGE_GENERATION_FILE}"); - if let Err(error) = persist_purge_generation_with_storage( + persist_purge_reset_with_storage( storage, - &path, - generation, + &format!("{dir}/{PURGE_RESET_FILE}"), + reset, self.created_revision, ) .await - { - self.purge_deferred = true; - warn!( - target: "iggy.partitions.diag", - plane = "partitions", - namespace_raw = namespace.inner(), - generation, - %error, - "purge reset the partition but could not record its applied generation; \ - deferring PrepareOk until the re-issued purge records it" - ); - return Err(PurgeError::GenerationNotRecorded(error)); - } - } - self.applied_purge_generation = generation; - // Same commit frontier, different (now empty) bytes: a cached offer - // built pre-purge would advertise files the purge just unlinked. - self.transfer_offer_cache.borrow_mut().take(); - // The reset itself already landed before the unlinks; this second write - // only re-stamps the record now that the view-scoped fields and the - // counter agree with it. A failure leaves the pre-unlink 0 on disk, - // which is the safe direction, so it is logged rather than refused. - if !self.reset_offset_frontier().await { - warn!( - target: "iggy.partitions.diag", - plane = "partitions", - namespace_raw = namespace.inner(), - generation, - "purge could not re-stamp the superblock after resetting the partition; \ - the frontier reset written before the unlinks still stands" - ); + .map_err(PurgeError::Unserviceable)?; } + self.purge_reset = Some(reset); Ok(()) } @@ -8098,7 +8302,7 @@ where /// op is already committed cluster-wide; there is nobody to ack to). The /// commit walk runs at `RepairDone`, after the floor is known. pub async fn apply_repaired_prepare(&mut self, message: Message) { - if self.materialization_missing { + if self.materialization_missing || self.purge_recovery_pending() { return; } let header = *message.header(); @@ -8449,7 +8653,7 @@ where } async fn send_prepare_ok(&self, header: &PrepareHeader) -> bool { - if self.fatal.is_some() || self.materialization_missing { + if self.fatal.is_some() || self.materialization_missing || self.purge_recovery_pending() { return false; } // Durable-before-send: a PrepareOk implies this replica's @@ -8494,9 +8698,9 @@ async fn purge_offset_files( storage: &S, directory: Option<&str>, kind: ConsumerKind, -) -> Vec<(ConsumerKind, u32, String)> { +) -> Result, IggyError> { let Some(directory) = directory else { - return Vec::new(); + return Ok(Vec::new()); }; let entries = futures::stream::once(storage.regular_files(Path::new(directory))).try_flatten(); futures::pin_mut!(entries); @@ -8512,7 +8716,10 @@ async fn purge_offset_files( %error, "failed to scan consumer offset directory during purge" ); - continue; + if error.kind() == std::io::ErrorKind::NotFound { + continue; + } + return Err(IggyError::CannotReadFile); } }; let Some(path) = path.to_str() else { @@ -8522,7 +8729,7 @@ async fn purge_offset_files( offsets.push((kind, consumer_id, path.to_owned())); } } - offsets + Ok(offsets) } /// Automatic commits remain monotone because an earlier poll can commit after @@ -9435,11 +9642,15 @@ mod tests { } pub(super) fn test_partition() -> IggyPartition { + test_partition_with_replicas(1) + } + + pub(super) fn test_partition_with_replicas(replica_count: u8) -> IggyPartition { let namespace = IggyNamespace::new(1, 1, 0); let consensus = VsrConsensus::new( TEST_CLUSTER, 0, - 1, + replica_count, namespace.inner(), IggyMessageBus::new(0), LocalPipeline::new(), @@ -9456,9 +9667,16 @@ mod tests { /// Keep the returned directory alive until every read using those files ends. pub(super) async fn disk_poll_partition( config: &PartitionsConfig, + ) -> (tempfile::TempDir, IggyPartition) { + Box::pin(disk_poll_partition_with_replicas(config, 1)).await + } + + pub(super) async fn disk_poll_partition_with_replicas( + config: &PartitionsConfig, + replica_count: u8, ) -> (tempfile::TempDir, IggyPartition) { let directory = tempfile::tempdir().expect("create partition directory"); - let mut partition = test_partition(); + let mut partition = test_partition_with_replicas(replica_count); partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); partition.log.retire_front().expect("retire empty segment"); partition @@ -11361,7 +11579,7 @@ mod tests { }) } - fn store_offset_request( + pub(super) fn store_offset_request( client_id: u128, request_id: u64, kind: ConsumerKind, @@ -17874,3 +18092,6 @@ mod purge_floor_tests { let _ = std::fs::remove_dir_all(&dir); } } + +#[cfg(test)] +mod purge_retry_tests; diff --git a/core/partitions/src/iggy_partition/purge_retry_tests.rs b/core/partitions/src/iggy_partition/purge_retry_tests.rs new file mode 100644 index 0000000000..630c0a0444 --- /dev/null +++ b/core/partitions/src/iggy_partition/purge_retry_tests.rs @@ -0,0 +1,870 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Retry full purge after cleanup fails, preserving messages committed meanwhile. +//! +//! The fixture commits replicated operations directly: a replica withholding its +//! own `PrepareOk` can still learn commits acknowledged by the other replicas. +//! These tests exercise the resulting local state, not client acknowledgment. + +use std::cell::Cell; +use std::io; +use std::mem::size_of; +use std::ops::RangeInclusive; +use std::path::{Path, PathBuf}; + +use consensus::{Consensus, PipelineEntry, Sequencer, oneshot_channel}; +use iggy_binary_protocol::primitives::consumer::WireConsumer; +use iggy_binary_protocol::requests::consumer_offsets::DeleteConsumerOffsetRequest; +use iggy_binary_protocol::{ + AckLevel, Command, Operation, PrepareHeader, WireEncode, WireIdentifier, +}; +use iggy_common::{ + ConsumerGroupId, ConsumerKind, ConsumerOffset, Durability, IggyError, PollingStrategy, +}; +use journal::durable_storage::{DiskStorage, DurableStorage, OpenMode, RegularFiles, StorageEntry}; +use message_bus::IggyMessageBus; +use server_common::Message; +use server_common::send_messages::decode_batch_slice; + +use super::tests::{ + checksummed_segment_prepare, disk_poll_partition, disk_poll_partition_with_replicas, + journal_store_offset, repair_config, store_offset_request, test_partition, + test_partition_with_replicas, +}; +use super::{IggyPartition, PurgeError}; +use crate::offset_storage::{PURGE_GENERATION_FILE, PURGE_RESET_FILE, persist_offset}; +use crate::{Partition, PollingArgs, PollingConsumer}; + +const OLD_LAST_OPERATION: u64 = 3; +const FRESH_LAST_OPERATION: u64 = 8; +const PURGE_GENERATION: u64 = 1; +const CONSUMER_ID: usize = 7; +const GROUP_ID: usize = 9; +const OLD_PAYLOAD: &[u8] = b"before purge"; +const FRESH_PAYLOAD: &[u8] = b"fresh after purge"; + +#[compio::test] +async fn given_replicated_consumer_sync_failure_when_retrying_purge_should_preserve_fresh_messages() +{ + Box::pin(assert_retry_preserves_fresh_messages( + Durability::Replicated, + PurgeFault::OffsetDirectorySync(ConsumerKind::Consumer), + )) + .await; +} + +#[compio::test] +async fn given_replicated_group_sync_failure_when_retrying_purge_should_preserve_fresh_messages() { + Box::pin(assert_retry_preserves_fresh_messages( + Durability::Replicated, + PurgeFault::OffsetDirectorySync(ConsumerKind::ConsumerGroup), + )) + .await; +} + +#[compio::test] +async fn given_persisted_consumer_sync_failure_when_retrying_purge_should_preserve_fresh_messages() +{ + Box::pin(assert_retry_preserves_fresh_messages( + Durability::Persisted, + PurgeFault::OffsetDirectorySync(ConsumerKind::Consumer), + )) + .await; +} + +#[compio::test] +async fn given_persisted_group_sync_failure_when_retrying_purge_should_preserve_fresh_messages() { + Box::pin(assert_retry_preserves_fresh_messages( + Durability::Persisted, + PurgeFault::OffsetDirectorySync(ConsumerKind::ConsumerGroup), + )) + .await; +} + +#[compio::test] +async fn given_replicated_marker_failure_when_retrying_purge_should_preserve_fresh_messages() { + Box::pin(assert_retry_preserves_fresh_messages( + Durability::Replicated, + PurgeFault::CompletionMarkerRename, + )) + .await; +} + +#[compio::test] +async fn given_persisted_marker_failure_when_retrying_purge_should_preserve_fresh_messages() { + Box::pin(assert_retry_preserves_fresh_messages( + Durability::Persisted, + PurgeFault::CompletionMarkerRename, + )) + .await; +} + +#[compio::test] +async fn given_fresh_history_after_purge_when_new_generation_arrives_should_reset_again() { + let config = repair_config(); + let (_directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + partition.purge(&config, PURGE_GENERATION).await.unwrap(); + + append_committed_history( + &mut partition, + OLD_LAST_OPERATION + 1..=FRESH_LAST_OPERATION, + FRESH_PAYLOAD, + ) + .await; + assert_eq!(partition.stats.messages_count_inconsistent(), 5); + + partition + .purge(&config, PURGE_GENERATION + 1) + .await + .unwrap(); + + assert_eq!(partition.applied_purge_generation(), PURGE_GENERATION + 1); + assert_eq!(partition.purge_floor_op(), FRESH_LAST_OPERATION); + assert_eq!(partition.offset_frontier(), 0); + assert_eq!(partition.mint_frontier(), 0); + assert_eq!(partition.stats.messages_count_inconsistent(), 0); + assert_eq!(partition.log.active_segment().size.as_bytes_u64(), 0); + assert_next_messages(&mut partition, &[], FRESH_PAYLOAD).await; +} + +#[compio::test] +async fn given_pending_purge_cleanup_when_offset_commits_arrive_should_apply_them_after_retry() { + for policy in [Durability::Replicated, Durability::Persisted] { + Box::pin(assert_offset_commits_wait_for_cleanup( + policy, + CommitPath::Journal, + )) + .await; + } +} + +#[compio::test] +async fn given_pending_purge_cleanup_when_primary_has_offset_commits_should_keep_them_in_pipeline() +{ + for policy in [Durability::Replicated, Durability::Persisted] { + Box::pin(assert_offset_commits_wait_for_cleanup( + policy, + CommitPath::PrimaryPipeline, + )) + .await; + } +} + +#[compio::test] +async fn given_pending_purge_cleanup_when_singleton_stores_without_ack_should_refuse_until_retry() { + let config = repair_config(); + let (directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; + assert_eq!(partition.consensus().replica_count(), 1); + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + persist_old_bookmarks(&partition).await; + let storage = FaultingPurgeStorage::new( + directory.path(), + PurgeFault::OffsetDirectorySync(ConsumerKind::Consumer), + ); + assert!(matches!( + partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await, + Err(PurgeError::OffsetCleanupNotRecorded(_)) + )); + append_committed_history( + &mut partition, + OLD_LAST_OPERATION + 1..=FRESH_LAST_OPERATION, + FRESH_PAYLOAD, + ) + .await; + + let consumer_id = u32::try_from(CONSUMER_ID).unwrap(); + let consumer = PollingConsumer::Consumer(CONSUMER_ID, 0); + let (denial_sender, denial_receiver) = oneshot_channel(); + partition + .on_request( + store_offset_request( + 42, + 1, + ConsumerKind::Consumer, + consumer_id, + 2, + AckLevel::NoAck, + ), + Some(denial_sender), + ) + .await; + assert_eq!( + denial_receiver.await.unwrap().header().status, + IggyError::TransientNotAccepted.as_code(), + "the singleton fast path must not write a bookmark that cleanup would erase" + ); + assert_eq!(partition.get_consumer_offset(consumer), None); + assert_eq!(partition.consensus().pipeline_len(), 0); + assert!(!directory.path().join("consumer_offsets/7").exists()); + assert!(partition.fatal().is_none()); + + partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await + .unwrap(); + let (success_sender, success_receiver) = oneshot_channel(); + partition + .on_request( + store_offset_request( + 42, + 2, + ConsumerKind::Consumer, + consumer_id, + 2, + AckLevel::NoAck, + ), + Some(success_sender), + ) + .await; + assert_eq!(success_receiver.await.unwrap().header().status, 0); + assert_eq!(partition.get_consumer_offset(consumer), Some(2)); + assert_eq!(partition.consensus().pipeline_len(), 0); + assert!(directory.path().join("consumer_offsets/7").exists()); +} + +#[compio::test] +async fn given_singleton_purge_marker_when_consensus_restarts_should_accept_new_operation_one() { + let config = repair_config(); + let (directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + partition.purge(&config, PURGE_GENERATION).await.unwrap(); + assert_eq!(partition.purge_floor_op(), OLD_LAST_OPERATION); + drop(partition); + + // Singleton boot starts a new consensus sequence. Reuse only its durable + // purge markers here; this fixture does not exercise the full boot loader. + let mut restarted = Box::new(test_partition()); + restarted.set_partition_dir(directory.path().to_string_lossy().into_owned()); + restarted.log.retire_front().unwrap(); + restarted.install_empty_segment(&config, 0).await.unwrap(); + restarted.hydrate_applied_purge_generation().await.unwrap(); + assert_eq!(restarted.applied_purge_generation(), PURGE_GENERATION); + assert_eq!(restarted.consensus().sequencer().current_sequence(), 0); + assert_eq!( + restarted.purge_floor_op(), + 0, + "an earlier process's purge floor must not fence the new consensus sequence" + ); + + append_committed_history(&mut restarted, 1..=1, FRESH_PAYLOAD).await; + + assert_eq!(restarted.consensus().commit_min(), 1); + assert_eq!(restarted.stats.messages_count_inconsistent(), 1); + assert_next_messages(&mut restarted, &[0], FRESH_PAYLOAD).await; +} + +#[compio::test] +async fn given_replicated_purge_when_empty_history_recovery_retries_should_poll_new_operation_one() +{ + for fault in [ + PurgeFault::ResetMarkerRename, + PurgeFault::PartitionDirectorySync, + ] { + let config = repair_config(); + let (directory, mut partition) = + Box::pin(disk_poll_partition_with_replicas(&config, 3)).await; + partition.runtime_options.durability = Durability::Replicated; + partition.runtime_options.consumer_offset_durability = Durability::Replicated; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + partition.purge(&config, PURGE_GENERATION).await.unwrap(); + drop(partition); + + let mut recovering = Box::pin(recover_replicated_partition(directory.path())).await; + assert!(recovering.persistence.is_none()); + assert_eq!(recovering.consensus().sequencer().current_sequence(), 0); + assert_eq!(recovering.purge_floor_op(), OLD_LAST_OPERATION); + assert!(recovering.purge_recovery_pending()); + assert!(!recovering.queued_requests_ready()); + + let storage = FaultingPurgeStorage::new(directory.path(), fault); + assert!( + recovering + .recover_purge_boundary_with_storage(&storage, 0) + .await + .is_err() + ); + assert!(storage.failed.get()); + assert!(recovering.purge_recovery_pending()); + assert_eq!(recovering.purge_floor_op(), OLD_LAST_OPERATION); + drop(recovering); + + // Lose all memory again, including the failed recovery attempt. The + // directory sync error can leave the replacement record visible to boot. + let mut restarted = Box::pin(recover_replicated_partition(directory.path())).await; + assert!(restarted.purge_recovery_pending()); + assert_eq!(restarted.applied_purge_generation(), PURGE_GENERATION); + restarted.recover_purge_boundary(0).await.unwrap(); + assert!(!restarted.purge_recovery_pending()); + assert_eq!(restarted.purge_floor_op(), 0); + assert_eq!(restarted.purge_reset.unwrap().floor, 0); + + append_committed_history(&mut restarted, 1..=1, FRESH_PAYLOAD).await; + assert_eq!(restarted.consensus().commit_min(), 1); + assert_eq!(restarted.stats.messages_count_inconsistent(), 1); + assert_next_messages(&mut restarted, &[0], FRESH_PAYLOAD).await; + + // A repeated recovery notification must not clear the newly committed tail. + restarted.recover_purge_boundary(0).await.unwrap(); + assert_next_messages(&mut restarted, &[0], FRESH_PAYLOAD).await; + } +} + +#[compio::test] +async fn given_replicated_purge_when_selected_history_is_nonempty_should_wait_for_snapshot() { + let config = repair_config(); + let (directory, mut partition) = Box::pin(disk_poll_partition_with_replicas(&config, 3)).await; + partition.runtime_options.durability = Durability::Replicated; + partition.runtime_options.consumer_offset_durability = Durability::Replicated; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + partition.purge(&config, PURGE_GENERATION).await.unwrap(); + drop(partition); + + let mut restarted = Box::pin(recover_replicated_partition(directory.path())).await; + let poll = restarted + .build_poll_plan( + PollingConsumer::Consumer(CONSUMER_ID, 0), + &PollingArgs::new(PollingStrategy::next(), 1, false), + true, + ) + .execute() + .await; + assert!(matches!( + restarted.complete_poll(poll), + Err(IggyError::TransientNotAccepted) + )); + + // The peer could hold either the old sequence or a new sequence reusing its + // numbers. Journal replay alone cannot tell which messages the purge removed. + restarted + .recover_purge_boundary(OLD_LAST_OPERATION) + .await + .unwrap(); + assert!(restarted.purge_recovery_pending()); + assert!(restarted.consensus().is_transferring()); + assert_eq!(restarted.purge_floor_op(), OLD_LAST_OPERATION); + restarted + .apply_repaired_prepare(checksummed_segment_prepare(1, 0, 0, OLD_PAYLOAD)) + .await; + assert!(restarted.log.journal().inner.header_by_op(1).is_none()); + assert_eq!(restarted.consensus().commit_min(), 0); +} + +// Rebuild only the partition and its empty segment after purge. Server boot and +// consensus selection are exercised separately; no volatile journal is retained. +async fn recover_replicated_partition(directory: &Path) -> Box> { + let mut partition = Box::new(test_partition_with_replicas(3)); + partition.runtime_options.durability = Durability::Replicated; + partition.runtime_options.consumer_offset_durability = Durability::Replicated; + partition.set_partition_dir(directory.to_string_lossy().into_owned()); + partition.log.retire_front().unwrap(); + partition + .install_empty_segment(&repair_config(), 0) + .await + .unwrap(); + partition.hydrate_applied_purge_generation().await.unwrap(); + partition +} + +#[compio::test] +async fn given_reset_marker_publication_failure_when_purging_should_require_fencing() { + let config = repair_config(); + let (directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + let storage = FaultingPurgeStorage::new(directory.path(), PurgeFault::ResetMarkerRename); + + let result = partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await; + + assert!( + storage.failed.get(), + "purge must reach reset marker publication" + ); + assert!(matches!(result, Err(PurgeError::Unserviceable(_)))); + assert_eq!(partition.applied_purge_generation(), 0); + assert!(partition.purge_deferred); + assert!(!directory.path().join(PURGE_RESET_FILE).exists()); + let mut reloaded = Box::new(test_partition()); + reloaded.set_partition_dir(directory.path().to_string_lossy().into_owned()); + reloaded.hydrate_applied_purge_generation().await.unwrap(); + assert_eq!(reloaded.applied_purge_generation(), 0); +} + +// Keep refusal, cleanup, and the deferred commits together so their ordering stays visible. +#[allow(clippy::too_many_lines)] +async fn assert_offset_commits_wait_for_cleanup(policy: Durability, commit_path: CommitPath) { + let config = repair_config(); + let (directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; + partition.runtime_options.consumer_offset_durability = policy; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + persist_old_bookmarks(&partition).await; + let storage = FaultingPurgeStorage::new( + directory.path(), + PurgeFault::OffsetDirectorySync(ConsumerKind::Consumer), + ); + assert!(matches!( + partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await, + Err(PurgeError::OffsetCleanupNotRecorded(_)) + )); + append_committed_history( + &mut partition, + OLD_LAST_OPERATION + 1..=FRESH_LAST_OPERATION, + FRESH_PAYLOAD, + ) + .await; + + // Fresh progress must not land in a directory that cleanup still has to sweep. + let consumer = PollingConsumer::Consumer(CONSUMER_ID, 0); + let pending_poll = partition + .build_poll_plan( + consumer, + &PollingArgs::new(PollingStrategy::next(), 5, true), + true, + ) + .execute() + .await; + assert!(matches!( + partition.complete_poll(pending_poll), + Err(IggyError::TransientNotAccepted) + )); + assert_eq!(partition.get_consumer_offset(consumer), None); + + let store_operation = FRESH_LAST_OPERATION + 1; + let delete_operation = store_operation + 1; + let consumer_id = u32::try_from(CONSUMER_ID).unwrap(); + journal_store_offset(&mut partition, store_operation, consumer_id, 2).await; + journal_consumer_offset_delete(&mut partition, delete_operation, consumer_id).await; + if matches!(commit_path, CommitPath::PrimaryPipeline) { + assert!(partition.consensus().is_primary()); + assert!( + partition.persistence.is_none(), + "exercise the pipeline path without a WAL" + ); + for operation in [store_operation, delete_operation] { + let header = partition + .log + .journal() + .inner + .header_by_op(operation) + .unwrap(); + partition.consensus().with_pipeline_mut(|pipeline| { + pipeline.push(PipelineEntry::new(header)); + }); + } + } + partition.consensus().advance_commit_max(store_operation); + partition.commit_journal(&config).await; + assert_eq!(partition.consensus().commit_min(), FRESH_LAST_OPERATION); + if matches!(commit_path, CommitPath::PrimaryPipeline) { + assert_eq!( + partition.consensus().pipeline_len(), + 2, + "cleanup must retain the primary's reply slots" + ); + } + assert_eq!(partition.get_consumer_offset(consumer), None); + assert!( + partition + .pending_consumer_offset_commits + .contains_key(&store_operation) + ); + assert!( + partition + .pending_consumer_offset_commits + .contains_key(&delete_operation) + ); + assert!(partition.fatal().is_none()); + + partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await + .expect("finish cleanup before applying the waiting offset operations"); + assert!( + partition + .pending_consumer_offset_commits + .contains_key(&store_operation) + ); + assert!( + partition + .pending_consumer_offset_commits + .contains_key(&delete_operation) + ); + partition.commit_journal(&config).await; + assert_eq!(partition.consensus().commit_min(), store_operation); + assert_eq!(partition.get_consumer_offset(consumer), Some(2)); + + partition.consensus().advance_commit_max(delete_operation); + partition.commit_journal(&config).await; + assert_eq!(partition.consensus().commit_min(), delete_operation); + assert_eq!(partition.consensus().pipeline_len(), 0); + assert_eq!(partition.get_consumer_offset(consumer), None); + assert_next_messages(&mut partition, &[0, 1, 2, 3, 4], FRESH_PAYLOAD).await; + + let fresh_poll = partition + .build_poll_plan( + consumer, + &PollingArgs::new(PollingStrategy::next(), 5, true), + true, + ) + .execute() + .await; + partition + .complete_poll(fresh_poll) + .expect("automatic progress resumes after cleanup"); + assert_eq!(partition.get_consumer_offset(consumer), Some(4)); +} + +async fn assert_retry_preserves_fresh_messages(policy: Durability, fault: PurgeFault) { + let config = repair_config(); + let (directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; + partition.runtime_options.consumer_offset_durability = policy; + append_committed_history(&mut partition, 1..=OLD_LAST_OPERATION, OLD_PAYLOAD).await; + persist_old_bookmarks(&partition).await; + assert_eq!(partition.stats.messages_count_inconsistent(), 3); + + let storage = FaultingPurgeStorage::new(directory.path(), fault); + let result = partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await; + assert!( + storage.failed.get(), + "purge must reach the selected I/O fault" + ); + match fault { + PurgeFault::OffsetDirectorySync(_) => { + assert!(matches!( + result, + Err(PurgeError::OffsetCleanupNotRecorded(_)) + )); + assert_eq!( + storage.entries_at_failed_sync.get(), + Some(0), + "the injected sync failure must follow bookmark deletion" + ); + } + PurgeFault::CompletionMarkerRename => { + assert!(matches!(result, Err(PurgeError::GenerationNotRecorded(_)))); + } + PurgeFault::ResetMarkerRename | PurgeFault::PartitionDirectorySync => { + unreachable!("reset marker failure requires fencing") + } + } + assert_eq!(partition.applied_purge_generation(), 0); + assert!(partition.purge_deferred); + assert_eq!(partition.purge_floor_op(), OLD_LAST_OPERATION); + assert_eq!(partition.stats.messages_count_inconsistent(), 0); + + // Other replicas can commit fresh sends while this replica defers PrepareOk. + // Apply those commits locally before the same purge generation is retried. + append_committed_history( + &mut partition, + OLD_LAST_OPERATION + 1..=FRESH_LAST_OPERATION, + FRESH_PAYLOAD, + ) + .await; + assert_eq!(partition.consensus().commit_min(), FRESH_LAST_OPERATION); + assert_eq!(partition.offset_frontier(), 5); + assert_eq!(partition.mint_frontier(), 5); + assert_eq!(partition.stats.messages_count_inconsistent(), 5); + let fresh_bytes = partition.stats.size_bytes_inconsistent(); + let log_path = directory.path().join("00000000000000000000.log"); + let fresh_log = std::fs::read(&log_path).unwrap(); + assert!(!fresh_log.is_empty()); + + partition + .purge_with_storage(&config, PURGE_GENERATION, &storage) + .await + .expect("retry cleanup without resetting the fresh history"); + + assert_eq!(partition.applied_purge_generation(), PURGE_GENERATION); + assert!(!partition.purge_deferred); + if matches!(fault, PurgeFault::OffsetDirectorySync(_)) { + assert_eq!( + storage.directory_sync_attempts.get(), + 2, + "retry must sync the directory even though its files are already gone" + ); + } + for directory_name in ["consumer_offsets", "consumer_group_offsets"] { + assert!( + DiskStorage + .entries(&directory.path().join(directory_name)) + .await + .unwrap() + .is_empty() + ); + } + assert_fresh_history_unchanged(&partition, fresh_bytes, &log_path, &fresh_log); + assert_next_messages(&mut partition, &[0, 1, 2, 3, 4], FRESH_PAYLOAD).await; + + // Redundant delivery after completion must preserve the same history too. + partition.purge(&config, PURGE_GENERATION).await.unwrap(); + assert_fresh_history_unchanged(&partition, fresh_bytes, &log_path, &fresh_log); + assert_next_messages(&mut partition, &[0, 1, 2, 3, 4], FRESH_PAYLOAD).await; +} + +fn assert_fresh_history_unchanged( + partition: &IggyPartition, + fresh_bytes: u64, + log_path: &Path, + fresh_log: &[u8], +) { + assert_eq!( + partition.purge_floor_op(), + OLD_LAST_OPERATION, + "retry must not fence the fresh committed operations" + ); + assert_eq!(partition.consensus().commit_min(), FRESH_LAST_OPERATION); + assert_eq!(partition.offset_frontier(), 5); + assert_eq!(partition.mint_frontier(), 5); + assert_eq!(partition.stats.messages_count_inconsistent(), 5); + assert_eq!(partition.stats.size_bytes_inconsistent(), fresh_bytes); + assert_eq!(std::fs::read(log_path).unwrap(), fresh_log); +} + +/// Poll both consumer kinds before checking results, with automatic commits disabled. +async fn assert_next_messages( + partition: &mut IggyPartition, + expected_offsets: &[u64], + expected_payload: &[u8], +) { + let mut completed_polls = Vec::new(); + for consumer in [ + PollingConsumer::Consumer(CONSUMER_ID, 0), + PollingConsumer::ConsumerGroup(GROUP_ID, 1), + ] { + let result = partition + .build_poll_plan( + consumer, + &PollingArgs::new(PollingStrategy::next(), 10, false), + true, + ) + .execute() + .await; + let completion = partition.complete_poll(result).expect("complete Next poll"); + completed_polls.push((consumer, completion)); + } + for (consumer, completion) in completed_polls { + let mut offsets = Vec::new(); + for fragment in &completion.fragments { + let batch = decode_batch_slice(fragment.as_slice()).unwrap(); + assert_eq!(batch.message_count(), 1); + assert_eq!(batch.iter().next().unwrap().payload, expected_payload); + offsets.push(batch.header.base_offset); + } + assert_eq!(offsets, expected_offsets, "{consumer:?}"); + assert_eq!(partition.get_consumer_offset(consumer), None); + } +} + +/// Apply replicated sends and their learned commit frontier through production paths. +async fn append_committed_history( + partition: &mut IggyPartition, + operations: RangeInclusive, + payload: &[u8], +) { + let last_operation = *operations.end(); + for operation in operations { + let prepare = checksummed_segment_prepare(operation, 0, 0, payload); + partition.apply_replicated_operation(prepare).await.unwrap(); + partition.consensus().sequencer().set_sequence(operation); + } + partition.consensus().advance_commit_max(last_operation); + partition.commit_journal(&repair_config()).await; + assert!(partition.fatal().is_none()); +} + +async fn journal_consumer_offset_delete( + partition: &mut IggyPartition, + operation: u64, + consumer_id: u32, +) { + let body = DeleteConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::Numeric(consumer_id)), + stream_id: WireIdentifier::Numeric(1), + topic_id: WireIdentifier::Numeric(1), + partition_id: Some(0), + ack: AckLevel::Quorum, + } + .to_bytes(); + let message_size = size_of::() + body.len(); + let mut prepare = Message::::new(message_size); + prepare.as_mut_slice()[size_of::()..].copy_from_slice(&body); + let prepare = prepare.transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command::Prepare; + header.operation = Operation::DeleteConsumerOffset; + header.op = operation; + header.group = partition.namespace().inner(); + header.size = u32::try_from(message_size).unwrap(); + }); + partition.apply_replicated_operation(prepare).await.unwrap(); + partition.consensus().sequencer().set_sequence(operation); +} + +/// Bookmark 2 remains a valid offset in the five messages written after the failure. +async fn persist_old_bookmarks(partition: &IggyPartition) { + for (kind, consumer_id) in [ + (ConsumerKind::Consumer, CONSUMER_ID), + (ConsumerKind::ConsumerGroup, GROUP_ID), + ] { + let numeric_id = u32::try_from(consumer_id).unwrap(); + let path = partition.persisted_offset_path(kind, numeric_id).unwrap(); + persist_offset(&path, 2, true).await.unwrap(); + let bookmark = ConsumerOffset::new(kind, numeric_id, 2, path); + match kind { + ConsumerKind::Consumer => { + partition + .consumer_offsets + .pin() + .insert(consumer_id, bookmark); + } + ConsumerKind::ConsumerGroup => { + partition + .consumer_group_offsets + .pin() + .insert(ConsumerGroupId(consumer_id), bookmark); + } + } + partition.seed_recovered_consumer_offset(kind, numeric_id, 2, 2); + } +} + +#[derive(Clone, Copy)] +enum CommitPath { + Journal, + PrimaryPipeline, +} + +#[derive(Clone, Copy)] +enum PurgeFault { + OffsetDirectorySync(ConsumerKind), + CompletionMarkerRename, + ResetMarkerRename, + PartitionDirectorySync, +} + +struct FaultingPurgeStorage { + fault: PurgeFault, + target: PathBuf, + failed: Cell, + entries_at_failed_sync: Cell>, + directory_sync_attempts: Cell, +} + +impl FaultingPurgeStorage { + fn new(directory: &Path, fault: PurgeFault) -> Self { + let target = directory.join(match fault { + PurgeFault::OffsetDirectorySync(ConsumerKind::Consumer) => "consumer_offsets", + PurgeFault::OffsetDirectorySync(ConsumerKind::ConsumerGroup) => { + "consumer_group_offsets" + } + PurgeFault::CompletionMarkerRename => PURGE_GENERATION_FILE, + PurgeFault::ResetMarkerRename => PURGE_RESET_FILE, + PurgeFault::PartitionDirectorySync => "", + }); + Self { + fault, + target, + failed: Cell::new(false), + entries_at_failed_sync: Cell::new(None), + directory_sync_attempts: Cell::new(0), + } + } +} + +impl DurableStorage for FaultingPurgeStorage { + type File = ::File; + + fn writer_identity(&self, path: &Path) -> io::Result> { + DiskStorage.writer_identity(path) + } + + async fn open(&self, path: &Path, mode: OpenMode) -> io::Result { + DiskStorage.open(path, mode).await + } + + async fn create_directories(&self, path: &Path) -> io::Result<()> { + DiskStorage.create_directories(path).await + } + + async fn sync_directory(&self, path: &Path) -> io::Result<()> { + if matches!(self.fault, PurgeFault::PartitionDirectorySync) + && path == self.target + && !self.failed.replace(true) + { + return Err(io::Error::other( + "injected partition directory sync failure", + )); + } + if matches!(self.fault, PurgeFault::OffsetDirectorySync(_)) && path == self.target { + self.directory_sync_attempts + .set(self.directory_sync_attempts.get() + 1); + if !self.failed.replace(true) { + self.entries_at_failed_sync + .set(Some(DiskStorage.entries(path).await?.len())); + return Err(io::Error::other("injected offset directory sync failure")); + } + } + DiskStorage.sync_directory(path).await + } + + async fn rename(&self, source: &Path, target: &Path) -> io::Result<()> { + if matches!( + self.fault, + PurgeFault::CompletionMarkerRename | PurgeFault::ResetMarkerRename + ) && target == self.target + && !self.failed.replace(true) + { + return Err(io::Error::other("injected purge marker rename failure")); + } + DiskStorage.rename(source, target).await + } + + async fn remove_file(&self, path: &Path) -> io::Result<()> { + DiskStorage.remove_file(path).await + } + + async fn hard_link(&self, source: &Path, target: &Path) -> io::Result<()> { + DiskStorage.hard_link(source, target).await + } + + async fn exists(&self, path: &Path) -> io::Result { + DiskStorage.exists(path).await + } + + async fn exists_following_links(&self, path: &Path) -> io::Result { + DiskStorage.exists_following_links(path).await + } + + async fn entries(&self, path: &Path) -> io::Result> { + DiskStorage.entries(path).await + } + + async fn regular_files(&self, path: &Path) -> io::Result { + DiskStorage.regular_files(path).await + } + + async fn remove_tree(&self, path: &Path) -> io::Result<()> { + DiskStorage.remove_tree(path).await + } +} diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 545377cd45..e854b6d652 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -328,6 +328,8 @@ impl PartitionJournal { unsafe { &mut *self.evicted_ring.get() }.clear(); self.evicted_ring_bytes.set(0); self.resident_control_ops.set(0); + // The replacement history may reuse operation numbers below the old seal. + self.poll_floor.set(0); } /// Disable repair retention (single-replica groups: nobody to repair). diff --git a/core/partitions/src/offset_storage.rs b/core/partitions/src/offset_storage.rs index 266f5a280a..9d48a5ab9e 100644 --- a/core/partitions/src/offset_storage.rs +++ b/core/partitions/src/offset_storage.rs @@ -15,11 +15,12 @@ // specific language governing permissions and limitations // under the License. -//! Store consumer bookmarks and the partition's applied purge generation. +//! Store consumer bookmarks and the partition's purge progress. //! -//! A bookmark records the last consumed offset. The purge marker instead records -//! which reset was applied to this incarnation of the partition. Recovery uses -//! them to restore progress and decide whether a purge must be repeated. +//! A bookmark records the last consumed offset. The reset marker records the +//! durable message reset boundary, while the applied generation records completion +//! of the remaining bookmark cleanup. Recovery uses these records to resume cleanup +//! without deleting messages acknowledged after the reset. //! //! Functions ending in `_with_storage` share the persistence sequence between //! real disk and simulated storage. File sync makes record contents durable; @@ -52,12 +53,30 @@ pub const OFFSET_RECORD_SIZE: usize = OFFSET_SIZE + CHECKSUM_SIZE; /// partition incarnation it was applied for. pub const PURGE_GENERATION_FILE: &str = "purge.gen"; +/// File retaining the durable message reset boundary independently of cleanup. +pub(crate) const PURGE_RESET_FILE: &str = "purge.reset"; + /// Sibling name an atomic offset replacement writes before its rename lands. const OFFSET_REPLACEMENT_SUFFIX: &str = ".tmp"; /// `[generation][created_revision]`, both LE u64. const PURGE_GENERATION_RECORD_SIZE: usize = 2 * OFFSET_SIZE; +/// `[generation][created_revision][floor][checksum]`, all LE u64. +const PURGE_RESET_RECORD_SIZE: usize = 3 * OFFSET_SIZE + CHECKSUM_SIZE; + +/// A durable message reset whose bookmark cleanup may still be pending. +/// +/// The marker remains after cleanup completes so recovery retains the reset floor. +/// The next successful message reset replaces it; `purge.gen` separately records +/// which purge has completed both reset and cleanup. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct PurgeReset { + pub generation: u64, + /// Highest journal operation invalidated by the reset, inclusive. + pub floor: u64, +} + /// What a consumer-offset file was found to hold. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum OffsetRecord { @@ -419,6 +438,75 @@ pub async fn persist_purge_generation_with_storage( replace_file(storage, path, record, true, true).await } +/// Durably record the message reset boundary before completing bookmark cleanup. +/// +/// The caller must first make the message reset durable. Atomic replacement syncs +/// both the record and its parent directory. Calls for one path must be serialized +/// because they share a temporary filename. +/// +/// # Errors +/// Returns an error if directory creation, replacement, or either sync fails. +pub(crate) async fn persist_purge_reset_with_storage( + storage: &S, + path: &str, + reset: PurgeReset, + created_revision: u64, +) -> Result<(), IggyError> { + let mut record = [0u8; PURGE_RESET_RECORD_SIZE]; + record[..OFFSET_SIZE].copy_from_slice(&reset.generation.to_le_bytes()); + record[OFFSET_SIZE..2 * OFFSET_SIZE].copy_from_slice(&created_revision.to_le_bytes()); + record[2 * OFFSET_SIZE..3 * OFFSET_SIZE].copy_from_slice(&reset.floor.to_le_bytes()); + let checksum = calculate_checksum(&record[..3 * OFFSET_SIZE]); + record[3 * OFFSET_SIZE..].copy_from_slice(&checksum.to_le_bytes()); + replace_file(storage, path, record, true, true).await +} + +/// Read the durable reset boundary for this incarnation of the partition. +/// +/// Only a missing file or a valid record from another incarnation returns `None`. +/// An unreadable or damaged marker must stop recovery: treating it as absent could +/// repeat the message reset and delete messages acknowledged after that reset. +/// +/// # Errors +/// Returns an error if the existence probe, open, read, or checksum validation fails. +pub(crate) async fn read_purge_reset( + storage: &S, + path: &str, + created_revision: u64, +) -> Result, IggyError> { + if !storage + .exists_following_links(Path::new(path)) + .await + .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))? + { + return Ok(None); + } + let file = storage + .open(Path::new(path), OpenMode::Read) + .await + .map_err(|_| IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?; + let bytes = file + .read(0, PURGE_RESET_RECORD_SIZE) + .await + .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?; + let record: [u8; PURGE_RESET_RECORD_SIZE] = bytes + .try_into() + .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?; + let words: [[u8; OFFSET_SIZE]; 4] = record + .as_chunks::() + .0 + .try_into() + .map_err(|_| IggyError::CannotReadConsumerOffsets(path.to_owned()))?; + let [generation, stored_revision, floor, checksum] = words.map(u64::from_le_bytes); + if checksum != calculate_checksum(&record[..3 * OFFSET_SIZE]) { + return Err(IggyError::CannotReadConsumerOffsets(path.to_owned())); + } + if stored_revision != created_revision { + return Ok(None); + } + Ok(Some(PurgeReset { generation, floor })) +} + /// Read the purge generation this replica applied for the `created_revision` /// incarnation of the partition through the supplied storage backend. /// @@ -846,6 +934,158 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[compio::test] + async fn purge_reset_absent_file_has_no_reset_boundary() { + let dir = unique_temp_dir(); + let path = dir.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + + assert_eq!( + read_purge_reset(&DiskStorage, &path, 11) + .await + .expect("read absent reset marker"), + None, + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn purge_reset_round_trips_generation_and_floor() { + let dir = unique_temp_dir(); + let path = dir.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + let reset = PurgeReset { + generation: 3, + floor: 114, + }; + + persist_purge_reset_with_storage(&DiskStorage, &path, reset, 11) + .await + .expect("persist reset boundary"); + + assert_eq!(std::fs::metadata(&path).expect("reset marker").len(), 32); + assert_eq!( + read_purge_reset(&DiskStorage, &path, 11) + .await + .expect("read reset boundary"), + Some(reset), + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn purge_reset_from_another_incarnation_has_no_reset_boundary() { + let dir = unique_temp_dir(); + let path = dir.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + let reset = PurgeReset { + generation: 9, + floor: 114, + }; + persist_purge_reset_with_storage(&DiskStorage, &path, reset, 41) + .await + .expect("persist reset boundary for old incarnation"); + + assert_eq!( + read_purge_reset(&DiskStorage, &path, 42) + .await + .expect("ignore boundary from old incarnation"), + None, + ); + assert_eq!( + read_purge_reset(&DiskStorage, &path, 41) + .await + .expect("read boundary for original incarnation"), + Some(reset), + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn purge_reset_truncated_record_rejects_recovery() { + let dir = unique_temp_dir(); + let path = dir.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + let reset = PurgeReset { + generation: 3, + floor: 114, + }; + persist_purge_reset_with_storage(&DiskStorage, &path, reset, 11) + .await + .expect("persist reset boundary"); + let record = std::fs::read(&path).expect("read complete reset marker"); + + for length in 0..record.len() { + std::fs::write(&path, &record[..length]).expect("truncate reset marker"); + let result = read_purge_reset(&DiskStorage, &path, 11).await; + assert!( + matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))), + "a reset marker truncated to {length} bytes must stop recovery, got {result:?}", + ); + } + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn purge_reset_corrupt_record_rejects_recovery() { + let dir = unique_temp_dir(); + let path = dir.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + let reset = PurgeReset { + generation: 3, + floor: 114, + }; + persist_purge_reset_with_storage(&DiskStorage, &path, reset, 11) + .await + .expect("persist reset boundary"); + let record = std::fs::read(&path).expect("read intact reset marker"); + + for byte_index in 0..record.len() { + let mut corrupt_record = record.clone(); + corrupt_record[byte_index] ^= 0x01; + std::fs::write(&path, corrupt_record).expect("corrupt reset marker"); + let result = read_purge_reset(&DiskStorage, &path, 11).await; + assert!( + matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))), + "corruption at byte {byte_index} must stop recovery, got {result:?}", + ); + } + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn purge_reset_existence_probe_error_rejects_recovery() { + let dir = unique_temp_dir(); + let parent = dir.join("not-a-directory"); + std::fs::write(&parent, []).expect("block traversal to reset marker"); + let path = parent.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + + let result = read_purge_reset(&DiskStorage, &path, 11).await; + assert!( + matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))), + "a failed existence probe must stop recovery, got {result:?}", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[compio::test] + async fn purge_reset_read_error_rejects_recovery() { + let dir = unique_temp_dir(); + let path = dir.join(PURGE_RESET_FILE).to_string_lossy().into_owned(); + // A directory opens successfully but fails reads, exercising a storage + // failure independently of short records or checksum corruption. + std::fs::create_dir(&path).expect("make reset marker unreadable as a file"); + + let result = read_purge_reset(&DiskStorage, &path, 11).await; + assert!( + matches!(result, Err(IggyError::CannotReadConsumerOffsets(_))), + "an unreadable reset marker must stop recovery, got {result:?}", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + #[compio::test] async fn persist_offset_max_recovers_torn_file() { let dir = unique_temp_dir(); diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index e4ffae9e27..4e0f787d94 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -29,7 +29,7 @@ use crate::messages_writer::MessagesWriter; use crate::offset_storage::{ - PURGE_GENERATION_FILE, commit_offset_replacement, delete_persisted_offset, + PURGE_GENERATION_FILE, PURGE_RESET_FILE, commit_offset_replacement, delete_persisted_offset, discard_offset_replacement, offset_replacement_id, persist_purge_generation, stage_offset_replacement, }; @@ -644,6 +644,457 @@ const fn validate_consumer_offset_transfer_count( #[cfg(test)] mod tests { use super::*; + use crate::journal::MessageLookup; + use crate::offset_storage::{PurgeReset, persist_purge_reset_with_storage, read_purge_reset}; + use crate::{Partition, PartitionPathLayout}; + use bytes::Bytes; + use consensus::{LocalPipeline, VsrConsensus}; + use iggy_binary_protocol::{Command, RoutedRequestHeader}; + use iggy_common::{ConsumerGroupOffsets, ConsumerOffsets, PartitionStats}; + use message_bus::IggyMessageBus; + use server_common::send_messages::{ + IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned, + }; + use server_common::sharding::IggyNamespace; + use std::sync::Arc; + + #[compio::test] + async fn given_purge_reset_when_quarantining_should_move_it_with_discarded_history() { + let directory = tempfile::tempdir().unwrap(); + let partition_dir = directory.path().join("partition"); + std::fs::create_dir(&partition_dir).unwrap(); + let reset_path = partition_dir.join(PURGE_RESET_FILE); + let reset = PurgeReset { + generation: 5, + floor: 3, + }; + persist_purge_reset_with_storage(&DiskStorage, reset_path.to_str().unwrap(), reset, 7) + .await + .unwrap(); + std::fs::write(partition_dir.join("0.log"), b"discarded history").unwrap(); + std::fs::write(partition_dir.join("superblock.a"), b"durable view").unwrap(); + std::fs::write(partition_dir.join("unrelated.reset"), b"unrelated record").unwrap(); + + let quarantined = quarantine_partition_files(partition_dir.to_str().unwrap(), None) + .await + .unwrap(); + + assert!(!reset_path.exists()); + assert_eq!( + read_purge_reset( + &DiskStorage, + Path::new(&quarantined) + .join(PURGE_RESET_FILE) + .to_str() + .unwrap(), + 7, + ) + .await + .unwrap(), + Some(reset) + ); + assert_eq!( + std::fs::read(Path::new(&quarantined).join("0.log")).unwrap(), + b"discarded history" + ); + assert_eq!( + std::fs::read(partition_dir.join("superblock.a")).unwrap(), + b"durable view" + ); + assert_eq!( + std::fs::read(partition_dir.join("unrelated.reset")).unwrap(), + b"unrelated record" + ); + } + + #[compio::test] + async fn given_pending_purge_cleanup_when_installing_should_preserve_partition_files() { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let consumer_directory = root.join("consumer_offsets"); + let group_directory = root.join("consumer_group_offsets"); + std::fs::create_dir(&consumer_directory).unwrap(); + std::fs::create_dir(&group_directory).unwrap(); + std::fs::write(root.join("0.log"), b"current message history").unwrap(); + std::fs::write(consumer_directory.join("7"), 2u64.to_le_bytes()).unwrap(); + let reset_path = root.join(PURGE_RESET_FILE); + persist_purge_reset_with_storage( + &DiskStorage, + reset_path.to_str().unwrap(), + PurgeReset { + generation: 5, + floor: 3, + }, + 7, + ) + .await + .unwrap(); + let reset_bytes = std::fs::read(&reset_path).unwrap(); + let consensus = + VsrConsensus::new(1, 0, 3, 42, IggyMessageBus::new(0), LocalPipeline::new()); + consensus.init(); + let mut partition: IggyPartition = IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + IggyByteSize::from(1024 * 1024), + ); + partition.set_partition_dir(root.to_string_lossy().into_owned()); + partition.set_created_revision(7); + partition.configure_consumer_offset_storage( + consumer_directory.to_string_lossy().into_owned(), + group_directory.to_string_lossy().into_owned(), + ConsumerOffsets::with_capacity(0), + ConsumerGroupOffsets::with_capacity(0), + ); + partition.hydrate_applied_purge_generation().await.unwrap(); + assert!(partition.purge_cleanup_pending()); + let config = PartitionsConfig { + messages_required_to_save: 1, + size_of_messages_required_to_save: IggyByteSize::from(1024 * 1024), + validate_checksum: true, + segment_size: IggyByteSize::from(1024 * 1024), + preallocate_segments: false, + encryptor: None, + path_layout: PartitionPathLayout::default(), + }; + let mut offered_offsets = table(); + offered_offsets.purge_generation = 5; + let offered_bytes = offered_offsets.encode(); + + let result = partition + .install_state_transfer(&config, 4, Vec::new(), &offered_bytes, 5) + .await; + + assert!(matches!( + result, + Err(PartitionInstallError::PurgeCleanupPending) + )); + assert!(partition.purge_cleanup_pending()); + assert_eq!(partition.applied_purge_generation(), 0); + assert_eq!(partition.purge_floor_op(), 3); + assert_eq!(std::fs::read(&reset_path).unwrap(), reset_bytes); + assert_eq!( + std::fs::read(root.join("0.log")).unwrap(), + b"current message history" + ); + assert_eq!( + std::fs::read(consumer_directory.join("7")).unwrap(), + 2u64.to_le_bytes() + ); + assert!(!root.join(PURGE_GENERATION_FILE).exists()); + assert!( + !root.join("00000000000000000043.log").exists(), + "the offered empty chain must not be planted" + ); + } + + #[compio::test] + async fn given_completed_purge_when_installing_lower_head_should_recover_rebased_floor() { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let mut partition = transfer_test_partition(root); + for op in 1..=9 { + append_transfer_test_message(&mut partition, op).await; + } + assert_eq!( + partition.log.journal().inner.oldest_resident_offset(), + Some(0) + ); + partition.purge(&transfer_test_config(), 5).await.unwrap(); + assert_eq!(partition.purge_floor_op(), 9); + assert_eq!(partition.consensus().sequencer().current_sequence(), 9); + assert!(partition.log.journal().inner.resident_entries().is_empty()); + assert!(partition.log.journal().inner.header_by_op(9).is_some()); + let mut offered = table(); + offered.purge_generation = 5; + + partition + .install_state_transfer(&transfer_test_config(), 4, Vec::new(), &offered.encode(), 5) + .await + .expect("install replacement history below the old uncommitted floor"); + + assert_eq!(partition.consensus().sequencer().current_sequence(), 4); + assert_eq!(partition.purge_floor_op(), 4); + + // Replacement operations reuse numbers below the old purge floor. + // They must enter the resident poll tier before any segment flush. + append_transfer_test_message(&mut partition, 5).await; + let journal = &partition.log.journal().inner; + let resident = journal.resident_entries(); + assert_eq!( + resident.len(), + 1, + "the replacement send must remain visible" + ); + assert_eq!(journal.resident_message_entries().len(), 1); + assert_eq!(journal.oldest_resident_offset(), Some(offered.next_offset)); + for query in [ + MessageLookup::Offset { + offset: offered.next_offset, + count: 1, + ceiling: u64::MAX, + }, + MessageLookup::Timestamp { + timestamp: 5, + count: 1, + ceiling: u64::MAX, + }, + ] { + let (fragments, last_offset) = journal.get_sync(&query).expect("poll replacement send"); + assert!(!fragments.is_empty()); + assert_eq!(last_offset, Some(offered.next_offset)); + } + drop(partition); + let mut restarted = transfer_test_partition(root); + restarted.hydrate_applied_purge_generation().await.unwrap(); + assert_eq!(restarted.applied_purge_generation(), 5); + assert_eq!(restarted.purge_floor_op(), 4); + assert!(!restarted.purge_cleanup_pending()); + assert!(restarted.purge_recovery_pending()); + restarted.recover_purge_boundary(4).await.unwrap(); + offered.purge_generation = 6; + restarted + .install_state_transfer(&transfer_test_config(), 4, Vec::new(), &offered.encode(), 6) + .await + .expect("a snapshot resolves the lost journal after restart"); + assert!(!restarted.purge_recovery_pending()); + drop(restarted); + + // The snapshot advanced the generation beyond the local reset record. + // Losing its journal on another restart must still require recovery. + let mut after_snapshot = transfer_test_partition(root); + after_snapshot + .hydrate_applied_purge_generation() + .await + .unwrap(); + assert_eq!(after_snapshot.applied_purge_generation(), 6); + assert!(after_snapshot.purge_recovery_pending()); + } + + async fn append_transfer_test_message(partition: &mut IggyPartition, op: u64) { + let mut messages = IggyMessages::with_capacity(1); + messages.push(IggyMessage { + header: IggyMessageHeader { + id: u128::from(op), + ..Default::default() + }, + payload: Bytes::from_static(b"message"), + user_headers: None, + }); + let message = SendMessagesOwned::from_messages( + IggyNamespace::from_raw(partition.consensus().group()), + &messages, + ) + .unwrap() + .encode_request(RoutedRequestHeader { + command: Command::Request, + operation: Operation::SendMessages, + client: 1, + session: 1, + request: op, + group: partition.consensus().group(), + ..Default::default() + }) + .unwrap() + .transmute_header( + |request: RoutedRequestHeader, prepare: &mut PrepareHeader| { + prepare.command = Command::Prepare; + prepare.operation = Operation::SendMessages; + prepare.group = partition.consensus().group(); + prepare.op = op; + prepare.timestamp = op; + prepare.size = request.size; + }, + ); + partition.append_messages(message).await.unwrap(); + partition.consensus().sequencer().set_sequence(op); + } + + #[compio::test] + async fn given_failed_purge_floor_rebase_when_installing_should_preserve_files_without_wal() { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let mut partition = Box::pin(partition_with_completed_reset(root)).await; + std::fs::write(root.join("0.log"), b"existing history").unwrap(); + let reset_path = root.join(PURGE_RESET_FILE); + let reset_bytes = std::fs::read(&reset_path).unwrap(); + // Refuse only the marker's replacement, after the offer passes validation. + std::fs::create_dir(root.join("purge.reset.tmp")).unwrap(); + let mut offered = table(); + offered.purge_generation = 5; + + let result = partition + .install_state_transfer(&transfer_test_config(), 4, Vec::new(), &offered.encode(), 5) + .await; + + assert!(matches!( + result, + Err(PartitionInstallError::PurgeResetNotDurable(_)) + )); + assert_eq!(partition.purge_floor_op(), 9); + assert_eq!(partition.consensus().sequencer().current_sequence(), 9); + assert!(partition.fatal().is_some()); + assert!(root.join(".install-backup").is_dir()); + assert_eq!( + std::fs::read(root.join("0.log")).unwrap(), + b"existing history" + ); + assert_eq!(std::fs::read(&reset_path).unwrap(), reset_bytes); + assert!(!root.join("00000000000000000043.log").exists()); + } + + #[compio::test] + async fn given_install_failure_after_rebase_without_wal_when_recovering_should_restore_history() + { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let mut partition = Box::pin(partition_with_completed_reset(root)).await; + assert!(partition.persistence.is_none()); + drop(partition.log.retire_front().unwrap()); + partition + .install_empty_segment(&transfer_test_config(), 0) + .await + .unwrap(); + let original_log = root.join("00000000000000000000.log"); + std::fs::write(&original_log, b"original history").unwrap(); + let mut offered = table(); + offered.purge_generation = 5; + // Missing staging files fail the swap after the floor changes and old + // segments are removed, exercising the production backup decision. + let missing_segment = StagedSegmentMeta { + start_offset: 0, + end_offset: 0, + index_size: 0, + size: 8, + start_timestamp: 0, + end_timestamp: 0, + max_timestamp: 0, + log_staging: root.join("00000000000000000000.log.staging"), + index_staging: root.join("00000000000000000000.index.staging"), + }; + + let result = partition + .install_state_transfer( + &transfer_test_config(), + 4, + vec![missing_segment], + &offered.encode(), + 5, + ) + .await; + + assert!(matches!(result, Err(PartitionInstallError::SwapIo { .. }))); + assert!(partition.fatal().is_some()); + assert!(root.join(".install-backup").is_dir()); + assert!(!original_log.clone().exists()); + assert_eq!(partition.purge_floor_op(), 4); + assert_eq!( + read_purge_reset( + &DiskStorage, + root.join(PURGE_RESET_FILE).to_str().unwrap(), + 7 + ) + .await + .unwrap() + .unwrap() + .floor, + 4, + ); + + drop(partition); + crate::install_backup::recover(root).await.unwrap(); + let mut restarted = transfer_test_partition(root); + restarted.hydrate_applied_purge_generation().await.unwrap(); + assert_eq!(restarted.applied_purge_generation(), 5); + assert_eq!(restarted.purge_floor_op(), 9); + assert_eq!( + std::fs::read(original_log.clone()).unwrap(), + b"original history" + ); + } + + #[compio::test] + async fn given_interrupted_install_when_rolled_back_should_restore_purge_floor_with_history() { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let mut partition = Box::pin(partition_with_completed_reset(root)).await; + std::fs::write(root.join("0.log"), b"original history").unwrap(); + crate::install_backup::begin(root).await.unwrap(); + + partition.rebase_purge_reset_for_install(4).await.unwrap(); + assert_eq!(partition.purge_floor_op(), 4); + std::fs::remove_file(root.join("0.log")).unwrap(); + std::fs::write(root.join("0.log"), b"replacement history").unwrap(); + drop(partition); + crate::install_backup::recover(root).await.unwrap(); + + let mut restarted = transfer_test_partition(root); + restarted.hydrate_applied_purge_generation().await.unwrap(); + assert_eq!(restarted.purge_floor_op(), 9); + assert_eq!(restarted.applied_purge_generation(), 5); + assert_eq!( + std::fs::read(root.join("0.log")).unwrap(), + b"original history" + ); + } + + async fn partition_with_completed_reset(root: &Path) -> IggyPartition { + let mut partition = transfer_test_partition(root); + persist_purge_generation(root.join(PURGE_GENERATION_FILE).to_str().unwrap(), 5, 7) + .await + .unwrap(); + persist_purge_reset_with_storage( + &DiskStorage, + root.join(PURGE_RESET_FILE).to_str().unwrap(), + PurgeReset { + generation: 5, + floor: 9, + }, + 7, + ) + .await + .unwrap(); + partition.hydrate_applied_purge_generation().await.unwrap(); + partition.consensus().sequencer().set_sequence(9); + assert!(!partition.purge_cleanup_pending()); + partition + } + + fn transfer_test_partition(root: &Path) -> IggyPartition { + let consensus = + VsrConsensus::new(1, 0, 3, 42, IggyMessageBus::new(0), LocalPipeline::new()); + consensus.init(); + let mut partition = IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + IggyByteSize::from(1024 * 1024), + ); + partition.set_partition_dir(root.to_string_lossy().into_owned()); + partition.set_created_revision(7); + let consumer_directory = root.join("consumer_offsets"); + let group_directory = root.join("consumer_group_offsets"); + std::fs::create_dir_all(&consumer_directory).unwrap(); + std::fs::create_dir_all(&group_directory).unwrap(); + partition.configure_consumer_offset_storage( + consumer_directory.to_string_lossy().into_owned(), + group_directory.to_string_lossy().into_owned(), + ConsumerOffsets::with_capacity(0), + ConsumerGroupOffsets::with_capacity(0), + ); + partition + } + + fn transfer_test_config() -> PartitionsConfig { + PartitionsConfig { + messages_required_to_save: 1, + size_of_messages_required_to_save: IggyByteSize::from(1024 * 1024), + validate_checksum: true, + segment_size: IggyByteSize::from(1024 * 1024), + preallocate_segments: false, + encryptor: None, + path_layout: PartitionPathLayout::default(), + } + } #[compio::test] async fn given_transient_offset_io_failure_when_retried_should_succeed_without_exhausting_budget() @@ -1384,6 +1835,10 @@ pub struct PartitionInstallOutcome { #[derive(Debug)] pub enum PartitionInstallError { NoPartitionDir, + /// Pending purge cleanup must finish before imported bookmarks replace the old table. + PurgeCleanupPending, + /// The old purge floor could not be rebased before replacing message history. + PurgeResetNotDurable(iggy_common::IggyError), NoOffsetDir { kind: ConsumerKind, }, @@ -1459,6 +1914,18 @@ pub enum PartitionInstallError { impl fmt::Display for PartitionInstallError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + Self::PurgeResetNotDurable(source) => { + write!( + f, + "could not rebase the purge floor before state transfer: {source}" + ) + } + Self::PurgeCleanupPending => { + write!( + f, + "purge cleanup must finish before state transfer can install" + ) + } Self::NoPartitionDir => write!(f, "partition has no on-disk directory"), Self::NoOffsetDir { kind } => { write!(f, "partition has no {kind:?} offset directory configured") @@ -1680,11 +2147,14 @@ pub async fn quarantine_partition_files( )); }; for path in segment_dir_entries(partition_dir)? { - let quarantined = path.to_str().is_some_and(|path| { - [".log", ".index", STAGING_SUFFIX, ANCHOR_SUFFIX] - .iter() - .any(|suffix| path.ends_with(suffix)) - }); + let quarantined = path + .file_name() + .is_some_and(|name| name == PURGE_RESET_FILE) + || path.to_str().is_some_and(|path| { + [".log", ".index", STAGING_SUFFIX, ANCHOR_SUFFIX] + .iter() + .any(|suffix| path.ends_with(suffix)) + }); if !quarantined { continue; } @@ -2565,6 +3035,9 @@ where ) -> Result { // ---- check phase: nothing below may mutate live state. Staging // writes only sibling files the install can abandon. ---- + if self.purge_cleanup_pending() { + return Err(PartitionInstallError::PurgeCleanupPending); + } let Some(partition_dir) = self.partition_dir.clone() else { return Err(PartitionInstallError::NoPartitionDir); }; @@ -2700,6 +3173,10 @@ where let write_lock = self.write_lock.clone(); let _guard = write_lock.lock().await; + // Lowering a durable purge floor and replacing its history must roll back + // together, including partitions that do not keep a durable journal. + let backup_required = + self.persistence.is_some() || self.purge_reset_needs_rebase(commit_op); if let Some(persistence) = &self.persistence { self.start_persistence(); persistence.drain_with_timeout().await.map_err(|source| { @@ -2708,15 +3185,17 @@ where source, } })?; - if let Err(source) = crate::install_backup::begin(Path::new(&partition_dir)).await { - // The backup rename may have landed before its directory barrier failed. - // Further commits could then be erased by rollback on the next boot. - self.fence_install_failure(commit_op); - return Err(PartitionInstallError::SwapIo { - path: partition_dir.clone(), - source, - }); - } + } + if backup_required + && let Err(source) = crate::install_backup::begin(Path::new(&partition_dir)).await + { + // The backup rename may have landed before its directory barrier failed. + // Further commits could then be erased by rollback on the next boot. + self.fence_install_failure(commit_op); + return Err(PartitionInstallError::SwapIo { + path: partition_dir.clone(), + source, + }); } // ---- mutate phase ---- @@ -2755,7 +3234,7 @@ where .await }; if !frontier_durable { - if self.persistence.is_some() { + if backup_required { self.fence_install_failure(commit_op); } discard_offset_writes(&planned_offsets).await; @@ -2763,6 +3242,16 @@ where frontier: offsets_wire.next_offset, }); } + // The install may replace an uncommitted suffix below the old purge + // floor. Rebase its durable record before any replacement can survive + // a crash. The install backup restores the old floor with its history. + if let Err(source) = self.rebase_purge_reset_for_install(commit_op).await { + if backup_required { + self.fence_install_failure(commit_op); + } + discard_offset_writes(&planned_offsets).await; + return Err(PartitionInstallError::PurgeResetNotDurable(source)); + } // The write lock spans the convergence too: a mutate failure leaves // the segment vectors drained, and a concurrent replicated append // indexing `segments().len() - 1` on the emptied vec is exactly the @@ -2783,7 +3272,7 @@ where ) .await; if outcome.is_err() { - if self.persistence.is_some() { + if backup_required { // Keep the rollback snapshot intact until boot reopens every file. self.fence_install_failure(commit_op); return outcome; @@ -2830,7 +3319,7 @@ where the durable record stays at the pre-swap claim until the next view change" ); } - if self.persistence.is_some() + if backup_required && let Err(source) = crate::install_backup::finish(Path::new(&partition_dir)).await { self.fence_install_failure(commit_op); @@ -2848,6 +3337,9 @@ where })?; self.materialization_missing = false; } + if outcome.is_ok() { + self.purge_recovery = crate::iggy_partition::PurgeRecovery::Ready; + } outcome } diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index d7ecfd7989..7df8e7bd33 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -150,8 +150,8 @@ pub async fn create_partition_file_hierarchy( /// /// # Errors /// -/// Returns [`ServerError::ConsumerOffsetsLoad`] when an existing offset -/// directory cannot be enumerated. A stored offset past the offset space is clamped +/// Returns an error if pending purge cleanup cannot complete or 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. pub async fn configure_consumer_offsets( partition: &mut IggyPartition>, @@ -171,9 +171,10 @@ pub async fn configure_consumer_offsets( /// 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. +/// Hydrate the purge markers and restore the message offset and reservation +/// frontier before calling this. Pass the restored counter as `current_offset`. +/// Pending purge cleanup finishes before any bookmark is loaded; the restored +/// frontiers then 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 @@ -184,9 +185,9 @@ pub async fn configure_consumer_offsets( /// 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. +/// Returns a cleanup error if an earlier message reset still has bookmarks that +/// cannot be removed durably, or [`ServerError::ConsumerOffsetsLoad`] if an offset +/// directory cannot be enumerated. Callers must discard a failed recovery. #[allow(clippy::too_many_lines)] pub async fn configure_consumer_offsets_with_storage( storage: &S, @@ -201,6 +202,16 @@ pub async fn configure_consumer_offsets_with_storage( let consumer_offsets_path = config.get_consumer_offsets_path(stream_id, topic_id, partition_id); let consumer_group_offsets_path = config.get_consumer_group_offsets_path(stream_id, topic_id, partition_id); + // A completed message reset may still have old bookmark files after power + // loss. Finish their cleanup before they can seed recovered progress. + partition.configure_consumer_offset_storage( + consumer_offsets_path.clone(), + consumer_group_offsets_path.clone(), + ConsumerOffsets::with_capacity(0), + ConsumerGroupOffsets::with_capacity(0), + ); + partition.resume_purge_cleanup_with_storage(storage).await?; + // The bound is the offset space this replica could have MINTED, not the data // it can still serve. A boot re-anchor leaves the append point a lease block // above the recovered chain, so on the restart after a crash that took @@ -906,7 +917,7 @@ async fn load_partition( .map(|state| (state.view, state.log_view)), view_fallback: None, seed_view: None, - incarnation: None, + incarnation: Some(rand::random::() | 1), join, }, ); @@ -1451,7 +1462,7 @@ pub async fn build_partition_fresh( // is restamped per delivery), so every replica commits the same // one and a late materialiser lands at or below its peers. seed_view, - incarnation: None, + incarnation: Some(rand::random::() | 1), join, }, ); @@ -1662,12 +1673,14 @@ pub async fn delete_partitions_from_disk( #[cfg(test)] mod tests { use super::*; + use crate::dispatch::test_support::prepare_message; use bytes::Bytes; use configs::server::ServerConfig; + use consensus::Sequencer; use iggy_binary_protocol::batch::BATCH_HEADER_SIZE; use iggy_binary_protocol::{Command, Operation, PrepareHeader}; - use journal::DurableAppend; use journal::superblock::SuperblockStore; + use journal::{DurableAppend, Journal}; use partitions::PartitionPathLayout; use server_common::Message; use server_common::send_messages::{ @@ -1679,6 +1692,104 @@ mod tests { const REPLICA: u8 = 1; const REPLICAS: u8 = 3; + #[compio::test] + async fn loading_replicated_purge_keeps_lost_history_fenced_with_a_new_boot_nonce() { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let namespace = IggyNamespace::new(1, 1, 0); + let runtime = TopicRuntimeOptions { + durability: iggy_common::Durability::Replicated, + consumer_offset_durability: iggy_common::Durability::Replicated, + preallocate_segments: Some(false), + ..Default::default() + }; + let partitions = solo_partitions(); + let metadata = Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, 0); + let mut partition = build_partition_fresh( + &config, + namespace, + Arc::new(PartitionStats::default()), + 0, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + 0, + Rc::new(IggyMessageBus::new(0)), + ) + .await + .unwrap(); + let mut previous_nonce = partition.consensus().incarnation(); + assert_ne!(previous_nonce, 0); + let mut messages = IggyMessages::with_capacity(1); + messages.push(IggyMessage { + header: IggyMessageHeader::default(), + payload: Bytes::from_static(b"before-purge"), + user_headers: None, + }); + let mut batch = SendMessagesOwned::from_messages(namespace, &messages).unwrap(); + for op in 1..=3 { + batch.header.base_offset = op - 1; + batch.header.batch_checksum = batch.header.checksum_for_blob(&batch.blob); + let mut body = vec![0; batch.header.total_size()]; + batch.header.encode_into(&mut body); + body[BATCH_HEADER_SIZE..].copy_from_slice(&batch.blob); + let prepare = prepare_message(Operation::SendMessages, 1, op, &body).transmute_header( + |original, header: &mut PrepareHeader| { + *header = original; + header.cluster = CLUSTER; + header.group = namespace.inner(); + header.op = op; + header.parent = partition.consensus().last_prepare_checksum(); + }, + ); + let prepare = consensus::seal_prepare_checksum(prepare); + partition.consensus().sequencer().set_sequence(op); + partition + .consensus() + .set_last_prepare_checksum(prepare.header().checksum); + partition + .log + .journal() + .inner + .append(prepare.into_frozen()) + .await + .unwrap(); + } + partition.purge(partitions.config(), 1).await.unwrap(); + assert_eq!(partition.purge_floor_op(), 3); + assert!(partition.log.journal().inner.header_by_op(3).is_some()); + drop(partition); + + // Only the directory survives each boot. Retaining a simulator journal + // here would conceal the mismatch between the old floor and new sequence. + for _ in 0..2 { + let recovered = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &metadata, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(recovered.applied_purge_generation(), 1); + assert_eq!(recovered.purge_floor_op(), 3); + assert_eq!(recovered.consensus().sequencer().current_sequence(), 0); + assert_eq!(recovered.log.journal().inner.last_op(), None); + assert!(recovered.purge_recovery_pending()); + assert_ne!(recovered.consensus().incarnation(), 0); + assert_ne!(recovered.consensus().incarnation(), previous_nonce); + previous_nonce = recovered.consensus().incarnation(); + } + } + #[compio::test] async fn loading_after_retention_preserves_the_wal_owned_tail_when_the_logical_chain_is_empty() { diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index dd188fc0d7..3343b88380 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4704,13 +4704,21 @@ where ); return; }; - partition.ensure_materialization_recovery(); - let actions = - partition - .consensus() - .handle_start_view(PlaneKind::Partitions, &header, suffix_body); + let actions = partition_start_view_actions(partition, &header, suffix_body); let adopted = !actions.is_empty(); if adopted { + let selected_head = partition.consensus().sequencer().current_sequence(); + if let Err(error) = partition.recover_purge_boundary(selected_head).await { + tracing::warn!( + shard = self.id, + namespace_raw = header.group, + view = header.view, + %error, + "cannot record the purge boundary for the adopted history; probing again" + ); + partition.consensus().begin_view_probe(); + return; + } // Any stream armed before this adoption belongs to the superseded // view. Repair bodies carry no nonce, so drop the receiving session // before reconciling or arming the new view's canonical range. @@ -4865,6 +4873,7 @@ where // the stale peer keeps heartbeating, so it re-triggers once a // later persist succeeds. if !partition.requires_state_transfer() + && !partition.purge_recovery_pending() && partition.persist_superblock_if_needed().await { respond_start_view::(consensus).await; @@ -5050,7 +5059,7 @@ where let Some(partition) = planes.1.0.get_mut_by_ns(&namespace) else { return; }; - if !partition.consensus().is_normal() { + if !partition.consensus().is_normal() || partition.purge_recovery_pending() { return; } let cluster = partition.consensus().cluster(); @@ -5862,6 +5871,23 @@ where let Some(pending) = partition.consensus().pending_view_log() else { return; }; + if partition.purge_recovery_pending() { + if pending.op_head > 0 { + // This replica cannot classify reused operation numbers from + // journal bodies. Leave the election timer running so a peer + // with the selected materialized history can become primary. + return; + } + if let Err(error) = partition.recover_purge_boundary(pending.op_head).await { + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + %error, + "cannot record the purge boundary before starting the empty view" + ); + return; + } + } // Before the scan can mean anything: the repair ingest skips an op it // already holds a header for, so the scan would report a gap nothing // fills. Backups reach this on StartView adoption; a primary-elect has no @@ -9011,7 +9037,11 @@ where B: MessageBus, { let consensus = partition.consensus(); - if !consensus.is_normal() || consensus.is_transferring() || partition.repair.is_some() { + if !consensus.is_normal() + || consensus.is_transferring() + || partition.purge_recovery_pending() + || partition.repair.is_some() + { return false; } // Read, never scanned: the tally is republished by each sweep (see @@ -9134,7 +9164,7 @@ where ) where B: MessageBus, { - if partition.repair.is_some() || from_op > to_op { + if partition.purge_recovery_pending() || partition.repair.is_some() || from_op > to_op { return; } if self.partition_repairs_inflight.get() >= PARTITION_REPAIRS_INFLIGHT_MAX { @@ -9885,6 +9915,9 @@ where peer: next_peer, after_ticks, }); + if partition.purge_recovery_pending() { + return; + } let config = self.plane.partitions().config().clone(); partition.commit_journal(&config).await; self.maybe_request_partition_repair(partition, peer).await; @@ -11125,7 +11158,8 @@ where let consensus = partition.consensus(); let commit_min = consensus.commit_min(); let commit_max = consensus.commit_max(); - let recovery_owned = partition.transfer.is_some() + let recovery_owned = partition.purge_recovery_pending() + || partition.transfer.is_some() || partition.transfer_rearm.is_some() || partition.repair.is_some(); let normal = consensus.is_normal(); @@ -11807,6 +11841,28 @@ async fn dispatch_vsr_actions( } } +fn partition_start_view_actions( + partition: &IggyPartition, + header: &StartViewHeader, + suffix_body: &[u8], +) -> Vec +where + B: MessageBus, + SB: SuperblockStore, +{ + let consensus = partition.consensus(); + // A delayed announcement could clear the purge boundary or pin recovery to + // a lost history. A newer view or the current boot's probe reply is required. + if partition.purge_recovery_pending() + && header.view <= consensus.view() + && (consensus.incarnation() == 0 || header.incarnation != consensus.incarnation()) + { + return Vec::new(); + } + partition.ensure_materialization_recovery(); + consensus.handle_start_view(PlaneKind::Partitions, header, suffix_body) +} + #[allow(clippy::future_not_send)] async fn dispatch_partition_wire_actions( consensus: &VsrConsensus, @@ -11824,6 +11880,15 @@ async fn dispatch_partition_wire_actions( } if partition.requires_state_transfer() { actions.retain(|action| matches!(action, VsrAction::SendRequestStartView { .. })); + } else if partition.purge_recovery_pending() { + actions.retain(|action| { + matches!( + action, + VsrAction::SendRequestStartView { .. } + | VsrAction::SendStartViewChange { .. } + | VsrAction::SendDoViewChange { .. } + ) + }); } dispatch_vsr_actions::(consensus, None, &actions).await; dispatch_partition_journal_actions(consensus, partition, &actions).await; @@ -11843,6 +11908,9 @@ async fn dispatch_partition_journal_actions( P: Pipeline, SB: SuperblockStore, { + if partition.purge_recovery_pending() { + return; + } let bus = consensus.message_bus(); let self_id = consensus.replica(); let journal = &partition.log.journal().inner; @@ -13404,12 +13472,137 @@ mod partition_ack_durability_tests { use consensus::LocalPipeline; use iggy_common::PartitionStats; use iggy_common::{Durability, IggyByteSize, TopicRuntimeOptions}; + use journal::durable_storage::DiskStorage; use journal::prepare_journal::PrepareJournal; use message_bus::IggyMessageBus; use server_common::iobuf::Owned; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; + #[compio::test] + #[allow(clippy::too_many_lines)] + async fn given_unresolved_purge_history_when_rejoining_should_wait_for_a_fresh_view() { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let directory = std::env::temp_dir().join(format!( + "iggy-purge-recovery-dispatch-{}-{unique}", + std::process::id() + )); + std::fs::create_dir(&directory).unwrap(); + let make_partition = |bus| { + let mut consensus = VsrConsensus::new(1, 0, 3, 42, bus, LocalPipeline::new()); + consensus.set_view(1); + consensus.set_log_view(1); + consensus.mark_superblock_durable(1, 1); + consensus.init_as_backup(); + consensus.begin_view_probe(); + let mut partition: IggyPartition = + IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + IggyByteSize::from(1024 * 1024), + ); + partition.set_partition_dir(directory.to_string_lossy().into_owned()); + partition.set_runtime_options(TopicRuntimeOptions { + durability: Durability::Replicated, + consumer_offset_durability: Durability::Replicated, + ..TopicRuntimeOptions::default() + }); + partition + }; + let mut before_restart = make_partition(IggyMessageBus::new(0)); + before_restart.consensus().sequencer().set_sequence(3); + // The empty in-memory segment is already the purge's reset state. + before_restart + .complete_purge_with_storage(&DiskStorage, 1) + .await + .unwrap(); + drop(before_restart); + + let bus = IggyMessageBus::new(0); + let sent = Rc::new(RefCell::new(Vec::new())); + let captured = sent.clone(); + bus.set_replica_forward_fn(Box::new(move |_, _, frame| { + captured.borrow_mut().push(frame); + Ok(()) + })); + for replica in 1..3 { + assert!(bus.owner_table().try_claim(replica, 1)); + } + let mut restarted = make_partition(bus); + restarted.consensus().set_incarnation(7); + restarted.hydrate_applied_purge_generation().await.unwrap(); + assert!(restarted.purge_recovery_pending()); + assert_eq!(restarted.purge_floor_op(), 3); + assert_eq!(restarted.consensus().sequencer().current_sequence(), 0); + let start_view = VsrAction::SendStartView { + view: 1, + op: 0, + commit: 0, + incarnation: 0, + target: Some(1), + group: 42, + suffix: Vec::new(), + }; + dispatch_partition_wire_actions::<_, _, PrepareJournal, _>( + restarted.consensus(), + &restarted, + vec![ + start_view.clone(), + VsrAction::SendStartViewChange { view: 1, group: 42 }, + VsrAction::SendRequestStartView { view: 1, group: 42 }, + ], + ) + .await; + assert_eq!(sent.borrow().len(), 4); + for frame in sent.borrow().iter() { + let message = + Message::::try_from(Owned::copy_from_slice(frame.as_slice())) + .unwrap(); + assert!(matches!( + message.header().command, + Command::StartViewChange | Command::RequestStartView + )); + } + + sent.borrow_mut().clear(); + let mut reply = *Message::::new(size_of::()) + .transmute_header(|_, header: &mut StartViewHeader| { + header.command = Command::StartView; + header.cluster = 1; + header.group = 42; + header.view = 1; + header.replica = 1; + header.size = u32::try_from(size_of::()).unwrap(); + header.seal(); + }) + .header(); + for (op, incarnation) in [(0, 0), (0, 6), (3, 0), (3, 6)] { + reply.op = op; + reply.incarnation = incarnation; + assert!(partition_start_view_actions(&restarted, &reply, &[]).is_empty()); + assert_eq!(restarted.consensus().status(), Status::Recovering); + assert!(restarted.purge_recovery_pending()); + assert_eq!(restarted.purge_floor_op(), 3); + } + reply.op = 0; + reply.incarnation = 7; + assert!(!partition_start_view_actions(&restarted, &reply, &[]).is_empty()); + restarted.recover_purge_boundary(0).await.unwrap(); + dispatch_partition_wire_actions::<_, _, PrepareJournal, _>( + restarted.consensus(), + &restarted, + vec![start_view], + ) + .await; + assert_eq!(sent.borrow().len(), 1); + assert!(!restarted.purge_recovery_pending()); + drop(restarted); + std::fs::remove_dir_all(directory).unwrap(); + } + #[compio::test] #[allow(clippy::too_many_lines)] async fn ordinary_start_view_replies_do_not_turn_missing_bodies_into_canonical_headers() { diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 9dec7e986b..47e3d5019c 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -1002,32 +1002,25 @@ where the reconciler re-issues it while the generation stays unapplied" ); } - Err(error @ partitions::PurgeError::GenerationNotRecorded(_)) => { - // NOT fenced: the wipe ran and a fresh chain is - // planted, so the partition is serviceable; only - // the durable generation record failed, which - // leaves `applied_purge_generation` unmoved and - // the reconciler re-issuing the (now cheap) purge. - // Same pacing argument as the frontier deferral - // above; the caches already describe wiped bytes. + Err( + error @ (partitions::PurgeError::OffsetCleanupNotRecorded(_) + | partitions::PurgeError::GenerationNotRecorded(_)), + ) => { + // Message reset is durable. The reconciler retries only + // cleanup and completion, preserving fresh message history. self.drop_partition_transfer_state(namespace, partition); tracing::warn!( shard = self.id, namespace_raw = namespace.inner(), generation, %error, - "purge-partition deferred: reset applied but the generation \ - record failed; the reconciler re-issues it" + "purge-partition deferred: message reset is durable but cleanup \ + completion must be retried" ); } Err(error @ partitions::PurgeError::Unserviceable(_)) => { - // Past the drain, so this group has no serviceable - // chain and the next append panics on - // `active_segment()`. Fence it for rebuild, exactly - // as a failed state-transfer convergence does. The - // counters were already reset to 0 before the - // fallible plant, so the fence's advancing write - // records the post-purge frontier. + // The reset lacks a usable chain or a durable boundary. + // Fence before new writes could be erased by a full retry. tracing::error!( shard = self.id, namespace_raw = namespace.inner(), diff --git a/core/simulator/src/storage/purge.rs b/core/simulator/src/storage/purge.rs index adb97dc600..d77f5bd1c6 100644 --- a/core/simulator/src/storage/purge.rs +++ b/core/simulator/src/storage/purge.rs @@ -31,17 +31,25 @@ //! purge completion and consumer recovery with `SimStorage`, then polls through //! the real `Next` path. Message recovery is narrower than server boot: a helper //! replays a durable journal into a new partition. No partition memory survives -//! recovery. These controls do not inject sync failures or exercise purge retries. +//! recovery. Failure cases inject an error before either offset directory sync +//! takes effect. They check both the immediate completion contract and recovery +//! with fresh history. Recovery finishes pending cleanup before loading bookmarks. +//! Partition tests cover full purge retries. This fixture does not model client +//! acknowledgments. + +use std::cell::Cell; +use std::io; +use std::path::{Path, PathBuf}; +use std::rc::Rc; +use std::sync::Arc; -use super::tests::owned_prepare; -use super::{Crash, SimStorage}; use configs::server::ServerConfig; use consensus::{LocalPipeline, Sequencer, VsrConsensus}; use futures::executor::block_on; use iggy_common::{ ConsumerKind, Durability, IggyByteSize, PartitionStats, PollingStrategy, TopicRuntimeOptions, }; -use journal::durable_storage::{DurableFile, DurableStorage, OpenMode}; +use journal::durable_storage::{DurableFile, DurableStorage, OpenMode, RegularFiles, StorageEntry}; use journal::{DurableAppend, PartitionPrepareJournal}; use message_bus::IggyMessageBus; use partitions::offset_storage::{ @@ -49,17 +57,18 @@ use partitions::offset_storage::{ }; use partitions::{ IggyPartition, IggyPartitions, Partition, PartitionPathLayout, PartitionsConfig, PollingArgs, - PollingConsumer, + PollingConsumer, PurgeError, }; use server::configure_consumer_offsets_with_storage; use server_common::send_messages::decode_batch_slice; use server_common::sharding::{IggyNamespace, ShardId}; -use std::path::{Path, PathBuf}; -use std::rc::Rc; -use std::sync::Arc; + +use super::tests::owned_prepare; +use super::{Crash, SimFile, SimStorage}; const CREATED_REVISION: u64 = 7; const OLD_GENERATION: u64 = 4; +const OLD_LAST_OPERATION: u64 = 3; const NEW_GENERATION: u64 = 5; const STORED_OFFSET: u64 = 2; const CONSUMER_ID: usize = 7; @@ -69,6 +78,360 @@ const FRESH_MESSAGE_COUNT: u64 = 5; type TestPartition = IggyPartition>; +async fn purge_with_failed_directory_sync_leaves_generation_pending( + policy: Durability, + failed_kind: ConsumerKind, +) { + let harness = PurgeStorageHarness::with_stored_progress(policy).await; + let mut partition = harness.partition_with_stored_progress().await; + + let purge_result = harness + .complete_purge_with_failed_directory_sync(&mut partition, failed_kind) + .await; + let live_generation = partition.applied_purge_generation(); + let stored_generation = harness.stored_purge_generation().await; + + assert!( + purge_result.is_err(), + "directory sync failure must reject purge; got {purge_result:?}, \ + live generation {live_generation}, stored generation {stored_generation}" + ); + assert_eq!( + live_generation, OLD_GENERATION, + "failed purge must not advance the live generation" + ); + assert_eq!( + stored_generation, OLD_GENERATION, + "failed purge must not record completion in storage" + ); +} + +async fn purge_sync_failure_preserves_recovery_contract_after_power_loss( + policy: Durability, + failed_kind: ConsumerKind, +) { + let harness = PurgeStorageHarness::with_stored_progress(policy).await; + let mut partition = harness.partition_with_stored_progress().await; + + let purge_result = harness + .complete_purge_with_failed_directory_sync(&mut partition, failed_kind) + .await; + drop(partition); + + // Bookmark 2 is still valid within fresh messages 0 through 4. + // Recovery cannot hide a restored bookmark by clamping it. + // Persist the messages without syncing the offset directories again. + harness.persist_fresh_history().await; + harness.storage.crash(Crash::PowerLoss); + let restored = harness + .storage + .entries(&harness.offset_directory(failed_kind)) + .await + .unwrap(); + assert!( + !restored.is_empty(), + "power loss must restore the unsynced bookmarks" + ); + let recovered = harness.recover_partition().await; + let recovered_generation = recovered.applied_purge_generation(); + assert_eq!(recovered.purge_floor_op(), OLD_LAST_OPERATION); + let polls = harness.poll_next(recovered).await; + + // Both real Next polls must finish before regression assertions, + // so a failure shows the results for the consumer and the group. + assert!(matches!( + purge_result, + Err(PurgeError::OffsetCleanupNotRecorded(_)) + )); + assert_eq!( + recovered_generation, NEW_GENERATION, + "recovery must finish pending cleanup before loading bookmarks; polls {polls:?}" + ); + for poll in &polls { + assert_eq!( + poll.stored_offset, None, + "recovery must remove restored bookmarks; polls {polls:?}" + ); + assert_eq!( + poll.offsets, + [0, 1, 2, 3, 4], + "cleanup after power loss must preserve every fresh message; polls {polls:?}" + ); + } +} + +#[test] +fn given_replicated_consumer_sync_failure_when_completing_purge_should_keep_generation_pending() { + block_on(purge_with_failed_directory_sync_leaves_generation_pending( + Durability::Replicated, + ConsumerKind::Consumer, + )); +} + +#[test] +fn given_replicated_consumer_sync_failure_when_power_is_lost_should_resume_cleanup_without_skipping_fresh_messages() + { + block_on( + purge_sync_failure_preserves_recovery_contract_after_power_loss( + Durability::Replicated, + ConsumerKind::Consumer, + ), + ); +} + +#[test] +fn given_replicated_group_sync_failure_when_completing_purge_should_keep_generation_pending() { + block_on(purge_with_failed_directory_sync_leaves_generation_pending( + Durability::Replicated, + ConsumerKind::ConsumerGroup, + )); +} + +#[test] +fn given_replicated_group_sync_failure_when_power_is_lost_should_resume_cleanup_without_skipping_fresh_messages() + { + block_on( + purge_sync_failure_preserves_recovery_contract_after_power_loss( + Durability::Replicated, + ConsumerKind::ConsumerGroup, + ), + ); +} + +#[test] +fn given_persisted_consumer_sync_failure_when_completing_purge_should_keep_generation_pending() { + block_on(purge_with_failed_directory_sync_leaves_generation_pending( + Durability::Persisted, + ConsumerKind::Consumer, + )); +} + +#[test] +fn given_persisted_consumer_sync_failure_when_power_is_lost_should_resume_cleanup_without_skipping_fresh_messages() + { + block_on( + purge_sync_failure_preserves_recovery_contract_after_power_loss( + Durability::Persisted, + ConsumerKind::Consumer, + ), + ); +} + +#[test] +fn given_persisted_group_sync_failure_when_completing_purge_should_keep_generation_pending() { + block_on(purge_with_failed_directory_sync_leaves_generation_pending( + Durability::Persisted, + ConsumerKind::ConsumerGroup, + )); +} + +#[test] +fn given_persisted_group_sync_failure_when_power_is_lost_should_resume_cleanup_without_skipping_fresh_messages() + { + block_on( + purge_sync_failure_preserves_recovery_contract_after_power_loss( + Durability::Persisted, + ConsumerKind::ConsumerGroup, + ), + ); +} + +#[test] +fn given_pending_cleanup_when_recovery_sync_fails_should_refuse_stale_bookmarks() { + block_on(async { + for policy in [Durability::Replicated, Durability::Persisted] { + for failed_kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + let harness = PurgeStorageHarness::with_stored_progress(policy).await; + let mut partition = harness.partition_with_stored_progress().await; + assert!( + harness + .complete_purge_with_failed_directory_sync(&mut partition, failed_kind) + .await + .is_err() + ); + drop(partition); + harness.persist_fresh_history().await; + harness.storage.crash(Crash::PowerLoss); + + let mut recovering = harness.empty_partition(); + recovering + .hydrate_applied_purge_generation_with_storage(&harness.storage) + .await + .unwrap(); + let directory = harness.offset_directory(failed_kind); + let failing_storage = FailingDirectorySync::new(&harness.storage, &directory); + let recovery_result = configure_consumer_offsets_with_storage( + &failing_storage, + &mut recovering, + &harness.config, + harness.namespace, + FRESH_MESSAGE_COUNT - 1, + ) + .await; + + assert!(recovery_result.is_err(), "{policy:?}, {failed_kind:?}"); + assert_eq!(failing_storage.entries_at_failed_sync(), Some(0)); + assert_eq!(harness.stored_purge_generation().await, OLD_GENERATION); + for consumer in consumers() { + assert_eq!(recovering.get_consumer_offset(consumer), None); + } + drop(recovering); + + // Discard another failed boot: cleanup must still succeed later. + harness.storage.crash(Crash::PowerLoss); + let recovered = harness.recover_partition().await; + assert_eq!(recovered.applied_purge_generation(), NEW_GENERATION); + harness + .poll_next_and_assert_messages(recovered, &[0, 1, 2, 3, 4]) + .await; + } + } + }); +} + +/// After power loss, a new partition must load the consumer and group +/// bookmarks from storage. Both saved bookmarks are 2, so Next must +/// return messages 3 and 4. +#[test] +fn given_stored_progress_when_power_is_lost_should_recover_both_consumer_bookmarks() { + block_on(async { + for policy in [Durability::Replicated, Durability::Persisted] { + let harness = PurgeStorageHarness::with_stored_progress(policy).await; + harness.persist_fresh_history().await; + + harness.storage.crash(Crash::PowerLoss); + let recovered = harness.recover_partition().await; + + assert_eq!(recovered.applied_purge_generation(), OLD_GENERATION); + for consumer in consumers() { + assert_eq!(recovered.get_consumer_offset(consumer), Some(STORED_OFFSET)); + } + // Bookmark 2 means the first three messages were already consumed. + harness + .poll_next_and_assert_messages(recovered, &[3, 4]) + .await; + } + }); +} + +/// Purge must remove the consumer and group bookmarks before fresh messages arrive. +/// After power loss, a new partition must find no saved bookmarks, so Next returns +/// all fresh messages, 0 through 4. Restoring either old bookmark of 2 would skip +/// messages 0 through 2 even though that bookmark is still within the new history. +#[test] +fn given_completed_purge_when_power_is_lost_should_read_all_fresh_messages() { + block_on(async { + for policy in [Durability::Replicated, Durability::Persisted] { + // Start at the completion phase: message history has been reset, + // but the old consumer and group bookmarks still need to be cleared. + let harness = PurgeStorageHarness::with_stored_progress(policy).await; + let mut partition = harness.partition_with_stored_progress().await; + + // These files arrive after recovery, so only the production directory + // sweep can discover them. Both directories must be cleaned durably. + harness.persist_unloaded_bookmarks().await; + + // Check the purge's effects on the live partition before discarding it. + partition + .complete_purge_with_storage(&harness.storage, NEW_GENERATION) + .await + .expect("complete purge cleanup"); + harness.assert_live_purge_completed(&partition).await; + drop(partition); + + // Save messages 0 through 4 after purge, then simulate power loss. + // The new partition must load its messages and bookmarks from storage. + harness.persist_fresh_history().await; + harness.storage.crash(Crash::PowerLoss); + let recovered = harness.recover_partition().await; + + harness.assert_recovered_purge_completed(&recovered); + harness + .poll_next_and_assert_messages(recovered, &[0, 1, 2, 3, 4]) + .await; + } + }); +} + +/// Inject one directory sync failure without changing the shared filesystem model. +/// The observation records the directory entry count before its sync is rejected. +struct FailingDirectorySync<'a> { + storage: &'a SimStorage, + failed_directory: &'a Path, + entries_at_failure: Cell>, +} + +impl<'a> FailingDirectorySync<'a> { + const fn new(storage: &'a SimStorage, failed_directory: &'a Path) -> Self { + Self { + storage, + failed_directory, + entries_at_failure: Cell::new(None), + } + } + + const fn entries_at_failed_sync(&self) -> Option { + self.entries_at_failure.get() + } +} + +impl DurableStorage for FailingDirectorySync<'_> { + type File = SimFile; + + fn writer_identity(&self, path: &Path) -> io::Result> { + self.storage.writer_identity(path) + } + + async fn open(&self, path: &Path, mode: OpenMode) -> io::Result { + self.storage.open(path, mode).await + } + + async fn create_directories(&self, path: &Path) -> io::Result<()> { + self.storage.create_directories(path).await + } + + async fn sync_directory(&self, path: &Path) -> io::Result<()> { + if path == self.failed_directory && self.entries_at_failure.get().is_none() { + let entries = self.storage.entries(path).await?; + self.entries_at_failure.set(Some(entries.len())); + return Err(io::Error::other("injected directory sync failure")); + } + self.storage.sync_directory(path).await + } + + async fn rename(&self, source: &Path, target: &Path) -> io::Result<()> { + self.storage.rename(source, target).await + } + + async fn remove_file(&self, path: &Path) -> io::Result<()> { + self.storage.remove_file(path).await + } + + async fn hard_link(&self, source: &Path, target: &Path) -> io::Result<()> { + self.storage.hard_link(source, target).await + } + + async fn exists(&self, path: &Path) -> io::Result { + self.storage.exists(path).await + } + + async fn exists_following_links(&self, path: &Path) -> io::Result { + self.storage.exists_following_links(path).await + } + + async fn entries(&self, path: &Path) -> io::Result> { + self.storage.entries(path).await + } + + async fn regular_files(&self, path: &Path) -> io::Result { + self.storage.regular_files(path).await + } + + async fn remove_tree(&self, path: &Path) -> io::Result<()> { + self.storage.remove_tree(path).await + } +} + /// Own the simulated filesystem and configuration, but no live partition state. /// /// Offset files and the purge marker use the server's directory layout. The @@ -77,10 +440,17 @@ struct PurgeStorageHarness { storage: SimStorage, config: ServerConfig, namespace: IggyNamespace, - /// Policy for consumer offsets; message durability is established by the fixture. + /// Policy for consumer offsets; the fixture persists and replays the message journal. policy: Durability, } +#[derive(Debug)] +struct ConsumerPoll { + consumer: PollingConsumer, + stored_offset: Option, + offsets: Vec, +} + impl PurgeStorageHarness { /// Store bookmark 2 for a consumer and a group, plus the earlier purge marker. /// All files and their directory entries are durable before the test starts, @@ -123,6 +493,119 @@ impl PurgeStorageHarness { harness } + /// Enter purge completion after message reset, with the old bookmarks loaded. + async fn partition_with_stored_progress(&self) -> TestPartition { + let policy = self.policy; + let mut partition = self.empty_partition(); + self.recover_progress(&mut partition, STORED_OFFSET).await; + assert_eq!( + partition.applied_purge_generation(), + OLD_GENERATION, + "{policy:?}: setup must load the earlier purge marker" + ); + for consumer in consumers() { + assert_eq!( + partition.get_consumer_offset(consumer), + Some(STORED_OFFSET), + "{policy:?}, {consumer:?}: setup must load the old bookmark" + ); + } + partition + } + + /// Reject the selected sync after unlinking, and verify that the fault was reached. + async fn complete_purge_with_failed_directory_sync( + &self, + partition: &mut TestPartition, + failed_kind: ConsumerKind, + ) -> Result<(), PurgeError> { + let failed_directory = self.offset_directory(failed_kind); + let failing_storage = FailingDirectorySync::new(&self.storage, &failed_directory); + let purge_result = partition + .complete_purge_with_storage(&failing_storage, NEW_GENERATION) + .await; + assert_eq!( + failing_storage.entries_at_failed_sync(), + Some(0), + "the selected directory sync must fail after its bookmark files were unlinked" + ); + purge_result + } + + async fn stored_purge_generation(&self) -> u64 { + let mut reloaded = self.empty_partition(); + reloaded + .hydrate_applied_purge_generation_with_storage(&self.storage) + .await + .unwrap(); + reloaded.applied_purge_generation() + } + + async fn persist_unloaded_bookmarks(&self) { + for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + self.persist_bookmark(kind, STRAY_ID).await; + } + } + + async fn assert_live_purge_completed(&self, partition: &TestPartition) { + let policy = self.policy; + assert_eq!( + partition.applied_purge_generation(), + NEW_GENERATION, + "{policy:?}: purge must advance the applied generation" + ); + for consumer in consumers() { + assert_eq!( + partition.get_consumer_offset(consumer), + None, + "{policy:?}, {consumer:?}: purge must clear the live bookmark" + ); + } + for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + assert_eq!( + partition.durable_consumer_offset_count(kind), + 0, + "{policy:?}, {kind:?}: purge must clear durability tracking" + ); + let directory = self.offset_directory(kind); + let remaining_paths: Vec<_> = self + .storage + .entries(&directory) + .await + .unwrap() + .into_iter() + .map(|entry| directory.join(entry.name)) + .collect(); + assert!( + remaining_paths.is_empty(), + "{policy:?}, {kind:?}: purge left bookmark files: {remaining_paths:?}" + ); + } + } + + fn assert_recovered_purge_completed(&self, recovered: &TestPartition) { + let policy = self.policy; + assert_eq!( + recovered.applied_purge_generation(), + NEW_GENERATION, + "{policy:?}: the completed purge marker must survive power loss" + ); + for consumer in consumers() { + assert_eq!( + recovered.get_consumer_offset(consumer), + None, + "{policy:?}, {consumer:?}: a deleted bookmark must not return after power loss" + ); + } + for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + assert_eq!( + recovered.durable_consumer_offset_count(kind), + 0, + "{policy:?}, {kind:?}: recovery must find no durable bookmarks" + ); + } + } + /// Persist bookmark 2 without adding it to any partition's live offset maps. /// This also creates stray files that the purge must discover by scanning. async fn persist_bookmark(&self, kind: ConsumerKind, consumer_id: usize) { @@ -159,9 +642,11 @@ impl PurgeStorageHarness { ) .await .unwrap(); + // The fixture starts after the old history, whose last operation was 3. + journal.reset(OLD_LAST_OPERATION, Some(0)).await.unwrap(); let mut parent_checksum = 0; for offset in 0..FRESH_MESSAGE_COUNT { - let operation_number = offset + 1; + let operation_number = OLD_LAST_OPERATION + offset + 1; // The helper fills the payload with the operation number, allowing // polls to verify message contents as well as their offsets. let prepare = owned_prepare(operation_number, parent_checksum, offset); @@ -192,6 +677,10 @@ impl PurgeStorageHarness { .await .unwrap(); let mut partition = self.empty_partition(); + partition + .hydrate_applied_purge_generation_with_storage(&self.storage) + .await + .unwrap(); for prepare in journal.prepares().await.unwrap() { let operation_number = prepare.header().op; partition.append_messages(prepare).await.unwrap(); @@ -227,17 +716,29 @@ impl PurgeStorageHarness { .unwrap(); } - /// Poll `Next` for the consumer and group, checking actual offsets and payloads. - /// This executes and completes real polls with automatic offset commits disabled; - /// it consumes the partition because completing polls can update live tracking. + /// Poll both consumers before asserting so a failure cannot hide the second result. async fn poll_next_and_assert_messages( &self, partition: TestPartition, expected_offsets: &[u64], ) { + for poll in Box::pin(self.poll_next(partition)).await { + assert_eq!( + poll.offsets, expected_offsets, + "{:?}, {:?}", + self.policy, poll.consumer + ); + } + } + + /// Complete real polls with automatic commits disabled, retaining each saved + /// bookmark for diagnostics. Poll completion can still update live tracking. + async fn poll_next(&self, partition: TestPartition) -> Vec { + let stored_offsets = consumers().map(|consumer| partition.get_consumer_offset(consumer)); let partitions = IggyPartitions::new(ShardId::new(0), partition_config()); partitions.insert(self.namespace, partition); - for consumer in consumers() { + let mut polls = Vec::with_capacity(consumers().len()); + for (consumer, stored_offset) in consumers().into_iter().zip(stored_offsets) { let plan = partitions .build_poll_snapshot( &self.namespace, @@ -254,24 +755,26 @@ impl PurgeStorageHarness { .complete_poll(&self.namespace, plan.execute().await) .unwrap(); assert!(completion.replication.is_none()); - let actual_offsets: Vec<_> = completion + let offsets = completion .fragments .iter() .map(|fragment| { let batch = decode_batch_slice(fragment.as_slice()).unwrap(); assert_eq!(batch.message_count(), 1); let message = batch.iter().next().unwrap(); - let expected_byte = u8::try_from(batch.header.base_offset + 1).unwrap(); + let expected_byte = + u8::try_from(OLD_LAST_OPERATION + batch.header.base_offset + 1).unwrap(); assert!(message.payload.iter().all(|byte| *byte == expected_byte)); batch.header.base_offset }) .collect(); - assert_eq!( - actual_offsets, expected_offsets, - "{:?}, {consumer:?}", - self.policy - ); + polls.push(ConsumerPoll { + consumer, + stored_offset, + offsets, + }); } + polls } /// Create a partition with no recovered messages or consumer progress. @@ -280,12 +783,14 @@ impl PurgeStorageHarness { let consensus = VsrConsensus::new( 1, 0, - 1, + 3, self.namespace.inner(), Rc::new(IggyMessageBus::new(0)), LocalPipeline::new(), ); consensus.init(); + consensus.sequencer().set_sequence(OLD_LAST_OPERATION); + consensus.restore_commit_state(OLD_LAST_OPERATION, OLD_LAST_OPERATION); let mut partition = IggyPartition::with_in_memory_storage( Arc::new(PartitionStats::default()), consensus, @@ -294,7 +799,9 @@ impl PurgeStorageHarness { partition.set_partition_dir(self.partition_directory()); partition.set_created_revision(CREATED_REVISION); partition.set_runtime_options(TopicRuntimeOptions { - durability: Durability::Replicated, + // The fixture restores a durable journal. Losing a volatile journal + // instead requires consensus recovery before this replay is safe. + durability: Durability::Persisted, consumer_offset_durability: self.policy, ..TopicRuntimeOptions::default() }); @@ -337,133 +844,3 @@ fn partition_config() -> PartitionsConfig { path_layout: PartitionPathLayout::default(), } } - -/// After power loss, a new partition must load the consumer and group -/// bookmarks from storage. Both saved bookmarks are 2, so Next must -/// return messages 3 and 4. -#[test] -fn given_stored_progress_when_power_is_lost_should_recover_both_consumer_bookmarks() { - block_on(async { - for policy in [Durability::Replicated, Durability::Persisted] { - let harness = PurgeStorageHarness::with_stored_progress(policy).await; - harness.persist_fresh_history().await; - - harness.storage.crash(Crash::PowerLoss); - let recovered = harness.recover_partition().await; - - assert_eq!(recovered.applied_purge_generation(), OLD_GENERATION); - for consumer in consumers() { - assert_eq!(recovered.get_consumer_offset(consumer), Some(STORED_OFFSET)); - } - // Bookmark 2 means the first three messages were already consumed. - harness - .poll_next_and_assert_messages(recovered, &[3, 4]) - .await; - } - }); -} - -/// Purge must remove the consumer and group bookmarks before fresh messages arrive. -/// After power loss, a new partition must find no saved bookmarks, so Next returns -/// all fresh messages, 0 through 4. Restoring either old bookmark of 2 would skip -/// messages 0 through 2 even though that bookmark is still within the new history. -#[test] -fn given_completed_purge_when_power_is_lost_should_read_all_fresh_messages() { - block_on(async { - for policy in [Durability::Replicated, Durability::Persisted] { - // Start at the completion phase: message history has been reset, - // but the old consumer and group bookmarks still need to be cleared. - let harness = PurgeStorageHarness::with_stored_progress(policy).await; - let mut partition = harness.empty_partition(); - harness - .recover_progress(&mut partition, STORED_OFFSET) - .await; - assert_eq!( - partition.applied_purge_generation(), - OLD_GENERATION, - "{policy:?}: setup must load the earlier purge marker" - ); - for consumer in consumers() { - assert_eq!( - partition.get_consumer_offset(consumer), - Some(STORED_OFFSET), - "{policy:?}, {consumer:?}: setup must load the old bookmark" - ); - } - - // These files arrive after recovery, so only the production directory - // sweep can discover them. Both directories must be cleaned durably. - for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { - harness.persist_bookmark(kind, STRAY_ID).await; - } - - // Check the purge's effects on the live partition before discarding it. - partition - .complete_purge_with_storage(&harness.storage, NEW_GENERATION) - .await - .expect("complete purge cleanup"); - assert_eq!( - partition.applied_purge_generation(), - NEW_GENERATION, - "{policy:?}: purge must advance the applied generation" - ); - for consumer in consumers() { - assert_eq!( - partition.get_consumer_offset(consumer), - None, - "{policy:?}, {consumer:?}: purge must clear the live bookmark" - ); - } - for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { - assert_eq!( - partition.durable_consumer_offset_count(kind), - 0, - "{policy:?}, {kind:?}: purge must clear durability tracking" - ); - let directory = harness.offset_directory(kind); - let remaining_paths: Vec<_> = harness - .storage - .entries(&directory) - .await - .unwrap() - .into_iter() - .map(|entry| directory.join(entry.name)) - .collect(); - assert!( - remaining_paths.is_empty(), - "{policy:?}, {kind:?}: purge left bookmark files: {remaining_paths:?}" - ); - } - drop(partition); - - // Save messages 0 through 4 after purge, then simulate power loss. - // The new partition must load its messages and bookmarks from storage. - harness.persist_fresh_history().await; - harness.storage.crash(Crash::PowerLoss); - let recovered = harness.recover_partition().await; - - assert_eq!( - recovered.applied_purge_generation(), - NEW_GENERATION, - "{policy:?}: the completed purge marker must survive power loss" - ); - for consumer in consumers() { - assert_eq!( - recovered.get_consumer_offset(consumer), - None, - "{policy:?}, {consumer:?}: a deleted bookmark must not return after power loss" - ); - } - for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { - assert_eq!( - recovered.durable_consumer_offset_count(kind), - 0, - "{policy:?}, {kind:?}: recovery must find no durable bookmarks" - ); - } - harness - .poll_next_and_assert_messages(recovered, &[0, 1, 2, 3, 4]) - .await; - } - }); -}