This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/durable-protocol-and-sessions in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 56cc05a969aeaf15ff7673f6c44890af9d5a4e6d Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 6 10:38:06 2026 +0200 fix(partitions): resume commits after persistence completes Replicated sends can wait for the next consensus tick after their body write completes. Resume eligible commits from the completion path while preserving recovery and durability gates. Cover replies without a tick and correct the WAL configuration comments for both durability policies. --- core/partitions/src/iggy_partition.rs | 147 +++++++++++++++++++++++++++++++++- core/server/config.toml | 22 ++--- core/shard/src/router.rs | 7 +- 3 files changed, 162 insertions(+), 14 deletions(-) diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index fe210cb82..43287b2a9 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -1109,7 +1109,11 @@ where } } - pub async fn on_persistence_completed(&mut self, completion: PersistenceCompletion) { + pub async fn on_persistence_completed( + &mut self, + completion: PersistenceCompletion, + config: &PartitionsConfig, + ) { if !self .persistence .as_ref() @@ -1129,6 +1133,12 @@ where warn!(%error, namespace_raw = self.namespace().inner(), "cannot reclaim obsolete receipt checkpoints"); } } + if self.consensus.is_normal() + && !self.consensus.is_transferring() + && self.consensus.commit_min() < self.consensus.commit_max() + { + self.commit_journal(config).await; + } } pub fn needs_persistence_checkpoint(&self) -> bool { @@ -9869,6 +9879,137 @@ mod tests { } } + #[compio::test] + async fn given_a_replicated_singleton_when_a_flush_write_completes_should_reply_without_a_tick() + { + let directory = tempfile::tempdir().unwrap(); + let (mut partition, replies) = recording_partition_at(0, 1); + partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); + partition.runtime_options.durability = iggy_common::Durability::Replicated; + partition.runtime_options.consumer_offset_durability = iggy_common::Durability::Replicated; + partition.open_persistence().await.unwrap(); + let config = repair_config(); + partition.log.retire_front().unwrap(); + partition.install_empty_segment(&config, 0).await.unwrap(); + let persistence = Rc::clone(partition.persistence.as_ref().unwrap()); + let completion = Rc::new(Cell::new(None)); + let captured = Rc::clone(&completion); + persistence.set_notifier(Rc::new(move |value| captured.set(Some(value)))); + + // Hold the writer until the self-ack has reached the physical-write gate. + persistence.sync(); + assert!(persistence.start()); + let writer = Rc::clone(&persistence).run(); + let mut request = checksumless_send_request(partition.namespace(), 2); + let body = &mut request.as_mut_slice()[size_of::<RoutedRequestHeader>()..]; + let mut batch = BatchHeader::decode(body).unwrap(); + batch.batch_checksum = batch.checksum_for_blob(&body[COMMAND_HEADER_SIZE..]); + batch.encode_into(&mut body[..COMMAND_HEADER_SIZE]); + partition.on_request(request.clone(), None).await; + assert_eq!( + partition.consensus.pipeline_len(), + 1, + "the send must be admitted: {:?}", + replies + .borrow() + .iter() + .map(|(_, reply)| bytemuck::checked::from_bytes::<ReplyHeader>( + &reply.as_slice()[..size_of::<ReplyHeader>()] + ) + .status) + .collect::<Vec<_>>() + ); + commit_recorded_loopback(&mut partition).await; + let header = partition.consensus.pipeline_head_header().unwrap(); + assert_eq!(partition.consensus.commit_max(), header.op); + assert_eq!(partition.consensus.commit_min(), 0); + assert!(!persistence.is_written(&header)); + assert!(partition.pending_persisted_acks.borrow().is_empty()); + assert!(replies.borrow().is_empty()); + + writer.await; + assert!(persistence.failure().is_none()); + assert!(persistence.is_written(&header)); + assert_eq!(persistence.durable_op(), 0); + assert!(replies.borrow().is_empty()); + let completed = completion + .get() + .expect("the buffered write must notify its owner"); + for stale in [ + PersistenceCompletion { + instance: completed.instance + 1, + ..completed + }, + PersistenceCompletion { + epoch: completed.epoch + 1, + ..completed + }, + ] { + partition.on_persistence_completed(stale, &config).await; + assert!(replies.borrow().is_empty()); + assert_eq!(partition.consensus.commit_min(), 0); + } + + partition.materialization_missing = true; + partition.on_persistence_completed(completed, &config).await; + assert!(replies.borrow().is_empty()); + assert_eq!(partition.consensus.commit_min(), 0); + partition.materialization_missing = false; + + partition.consensus().begin_state_transfer_await(); + partition.on_persistence_completed(completed, &config).await; + assert!(replies.borrow().is_empty()); + assert_eq!(partition.consensus.commit_min(), 0); + partition + .consensus() + .set_state_transfer_stage(consensus::StateTransferStage::Idle); + + partition.consensus().begin_view_probe(); + partition.on_persistence_completed(completed, &config).await; + assert!(replies.borrow().is_empty()); + assert_eq!(partition.consensus.commit_min(), 0); + partition.consensus().init(); + assert_eq!(partition.consensus.commit_max(), header.op); + assert_eq!(partition.consensus.pipeline_len(), 1); + + partition.on_persistence_completed(completed, &config).await; + + assert!(partition.fatal().is_none()); + assert_eq!( + partition.consensus.commit_min(), + header.op, + "a completed flush must release the self-acked reply without a consensus tick" + ); + assert_eq!(partition.consensus.pipeline_len(), 0); + assert_eq!(replies.borrow().len(), 1); + assert_eq!(partition.stats.messages_count_inconsistent(), 2); + assert_eq!(partition.offset.load(Ordering::Acquire), 1); + let consensus::client_table::RequestStatus::Duplicate(cached) = partition + .dedup + .check_partition_request(1, 0, 1, 1, Operation::SendMessages) + else { + panic!("the completed send must retain its original receipt"); + }; + assert!( + replies.borrow()[0] + .1 + .shares_allocation(&cached.into_wire_bytes()) + ); + + partition.on_persistence_completed(completed, &config).await; + assert_eq!(replies.borrow().len(), 1); + partition.on_request(request, None).await; + assert_eq!(replies.borrow().len(), 2); + assert!( + replies.borrow()[0] + .1 + .shares_allocation(&replies.borrow()[1].1) + ); + assert_eq!(partition.stats.messages_count_inconsistent(), 2); + assert_eq!(partition.offset.load(Ordering::Acquire), 1); + assert_eq!(persistence.head(), header.op); + } + #[compio::test] async fn mixed_durability_replies_before_body_writes_below_the_flush_threshold() { let directory = tempfile::tempdir().unwrap(); @@ -11166,7 +11307,9 @@ mod tests { assert!(current.exists()); let completion = completion.get().unwrap(); assert!(persistence.accepts_completion(completion)); - partition.on_persistence_completed(completion).await; + partition + .on_persistence_completed(completion, &config) + .await; assert!(!previous.exists()); assert!( current.exists(), diff --git a/core/server/config.toml b/core/server/config.toml index 6caaaf5bf..2397cac0f 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -893,9 +893,10 @@ partitions_max = 0 # Topic durability and flush scheduling are set at creation and returned by GetTopic. # durability controls message completion. consumer_offset_durability controls explicit offset stores and deletes. # Both independently default to "replicated". Neither inherits the other. -# In replicated groups, either persisted policy enables WAL references to segment bodies, retained by hard link until reclamation. -# This includes replicated messages with persisted offsets. Full message payloads count against wal_bytes_max. -# Their barriers can also persist co-batched messages. Only replicated/replicated stays off this WAL. +# Every disk-backed partition uses a WAL for recovery and retries. +# WAL references retain segment bodies by hard link until reclamation. +# Full message payloads count against wal_bytes_max under both durability policies. +# Persistence barriers can also persist co-batched messages. # Persisted prepares are limited to 64 MiB including the consensus header. Oversized operations are rejected permanently. # Both policies write data to storage. "replicated" waits for VSR commit and application without an additional stable-storage barrier. # "persisted" also requires recoverable stable-storage copies at the replication quorum before explicit success. @@ -928,21 +929,22 @@ partitions_max = 0 # Fenced files remain available for diagnosis. Recovery must not replace an acknowledged durable history with an empty healthy partition. # Metadata and partition journals verify their own integrity and recovery frontiers independently of sparse indexes. -# Per-partition active WAL and queued/in-flight prepare budget when either -# topic policy is persisted. A 4 KiB multiple, from 128 MiB + 8 KiB to 4 GiB. +# Per-partition active WAL and queued/in-flight prepare budget under both +# durability policies. A 4 KiB multiple, from 128 MiB + 8 KiB to 4 GiB. # The minimum accommodates the retained commit-point prepare and a maximum-size successor. -# Checkpointing begins at half the byte budget or 1024 dirty files. +# Checkpointing starts at half the byte budget. +# Explicit requests or 1024 dirty files can trigger checkpointing sooner. # File barriers run with at most 16 concurrent operations. Temporary rewrites need extra disk # space. Lowering this budget preserves existing history and blocks new writes # until recovery and checkpointing reclaim enough space. -# Checkpoints synchronize changed files through the partition writer. Other -# partitions continue while this partition defers materialized commit application. +# Checkpoints synchronize changed files through the partition writer. Preparing +# indexes and receipts still awaits I/O in the owning shard pump. # The HTTP metrics endpoint exposes per-shard partition_wal_* byte gauges and # completed batch, prepare, checkpoint and error counters. # partition_wal_retained_bytes plus queued_bytes and in_flight_bytes measures budget usage; # partition_wal_disk_bytes measures WAL file length, excluding referenced segment bodies. -# partition_wal_prepares over partition_wal_batches is the prepares per durable group, and -# partition_wal_group_commit_waits counts the groups that actually took the optional wait +# partition_wal_prepares_total over partition_wal_batches_total is the prepares per durable group, and +# partition_wal_group_commit_waits_total counts the groups that actually took the optional wait # below, so a delay that never fires is distinguishable from one that fires and buys nothing. wal_bytes_max = "256 MiB" diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index e96b8dfaf..450044827 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -874,8 +874,11 @@ where } LifecycleFrame::PartitionPersistenceCompleted(completion) => { let namespace = IggyNamespace::from_raw(completion.group); - if let Some(partition) = self.plane.partitions().get_mut_by_ns(&namespace) { - partition.on_persistence_completed(completion).await; + let partitions = self.plane.partitions(); + if let Some(partition) = partitions.get_mut_by_ns(&namespace) { + partition + .on_persistence_completed(completion, partitions.config()) + .await; } } LifecycleFrame::ReconcileApply => {
