This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch fix_writer_lease_drain in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 39ddd71e35cb343e8b4d3f733129dc5e0a398bf1 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Oct 7 19:07:19 2026 +0200 fix(partitions): make writer-lease drain wait executor-agnostic --- core/journal/src/durable_storage.rs | 10 +++++++ core/partitions/src/persistence.rs | 60 +++++++++++++++++++------------------ core/simulator/src/storage.rs | 7 +++++ core/simulator/src/storage/tests.rs | 40 +++++++++++++++---------- 4 files changed, 72 insertions(+), 45 deletions(-) diff --git a/core/journal/src/durable_storage.rs b/core/journal/src/durable_storage.rs index 45ce7d432..5348e3a9c 100644 --- a/core/journal/src/durable_storage.rs +++ b/core/journal/src/durable_storage.rs @@ -29,6 +29,7 @@ use std::ffi::OsString; use std::io; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::time::Duration; use tokio::sync::{Semaphore, mpsc}; use tracing::warn; @@ -73,6 +74,15 @@ pub trait DurableStorage { Ok(None) } + /// Timer bounding a wait. The future must not resolve before `duration` + /// has elapsed, and one that never resolves disables the timeout. The + /// default needs a compio runtime and panics outside one. Only the + /// writer-lease drain wait on partition WAL reopen uses it, so overriding + /// it does not make other partition WAL timers executor-agnostic. + fn sleep(&self, duration: Duration) -> impl Future<Output = ()> { + compio::runtime::time::sleep(duration) + } + /// # Errors /// Returns the underlying filesystem error. fn open(&self, path: &Path, mode: OpenMode) -> impl Future<Output = io::Result<Self::File>>; diff --git a/core/partitions/src/persistence.rs b/core/partitions/src/persistence.rs index b10539f0a..32c61aad6 100644 --- a/core/partitions/src/persistence.rs +++ b/core/partitions/src/persistence.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use futures::TryStreamExt; +use futures::{FutureExt, TryStreamExt}; use iggy_binary_protocol::{Operation, PrepareHeader}; use iggy_common::ConsumerKind; use journal::PartitionPrepareJournal; @@ -234,7 +234,7 @@ static WRITERS: LazyLock<Mutex<HashMap<PathBuf, WriterRegistration>>> = LazyLock::new(|| Mutex::new(HashMap::new())); impl WriterLease { - async fn acquire(key: PathBuf) -> io::Result<Arc<Self>> { + async fn acquire<S: DurableStorage>(key: PathBuf, storage: &S) -> io::Result<Arc<Self>> { loop { let previous = { let mut writers = WRITERS @@ -281,32 +281,34 @@ impl WriterLease { "partition WAL already has an active writer", )); } - compio::runtime::time::timeout( - PERSISTENCE_DRAIN_TIMEOUT, - futures::future::poll_fn(|context| { - let mut waiters = previous - .waiters - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if !previous.running.load(Ordering::Acquire) { - return std::task::Poll::Ready(()); - } - if !waiters - .iter() - .any(|waiter| waiter.will_wake(context.waker())) - { - waiters.push(context.waker().clone()); - } - std::task::Poll::Pending - }), - ) - .await - .map_err(|_| { - io::Error::new( - io::ErrorKind::TimedOut, - "retired partition WAL writer did not stop", - ) - })?; + let drained = futures::future::poll_fn(|context| { + let mut waiters = previous + .waiters + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !previous.running.load(Ordering::Acquire) { + return std::task::Poll::Ready(()); + } + if !waiters + .iter() + .any(|waiter| waiter.will_wake(context.waker())) + { + waiters.push(context.waker().clone()); + } + std::task::Poll::Pending + }) + .fuse(); + let expired = storage.sleep(PERSISTENCE_DRAIN_TIMEOUT).fuse(); + futures::pin_mut!(drained, expired); + futures::select_biased! { + () = drained => {} + () = expired => { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "retired partition WAL writer did not stop", + )); + } + } let mut writers = WRITERS .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); @@ -562,7 +564,7 @@ impl<S: DurableStorage> PartitionPersistence<S> { ) })?; let lease = if let Some(key) = storage.writer_identity(partition_directory)? { - Some(WriterLease::acquire(key).await?) + Some(WriterLease::acquire(key, &storage).await?) } else { None }; diff --git a/core/simulator/src/storage.rs b/core/simulator/src/storage.rs index 441134971..1c0c922d5 100644 --- a/core/simulator/src/storage.rs +++ b/core/simulator/src/storage.rs @@ -28,6 +28,7 @@ use std::io; use std::path::{Component, Path}; use std::rc::Rc; use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; /// Each `SimStorage` is an independent filesystem, so writer identities must not /// collide between instances the way bare paths would. @@ -276,6 +277,12 @@ impl DurableStorage for SimStorage { )) } + // Unit-test storage driven by `block_on` with no virtual timer wired in, so + // a writer that never drains hangs the test instead of timing out. + fn sleep(&self, _duration: Duration) -> impl Future<Output = ()> { + futures::future::pending() + } + async fn open(&self, path: &Path, mode: OpenMode) -> io::Result<SimFile> { let creates = matches!( mode, diff --git a/core/simulator/src/storage/tests.rs b/core/simulator/src/storage/tests.rs index 68640ebe8..6e343536f 100644 --- a/core/simulator/src/storage/tests.rs +++ b/core/simulator/src/storage/tests.rs @@ -892,7 +892,6 @@ fn first_open_recovers_after_each_initialization_fault() { } #[test] -#[ignore = "PR #4092 review: PRE-EXISTING test. It passed vacuously while `SimStorage::writer_identity` returned `None` and never took a lease; now that the lease is real it is blocked on the compio-bound drain wait"] fn deleting_and_recreating_a_partition_fences_an_old_writer_completion() { block_on(async { let storage = storage_for_partition().await; @@ -919,15 +918,18 @@ fn deleting_and_recreating_a_partition_fences_an_old_writer_completion() { .await .unwrap(); storage.sync_directory(Path::new("/")).await.unwrap(); - let (new, _) = PartitionPersistence::open_with_storage( + let mut reopen = Box::pin(PartitionPersistence::open_with_storage( Path::new("/partition/prepares-8"), 42, 8, storage.clone(), - ) - .await - .unwrap(); - writer.await; + )); + assert!( + poll!(&mut reopen).is_pending(), + "a retired writer still mid-write must fence the recreated partition" + ); + let (reopened, ()) = futures::join!(reopen, writer); + let (new, _) = reopened.unwrap(); assert!(!old.is_durable(original.header())); let replacement = prepare_with_payload(1, 0, b"new incarnation"); new.append(replacement.clone().into_frozen(), true).unwrap(); @@ -3104,7 +3106,6 @@ async fn read_all(storage: &SimStorage, path: &Path) -> Vec<u8> { } #[test] -#[ignore = "PR #4092 review: `WriterLease::acquire` drains through `compio::runtime::time::timeout`, so the writer fence cannot be driven by the deterministic executor"] fn given_a_retired_writer_when_reacquiring_then_the_drain_wait_should_be_executor_agnostic() { block_on(async { let storage = storage_for_partition().await; @@ -3119,17 +3120,24 @@ fn given_a_retired_writer_when_reacquiring_then_the_drain_wait_should_be_executo assert!(poll!(&mut writer).is_pending()); first.retire(); - // `WriterLease::acquire` waits for the previous writer to drain through - // `compio::runtime::time::timeout` (`persistence.rs:220`), so the fence - // cannot be driven by the deterministic executor at all. Every - // simulator fault case runs with `lease = None` for this reason. - let reacquired = - PartitionPersistence::open_with_storage(Path::new(WAL), 42, 8, storage.clone()).await; + let mut reacquire = Box::pin(PartitionPersistence::open_with_storage( + Path::new(WAL), + 42, + 7, + storage.clone(), + )); + assert!( + poll!(&mut reacquire).is_pending(), + "a retired writer still mid-write must fence its replacement" + ); storage.resume(); - writer.await; + let (reacquired, ()) = futures::join!(reacquire, writer); + let (new, _) = reacquired.unwrap_or_else(|error| { + panic!("retired writer could not be replaced under a non-compio executor: {error}") + }); assert!( - reacquired.is_ok(), - "retired writer could not be replaced under the deterministic executor" + new.is_durable(prepare(1, 0).header()), + "replacement must recover the write the retired writer finished" ); }); }
