This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch perf/task-per-partition in repository https://gitbox.apache.org/repos/asf/iggy.git
commit bd92da9f94c445679d559a8c0ba2ba7187050166 Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Oct 7 23:00:13 2026 +0200 fix(partitions): resume shutdown after interrupted file jobs An interrupted file job could leave its shutdown owner asleep until the drain timed out. Notify the fenced owner without releasing its held resources, and avoid rescanning owners while file jobs remain. --- core/partitions/src/iggy_partition.rs | 39 ++++++++++++++++++++++++++++++++++- core/shard/src/router.rs | 3 ++- 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 5dfb69540..ed6fdf809 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -1957,7 +1957,7 @@ where if self.materialization_missing || !self.ensure_wal_view() { return; } - // The shard's bounded pre-pass owns superblock I/O across partitions. + // The file lane must persist the superblock before WAL acks can advance. if self.superblock.is_some() && self.consensus.needs_superblock_persist() { return; } @@ -5848,6 +5848,7 @@ where pub fn interrupt_io(&mut self, identity: crate::PartitionIoIdentity) { if self.io_active.get() == Some(identity) { self.fence_flush_failure(); + self.notify_io(); } } @@ -20172,6 +20173,42 @@ mod tests { assert_eq!(reply.status, IggyError::TransientNotAccepted.as_code()); } + #[compio::test] + async fn given_shutdown_waiting_for_io_when_interrupted_should_notify_and_keep_resources() { + let mut partition = partition_at_view(1, 1); + partition.set_superblock(Rc::new(RecordingSuperblock::default()), None); + let notifications = Rc::new(Cell::new(0)); + let observed = Rc::clone(¬ifications); + partition.set_io_notifier( + Rc::new(move |_, _| observed.set(observed.get() + 1)), + crate::largest_legal_job_charge().unwrap(), + ); + assert!(!partition.persist_superblock_if_needed().await); + let config = repair_config(); + let crate::PartitionIoStep::Ready(plan) = partition.resume_io(&config).await else { + panic!("view persistence must be ready"); + }; + let captured = partition.capture_io(plan, &config).unwrap().unwrap(); + + partition.begin_shutdown_io(); + assert!(matches!( + partition.resume_io(&config).await, + crate::PartitionIoStep::Pending + )); + assert!(!partition.shutdown_io_complete()); + notifications.set(0); + + captured.quiescence.interrupt(); + partition.interrupt_io(captured.identity); + assert!(notifications.get() > 0, "the shutdown owner must be ready"); + partition.resume_io(&config).await; + + assert!(partition.shutdown_io_complete()); + assert_eq!(partition.io_active.get(), Some(captured.identity)); + assert_eq!(captured.quiescence.get(), Some(captured.identity)); + assert!(partition.superblock_lock.try_acquire().is_none()); + } + #[compio::test] async fn shutdown_flushes_committed_messages_with_an_unacknowledged_pipeline_tail() { for (replica_count, store_offset) in [(3, false), (1, true), (3, true)] { diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 969cdec2b..f2022596d 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -588,12 +588,13 @@ where self.service_partition_io().await; self.process_loopback(loopback).await; let finished = !self.partition_io.has_ready() + && self.partition_io.outstanding() == 0 && namespaces.iter().all(|namespace| { partitions .get_io_owner(namespace) .is_none_or(|partition| partition.shutdown_io_complete()) }); - if finished && self.partition_io.outstanding() == 0 { + if finished { break; } futures::select_biased! {
