You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/07 09:46:57 UTC
(iggy) 02/10: fix(partitions): preserve durability and progress during file I/O
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 3bd7a222b6067b732e8c5c1d31f4f58ff1aa9917
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Oct 7 00:12:16 2026 +0200
fix(partitions): preserve durability and progress during file I/O
Failed purges could leave partially reset history serviceable, and
state transfer could reclaim the replacement receipt checkpoint.
File-job saturation also delayed unrelated replies, while shutdown
could wait beyond the WAL head.
Fence purge durability failures, cancel obsolete maintenance and
persist retirement barriers. Keep reply continuations runnable under
file-job saturation and validate commit headers in one journal pass.
Use local completion delivery, account for cleanup allocations and
align byte-limit configuration with other size settings. Add recovery
regressions and remove obsolete helpers.
---
core/configs/src/server_config/sharding.rs | 42 +-
core/consensus/src/impls.rs | 10 +-
core/consensus/src/plane_helpers.rs | 104 +--
core/journal/src/local_gate.rs | 11 +-
core/partitions/src/consumer_offset_capacity.rs | 25 +-
core/partitions/src/iggy_partition.rs | 1101 +++++++++++++----------
core/partitions/src/iggy_partitions.rs | 12 +-
core/partitions/src/io.rs | 20 +-
core/partitions/src/journal.rs | 111 ++-
core/partitions/src/lib.rs | 7 +-
core/partitions/src/offset_storage.rs | 6 +-
core/partitions/src/persistence.rs | 7 +-
core/partitions/src/state_transfer.rs | 89 +-
core/server/config.toml | 6 +-
core/server/src/boot/recovery.rs | 7 +-
core/server/src/boot/threads.rs | 6 +-
core/server/src/consumer_group.rs | 1 +
core/server/src/dispatch/mod.rs | 1 +
core/server/src/dispatch/partition.rs | 1 +
core/server_common/src/sharding/mod.rs | 3 +
core/shard/src/builder.rs | 7 +-
core/shard/src/lib.rs | 19 +-
core/shard/src/metrics.rs | 10 +-
core/shard/src/partition_io.rs | 439 +++++++--
core/shard/src/poll/completion_tests.rs | 1 +
core/shard/src/router.rs | 45 +-
core/simulator/src/replica.rs | 1 +
27 files changed, 1270 insertions(+), 822 deletions(-)
diff --git a/core/configs/src/server_config/sharding.rs b/core/configs/src/server_config/sharding.rs
index a4ea87b08..d8fd06bff 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -18,11 +18,12 @@
//! Sharding config: the full thread-per-core + bus surface, with
//! defaults read from the embedded `core/server/config.toml`.
-use iggy_common::IggyDuration;
-use iggy_common::Validatable;
+use std::time::Duration;
+
+use iggy_common::{IggyByteSize, IggyDuration, Validatable};
use serde::{Deserialize, Serialize};
use serde_with::{DisplayFromStr, serde_as};
-use std::time::Duration;
+use server_common::sharding::PARTITION_IO_CAPACITY_MAX;
use super::defaults::SERVER_CONFIG;
use crate::ConfigurationError;
@@ -104,8 +105,10 @@ pub struct ShardingConfig {
/// Active partition file jobs and results awaiting owner acceptance.
pub partition_io_capacity: usize,
/// Retained job allocations, resolved at boot against the largest legal record.
- /// Omission selects max(256 MiB, the single-job minimum).
- pub partition_io_bytes_max: Option<usize>,
+ /// Omission selects the single-job minimum, slightly above 4 GiB per shard.
+ /// This is an admission ceiling; memory is allocated only for admitted jobs.
+ #[config_env(leaf)]
+ pub partition_io_bytes_max: Option<IggyByteSize>,
/// Wall-clock budget for a single shard's bus drain on shutdown.
/// Drives `IggyMessageBus::shutdown(..)` from the per-shard watchdog
/// and the parallel-join survivor path. Sized larger than typical
@@ -182,14 +185,21 @@ impl Default for ShardingConfig {
impl Validatable<ConfigurationError> for ShardingConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
- if self.partition_io_capacity == 0
- || self.partition_io_capacity > INBOX_CAPACITY_MAX
- || self
- .partition_io_bytes_max
- .is_some_and(|bytes| bytes == 0 || bytes > isize::MAX as usize)
+ if self.partition_io_capacity == 0 || self.partition_io_capacity > PARTITION_IO_CAPACITY_MAX
+ {
+ eprintln!(
+ "Invalid sharding.partition_io_capacity: {} must be in 1..={PARTITION_IO_CAPACITY_MAX}",
+ self.partition_io_capacity
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if let Some(bytes) = self.partition_io_bytes_max
+ && (bytes.as_bytes_u64() == 0 || bytes.as_bytes_u64() > isize::MAX as u64)
{
eprintln!(
- "Invalid sharding configuration: partition I/O limits must be positive and fit addressable capacity"
+ "Invalid sharding.partition_io_bytes_max: {} bytes must be in 1..={}",
+ bytes.as_bytes_u64(),
+ isize::MAX
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
@@ -349,14 +359,14 @@ mod tests {
let configured: ShardingConfig = Figment::new()
.merge(Toml::string(
- "partition_io_capacity = 3\npartition_io_bytes_max = 8589934592",
+ "partition_io_capacity = 3\npartition_io_bytes_max = \"8 GiB\"",
))
.extract()
.unwrap();
assert_eq!(configured.partition_io_capacity, 3);
assert_eq!(
configured.partition_io_bytes_max,
- Some(8 * 1024 * 1024 * 1024)
+ Some(IggyByteSize::from(8 * 1024 * 1024 * 1024))
);
assert!(configured.validate().is_ok());
let mappings = <ShardingConfig as configs::ConfigEnvMappings>::env_mappings();
@@ -365,9 +375,9 @@ mod tests {
}
for (capacity, bytes) in [
(0, None),
- (INBOX_CAPACITY_MAX + 1, None),
- (1, Some(0)),
- (1, Some(usize::MAX)),
+ (PARTITION_IO_CAPACITY_MAX + 1, None),
+ (1, Some(IggyByteSize::from(0))),
+ (1, Some(IggyByteSize::from(u64::MAX))),
] {
let invalid = ShardingConfig {
partition_io_capacity: capacity,
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index f5a90aedc..c8d732825 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -633,11 +633,11 @@ impl LocalPipeline {
pub fn has_request_before(&self, order: LocalRequestOrder) -> bool {
self.prepare_queue
.iter()
- .any(|entry| entry.local_order.is_none_or(|entry| entry < order))
+ .any(|entry| entry.local_order.is_none_or(|admitted| admitted < order))
|| self
.request_queue
.iter()
- .any(|entry| entry.local_order.is_none_or(|entry| entry < order))
+ .any(|entry| entry.local_order.is_none_or(|admitted| admitted < order))
}
#[must_use]
@@ -5866,7 +5866,6 @@ mod vsr_consensus_tests {
consensus.pipeline_message(PlaneKind::Metadata, message);
}
- use crate::drain_committable_prefix;
use iggy_binary_protocol::Operation;
/// Clock frozen at a fixed instant, so a stamp read off it is assertable.
@@ -5926,8 +5925,7 @@ mod vsr_consensus_tests {
// Committing the head leaves op 2 in flight, so the timer stays armed --
// now measuring op 2 rather than carrying op 1's elapsed ticks.
consensus.advance_commit_max(1);
- assert_eq!(drain_committable_prefix(&consensus).len(), 1);
- // As real callers do, per entry: the next drain starts at the op now owed.
+ assert_eq!(consensus.pop_committed_prepare().unwrap().header.op, 1);
consensus.advance_commit_min(1);
assert!(
prepare_ticking(&consensus),
@@ -5935,7 +5933,7 @@ mod vsr_consensus_tests {
);
consensus.advance_commit_max(2);
- assert_eq!(drain_committable_prefix(&consensus).len(), 1);
+ assert_eq!(consensus.pop_committed_prepare().unwrap().header.op, 2);
consensus.advance_commit_min(2);
assert!(
!prepare_ticking(&consensus),
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index ff42cfb4e..d846c9914 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -501,91 +501,18 @@ where
|| (consensus.view_log_is_pending() && consensus.is_primary_for_view(consensus.view()))
}
-/// Drain and return committable prepares from the pipeline head.
-///
-/// Entries are drained from the head, while covered by the commit frontier, and
-/// only as a contiguous run starting at the next op owed to the state machine.
-///
-/// Callers `advance_commit_min` per entry, so a run starting above
-/// `commit_min + 1` or breaking partway hits that counter's sequential-advance
-/// assert. Pipeline-side twin of `commit_journal`'s gap-stop.
-///
-/// Reported, not asserted: a promoted partition primary reaches it legitimately
-/// while its journal walk clears the apply backlog. Repair arms on the level, and
-/// a hold that never clears fails the simulator's contiguity invariant.
-///
-/// Do NOT read that as "the journal walk always finishes". It can stall on the
-/// partition plane: `committed_headers_from` reads resident headers only, while
-/// `commit_messages` evicts up to `commit_max` (the cluster frontier), so ops past
-/// the per-call cap can be flushed out from under the walk. Pre-existing; see
-/// `IggyPartition::collect_committable_from_journal`.
-///
-/// A head at or below `commit_min` is the other shape: applied already, no repair
-/// owed, only a pop that can no longer happen. The journal walks stop below the
-/// head so they cannot create it, but a `set_commit_floor` jump past a live
-/// pipeline still can, so it is logged apart rather than treated as impossible.
-///
-/// # Panics
-/// If `head()` returns `Some` but `pop()` returns `None` (unreachable).
-pub fn drain_committable_prefix<B, P>(consensus: &VsrConsensus<B, P>) -> Vec<PipelineEntry>
-where
- B: MessageBus,
- P: Pipeline<Entry = PipelineEntry>,
-{
- let commit = consensus.commit_max();
- let commit_min = consensus.commit_min();
- let mut drained = Vec::new();
-
- consensus.with_pipeline_mut(|pipeline| {
- let mut next = commit_min + 1;
- while let Some(head_op) = pipeline.head().map(|entry| entry.header.op) {
- if head_op > commit {
- break;
- }
- if head_op != next {
- report_uncommittable_head(
- consensus.replica(),
- head_op,
- commit_min,
- commit,
- drained.len(),
- );
- break;
- }
-
- let entry = pipeline
- .pop()
- .expect("drain_committable_prefix: head exists");
- drained.push(entry);
- next += 1;
- }
- });
-
- // Popping through the pipeline directly bypasses
- // `VsrConsensus::pop_committed_prepare`, so re-establish the prepare
- // timeout's ticking-iff-non-empty invariant here: an emptied pipeline
- // disarms it, and a remaining head becomes the entry the timer measures,
- // timed from now rather than inheriting the drained entry's elapsed ticks.
- if !drained.is_empty() {
- consensus.sync_prepare_timeout();
- }
-
- drained
-}
-
/// Header of the pipeline head, iff its op is the next one this replica owes its
/// state machine and is covered by the commit frontier.
///
-/// Peek-only counterpart of [`drain_committable_prefix`] for commit paths that
-/// must survive their driving future being canceled between "committable" and
+/// Commit paths must survive their driving future being canceled between "committable" and
/// "applied" (see `IggyMetadata::on_ack`): the caller peeks here, performs its
-/// awaits with the entry still in the pipeline, then — in a sync region —
+/// awaits with the entry still in the pipeline, then synchronously
/// revalidates that the head is still this exact entry before popping and
/// applying it. A driver dropped at an await strands nothing; a sibling driver
/// that committed the op first fails the caller's revalidation and re-peeks.
///
-/// Bounded below for the reason [`drain_committable_prefix`] is, and reported
-/// rather than asserted for the same one. Holding is safe: a shard pump's panic is
+/// A head that is not `commit_min + 1` is held and reported, because applying it
+/// would violate sequential advancement. Holding is safe: a shard pump's panic is
/// swallowed by `compio::runtime::spawn`, while `tick_metadata` re-arms repair on
/// the level.
pub fn peek_committable_head<B, P>(consensus: &VsrConsensus<B, P>) -> Option<PrepareHeader>
@@ -2423,9 +2350,9 @@ mod tests {
}
/// `advance_commit_min` is strictly sequential, so draining op 7 with 6 never
- /// applied panics the shard pump. Both gates must hold instead.
+ /// applied panics the shard pump. The commit gate must hold instead.
#[test]
- fn given_a_hole_below_the_head_when_committing_should_hold_both_gates() {
+ fn given_a_hole_below_the_head_when_committing_should_hold_the_head() {
let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus, LocalPipeline::new());
consensus.init();
consensus.restore_commit_state(5, 5);
@@ -2439,10 +2366,6 @@ mod tests {
peek_committable_head(&consensus).is_none(),
"the head is covered by the frontier but op 6 is not applied"
);
- assert!(
- drain_committable_prefix(&consensus).is_empty(),
- "the drain must hold on the same hole the peek does"
- );
assert_eq!(
consensus.pipeline_head_header().map(|header| header.op),
Some(7),
@@ -2450,7 +2373,7 @@ mod tests {
);
}
- /// The shape the journal-walk caps prevent. Both gates must still refuse it:
+ /// The shape the journal-walk caps prevent. The gate must still refuse it:
/// re-applying an applied op panics `advance_commit_min`.
#[test]
fn given_an_applied_head_when_committing_should_refuse_it() {
@@ -2468,10 +2391,7 @@ mod tests {
peek_committable_head(&consensus).is_none(),
"op 5 is already applied; re-applying it panics advance_commit_min"
);
- assert!(
- drain_committable_prefix(&consensus).is_empty(),
- "the drain must refuse an applied head too"
- );
+ assert_eq!(consensus.pipeline_head_header().unwrap().op, 5);
}
#[test]
@@ -2486,8 +2406,12 @@ mod tests {
consensus.pipeline_message(PlaneKind::Metadata, &prepare_message(7, 60, 70));
consensus.advance_commit_max(6);
- let drained = drain_committable_prefix(&consensus);
- let drained_ops: Vec<_> = drained.into_iter().map(|entry| entry.header.op).collect();
+ let mut drained_ops = Vec::new();
+ while let Some(header) = peek_committable_head(&consensus) {
+ assert_eq!(consensus.pop_committed_prepare().unwrap().header, header);
+ consensus.advance_commit_min(header.op);
+ drained_ops.push(header.op);
+ }
assert_eq!(drained_ops, vec![5, 6]);
assert_eq!(
diff --git a/core/journal/src/local_gate.rs b/core/journal/src/local_gate.rs
index e5a716df1..3e65f8758 100644
--- a/core/journal/src/local_gate.rs
+++ b/core/journal/src/local_gate.rs
@@ -24,15 +24,16 @@
//!
//! Not a general lock: single-threaded (`Cell`/`RefCell`, never `Sync`),
//! release wakes every waiter and poll order re-races (arrival-order FIFO
-//! under `futures::join!`-style drivers), cancel-safe (dropping the guard
-//! releases; dropping a waiter leaves only a stale waker). Non-reentrant: a
-//! holder that re-acquires deadlocks itself.
+//! under `futures::join!`-style drivers). Dropping a borrowed guard releases;
+//! dropping a waiter leaves only a stale waker. An owned lease requires explicit
+//! release after physical completion: dropping it keeps the resource fenced.
+//! Non-reentrant: a holder that re-acquires deadlocks itself.
use std::cell::{Cell, RefCell};
use std::rc::Rc;
-/// See the module docs. Callers hold the returned guard across the awaited
-/// critical section; dropping it releases the gate and wakes every waiter.
+/// See the module docs. Borrowed guards release on drop; owned leases release
+/// explicitly after physical completion and otherwise keep the gate closed.
///
/// Its exclusion is load-bearing in RELEASE, not only under
/// `debug_assertions`: this gate is the only enforcement of the superblock
diff --git a/core/partitions/src/consumer_offset_capacity.rs b/core/partitions/src/consumer_offset_capacity.rs
index d1cf35e34..193a4cd0e 100644
--- a/core/partitions/src/consumer_offset_capacity.rs
+++ b/core/partitions/src/consumer_offset_capacity.rs
@@ -309,10 +309,8 @@ impl ConsumerOffsetCapacity {
.is_some_and(|token| token.active_count() > 0)
}
- /// Assigns the pending count outright while [`Self::release_reservation`]
- /// decrements it. Both take `&self` and neither locks: they are serialized
- /// by their call sites, which all run under the partition's `&mut self` on
- /// its own shard thread.
+ /// Assign the pending count under the partition's exclusive shard-local
+ /// ownership. Callers derive it from the remaining admitted operations.
pub(crate) fn set_pending_count(&self, id: u32, count: usize) {
if count == 0 {
if self.pending.borrow_mut().remove(&id).is_some() {
@@ -323,21 +321,6 @@ impl ConsumerOffsetCapacity {
}
}
- /// See [`Self::set_pending_count`] for the serialization contract.
- #[cfg(test)]
- pub(crate) fn release_reservation(&self, id: u32) {
- let mut pending = self.pending.borrow_mut();
- let Some(count) = pending.get_mut(&id) else {
- return;
- };
- if *count == 1 {
- pending.remove(&id);
- self.note_local_key_change();
- } else {
- *count -= 1;
- }
- }
-
pub(crate) const fn is_uncertain(&self) -> bool {
self.uncertain.get()
}
@@ -616,12 +599,12 @@ mod tests {
assert_eq!(capacity.try_reserve(7, &durable), Ok(()));
assert_eq!(capacity.try_reserve(7, &durable), Ok(()));
assert!(capacity.try_reserve(8, &durable).is_err());
- capacity.release_reservation(7);
+ capacity.set_pending_count(7, 1);
assert!(
capacity.try_reserve(8, &durable).is_err(),
"one of two reservations still owns the slot"
);
- capacity.release_reservation(7);
+ capacity.set_pending_count(7, 0);
assert!(
capacity.try_reserve(8, &durable).is_ok(),
"the slot is released after the last reservation"
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index bae775ebf..4a5845975 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -649,8 +649,8 @@ enum PurgePhase {
Cleanup,
Plant,
PartitionDirectory,
- Offsets(usize),
- OffsetDirectory(usize),
+ Offsets(Option<ConsumerKind>),
+ OffsetDirectory(ConsumerKind),
Generation,
FinalFrontier,
Submitted(PurgeFilePhase),
@@ -662,8 +662,8 @@ enum PurgeFilePhase {
Cleanup,
Plant,
PartitionDirectory,
- Offsets(usize),
- OffsetDirectory(usize),
+ Offsets(ConsumerKind),
+ OffsetDirectory(ConsumerKind),
Generation,
}
@@ -1792,17 +1792,27 @@ where
}
}
+ pub(crate) fn cancel_history_maintenance(&mut self) {
+ self.maintenance.checkpoint = None;
+ self.maintenance.retention = None;
+ }
+
fn capture_checkpoint_phase(
&mut self,
) -> Result<Option<crate::io::PartitionIoJob<SB>>, IggyError> {
let job = match &mut self.maintenance.checkpoint {
- Some(PendingCheckpoint::Receipts(through, receipts)) => {
+ Some(PendingCheckpoint::Receipts(through, _)) => {
let through = *through;
- let bytes = std::mem::take(receipts);
let path = self
.retry_checkpoint_path(through)
.map_err(|_| IggyError::CannotWriteToFile)?;
- self.maintenance.checkpoint = Some(PendingCheckpoint::ReceiptsWriting(through));
+ let Some(PendingCheckpoint::Receipts(_, bytes)) = self
+ .maintenance
+ .checkpoint
+ .replace(PendingCheckpoint::ReceiptsWriting(through))
+ else {
+ unreachable!("checkpoint phase is unchanged during capture");
+ };
crate::PartitionIoJob::RetryCheckpoint { path, bytes }
}
Some(PendingCheckpoint::Index(through)) => {
@@ -1939,13 +1949,11 @@ where
if self.superblock.is_some() && self.consensus.needs_superblock_persist() {
return;
}
- let durable_op = persistence.durable_op();
loop {
let pending = self
.pending_persisted_acks
.borrow()
.first_key_value()
- .filter(|(op, _)| **op <= durable_op)
.map(|(_, header)| *header);
let Some(header) = pending else {
break;
@@ -1975,12 +1983,9 @@ where
const fn requires_persistence(&self, operation: Operation) -> bool {
match operation {
- // Follows the partition policy, so that a retirement sweep does
- // not sync a Replicated WAL. A crash can then lose the fence and
- // bring back the receipts that the last checkpoint holds.
- Operation::RetireSession => {
- self.durability().is_persisted() || self.consumer_offset_durability().is_persisted()
- }
+ // Metadata can durably finalize a session after this fence commits.
+ // Recovery must not resurrect receipts that can no longer be retired.
+ Operation::RetireSession => true,
Operation::SendMessages => self.durability().is_persisted(),
Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset => {
self.consumer_offset_durability().is_persisted()
@@ -2355,13 +2360,14 @@ where
}
let _guard = self.superblock_lock.acquire().await;
let state = self.superblock_state(intent);
+ let inside_backoff_window = self.superblock_write_is_backed_off();
let result = crate::io::SuperblockIoJob {
superblock: Rc::clone(superblock),
state,
}
.execute()
.await;
- self.accept_superblock(result, self.superblock_write_is_backed_off())
+ self.accept_superblock(result, inside_backoff_window)
}
#[allow(clippy::future_not_send)]
@@ -3045,9 +3051,8 @@ where
///
/// Pairs with [`Self::needs_offset_reservation_extension`]; the caller is the
/// shard tick, so this write is off the append path. A failure needs no
- /// handling beyond the writer's own logging and backoff: the fence at the
- /// mint is still there, and it is what refuses the append if the ceiling
- /// never caught up.
+ /// handling beyond the writer's own logging and backoff: admission reserves
+ /// each request's exact mint range before it can be sequenced.
///
/// From the ceiling, NOT [`Self::mint_frontier`]. The trigger fires while the
/// append point still sits under the ceiling -- that being the point of
@@ -3065,16 +3070,10 @@ where
.await
}
- /// Upper bound on the offsets a pending `SendMessages` request will mint, for
- /// fencing it BEFORE it enters the pipeline.
- ///
- /// `project` assigns an op, not a base offset, so the exact range is unknown
- /// until the mint runs under `write_lock`. This is deliberately loose: a
- /// request pipelined behind others can land above it, and the fence at the
- /// mint stays as the exact check. It does not need to be tight -- the claim
- /// runs a whole lease block past whatever it is handed, so one of these
- /// covers every batch in flight unless a run of them crosses a block
- /// boundary.
+ /// Last offset a `SendMessages` request will mint when sequenced now.
+ /// Admission checks this exact range before sequencing or queue promotion;
+ /// a queued request must recompute it when preceding requests have advanced
+ /// the mint frontier.
///
/// `None` above one replica, where nothing is reserved, and when the body is
/// not one canonical batch, which `convert_request_message` has already
@@ -3083,7 +3082,7 @@ where
/// Header decode, NOT `decode_batch_slice`: the verifying decode fails on
/// every ordinary send, because `convert_request_message` runs at
/// [`ChecksumMode::Skip`] and leaves `batch_checksum` zeroed, which would
- /// silently drop the fence back to the mint. It also re-hashes every body
+ /// silently skip reservation during admission. It also re-hashes every body
/// `admit_wire_request` already hashed.
fn request_mint_ceiling(&self, message: &Message<RoutedRequestHeader>) -> Option<u64> {
if self.consensus.replica_count() > 1 {
@@ -3106,58 +3105,17 @@ where
Some(self.mint_frontier().saturating_add(last))
}
- /// The append fence: make sure the durable record already permits every
- /// offset up to and including `end_offset` before the caller lets them
- /// exist.
- ///
- /// `SendMessagesResponse` hands clients concrete base offsets and the poll
- /// path serves committed messages out of the resident journal, so an offset
- /// is client-visible long before the threshold-gated flush names it in a
- /// segment, and a crash in between hands a second message an offset a client
- /// already holds. Fencing here rather than at commit puts it upstream of
- /// every way a NEWLY minted offset escapes -- the reply, the poll tier, the
- /// peers a prepare reaches -- on the one path both a primary's mint and a
- /// backup's re-stamp take. Journal repair
- /// (`append_repaired_send_messages`) is the exception and needs none: it
- /// re-journals offsets a peer already minted and fenced, so there is nothing
- /// new to claim. Claiming through `end_offset + 1 + lease` rather than from
- /// the live counter needs no special case for an oversized batch.
- ///
- /// SOLO ONLY, like everything the reservation feeds: `restore_offset_frontier`
- /// seeds no counter from it above one replica, and the boot re-anchor that
- /// shapes the chain around it never runs there either. A replicated group
- /// paying a superblock write per block would buy nothing -- and an ack there
- /// already means a quorum journaled the batch, so re-minting needs a
- /// FULL-cluster crash.
- ///
- /// `false` when the write was attempted and failed. Fail-closed: the send is
- /// rejected with nothing externalised, exactly as a failed view persist
- /// withholds its sends.
+ /// Reserve a single-replica offset range before CREATE publishes the partition.
+ /// Admission uses `reserve_offsets_through_retryable` for the exact range a
+ /// request will mint. Minting only verifies that range; it never extends it.
+ /// Repair reuses offsets already fenced by their original replica.
///
- /// BYPASSES the retry backoff, like `record_frontier_before_quarantine` and
- /// for the same reason: this is the fence at the MINT, where a refusal is
- /// terminal. The failure cell is shared with every other superblock writer on
- /// the partition, so a purge's frontier reset or the tick's own extension
- /// failing once would otherwise open a 20 ms window (up to 1 s after repeats)
- /// in which this takes the node down over a fault on another path entirely.
- /// The refusal is also not deferred work: `superblock_wedged` is the gate that
- /// decides a run of failures is terminal.
+ /// This path bypasses retry backoff because a failed CREATE can be retried
+ /// before the namespace is published. Mounted partitions defer the write to
+ /// their I/O lane. Replicated groups do not use the reservation ceiling.
///
- /// The ADMITTED path does honour the backoff, through
- /// `reserve_offsets_through_retryable`, because a refusal there costs
- /// one client retry and re-running a full atomic replace per retry starves
- /// the shard pump on a failing disk.
- ///
- /// WHERE it is called decides how much a refusal costs. Ahead of the pipeline
- /// (`on_request`) the client gets a retryable transient and the group keeps
- /// serving. At the mint the op already has its number and its ack is already
- /// skipped, so `commit_max` can never pass it and nothing later can commit
- /// either: `on_replicate` fences the partition there and takes the node down.
- /// At CREATE (`build_partition_fresh`) nothing has been externalised at all,
- /// so a refusal fails the build and leaves the namespace unmaterialised for
- /// the reconciler to retry, boot included. Going live without the
- /// block instead would let the first send land inside the backoff the failed
- /// write just armed, where the admitted path refuses it with a transient.
+ /// The frontier is exclusive: `u64::MAX` cannot be covered. Saturating its
+ /// successor would permit an offset that recovery could mint again.
#[allow(clippy::future_not_send)]
#[must_use = "the bool is the fence verdict; dropping it lets the append escape unreserved"]
pub async fn reserve_offsets_through(&self, end_offset: u64) -> bool {
@@ -3172,7 +3130,7 @@ where
return true;
};
let _guard = self.superblock_lock.acquire().await;
- return self.write_offset_claim(superblock, end_offset).await;
+ return self.write_claim_from(superblock, exclusive).await;
}
self.complete_superblock_intent(SuperblockIntent::Advance {
frontier: 0,
@@ -3188,13 +3146,9 @@ where
/// coverage fast path: a batch the record already permits owes the disk
/// nothing and must not be refused by a window some other writer opened.
///
- /// The fence at the mint deliberately does not honour it. By the time that
- /// one runs the op has its number and its ack is already skipped, so a
- /// refusal fences the partition and takes the node down; attempting the write
- /// against a disk that just refused one is strictly better than declining to
- /// try. Here the client simply retries, so re-running a full create, write,
- /// file fsync, rename and directory fsync per retry inside an open window
- /// buys nothing and starves the shard pump for as long as the fault lasts.
+ /// Minting performs no I/O and requires this exact range to be covered.
+ /// Backoff keeps client retries from repeating atomic replacements on a
+ /// failing disk while already-covered requests can continue.
#[allow(clippy::future_not_send)]
#[must_use = "the bool is the fence verdict; dropping it lets the append escape unreserved"]
async fn reserve_offsets_through_retryable(
@@ -3267,38 +3221,10 @@ where
.await;
}
- /// Claim a block past `end_offset` unconditionally.
- ///
- /// Split from [`Self::reserve_offsets_through`] because the tick's extension
- /// has to write while the ceiling still covers the append point -- that is the
- /// point of extending early -- and the fence's coverage fast path would
- /// short-circuit exactly that call. The advancing write is what keeps this
- /// monotone: a claim below the record cannot lower it.
- ///
- /// `false` for an `end_offset` of `u64::MAX`, which EXHAUSTS the offset space:
- /// the record is an exclusive frontier, so covering an offset needs a value
- /// strictly above it and `u64::MAX + 1` does not exist. Saturating instead
- /// would record `u64::MAX`, report success for an offset it does not cover,
- /// and let the next boot seed the append counter at `u64::MAX - 1` and re-mint
- /// an offset a client already holds -- the one defect this whole path exists
- /// to prevent, at the one offset where it would be silent.
- #[allow(clippy::future_not_send)]
- async fn write_offset_claim(&self, superblock: &Rc<SB>, end_offset: u64) -> bool {
- let Some(exclusive) = end_offset.checked_add(1) else {
- tracing::error!(
- namespace_raw = self.consensus.group(),
- end_offset,
- "refusing an append that would exhaust the partition's offset space"
- );
- return false;
- };
- self.write_claim_from(superblock, exclusive).await
- }
-
/// Claim a lease of offsets past `exclusive`, the lowest offset the record
/// does not yet permit.
///
- /// Saturating on the lease is safe where [`Self::write_offset_claim`]'s
+ /// Saturating on the lease is safe where [`Self::reserve_offsets_through`]'s
/// successor is not: a claim clamped to `u64::MAX` still sits strictly above
/// every `end_offset` below it, so the coverage test still holds. Only the
/// successor itself can push the frontier off the end of the space.
@@ -4519,19 +4445,8 @@ where
match &head.phase {
NoAckPhase::Validate => {
let pending = head.pending;
- if head.poll_history != self.poll_history {
- self.fail_no_ack(pending.history_reset_refusal(), false);
- return NoAckStep::Reply;
- }
- if self.fatal.is_some()
- || !self.consensus.is_primary()
- || !self.consensus.is_normal()
- || self.consensus.is_transferring()
- || !self
- .consumer_offset_capacity_for(pending.kind)
- .owns(&head.reservation)
- {
- self.fail_no_ack(IggyError::TransientNotAccepted, false);
+ if let Some(error) = self.no_ack_refusal(head) {
+ self.fail_no_ack(error, false);
return NoAckStep::Reply;
}
if self.pending_commit.is_some() {
@@ -5294,6 +5209,30 @@ where
}
}
+ pub fn retry_io(&self) {
+ if self.io_active.get().is_none()
+ && (self.pending_commit.is_some()
+ || self.materialization.is_some()
+ || self.transition.is_some()
+ || self.superblock_intent.get().is_some()
+ || self.pending_wire_actions.borrow().is_some()
+ || !self.pending_persisted_acks.borrow().is_empty()
+ || !self.no_ack.is_empty()
+ || !self.io_replies.is_empty()
+ || self.maintenance.flush_through.is_some()
+ || self.maintenance.checkpoint.is_some()
+ || self.maintenance.retention.is_some()
+ || self
+ .maintenance
+ .shutdown
+ .is_some_and(|phase| phase != PartitionShutdownPhase::Complete)
+ || self.io_promotions > 0
+ || self.queued_requests_ready())
+ {
+ self.notify_io();
+ }
+ }
+
#[must_use]
pub fn io_active(&self) -> Option<crate::PartitionIoIdentity> {
self.io_active.get()
@@ -5338,11 +5277,23 @@ where
}
pub fn defer_view_transition(&mut self, message: server_common::MessageBag) {
- if self.transition.is_none() {
- self.transition = Some(PendingPartitionTransition::View {
- message,
- drain: None,
- });
+ match &mut self.transition {
+ Some(PendingPartitionTransition::View {
+ message: pending, ..
+ }) => {
+ if deferred_view_order(&message).is_some_and(|incoming| {
+ deferred_view_order(pending).is_some_and(|current| incoming > current)
+ }) {
+ *pending = message;
+ }
+ }
+ None => {
+ self.transition = Some(PendingPartitionTransition::View {
+ message,
+ drain: None,
+ });
+ }
+ Some(_) => {}
}
self.notify_io();
}
@@ -5382,7 +5333,7 @@ where
match self.resume_io_inner(config).await {
Ok(step) => step,
Err(error) => {
- self.fail_io_work(&error);
+ self.fail_io_work(&error, false);
if self.maintenance.shutdown.is_some() {
self.maintenance.shutdown = Some(PartitionShutdownPhase::Complete);
}
@@ -5624,8 +5575,13 @@ where
if self.maintenance.shutdown.is_none() && self.drive_transfer_plan() {
return Ok(crate::PartitionIoStep::Progress);
}
- if let Some(step) = self.drive_checkpoint_io(config)? {
- return Ok(step);
+ match self.drive_checkpoint_io(config) {
+ Ok(Some(step)) => return Ok(step),
+ Ok(None) => {}
+ Err(error) => {
+ self.fail_io_work(&error, true);
+ return Ok(crate::PartitionIoStep::Pending);
+ }
}
if let Some(through) = self.maintenance.retention {
if self
@@ -5708,13 +5664,20 @@ where
}
if continuation == crate::PartitionIoContinuation::Purge
&& let Some(PendingPartitionTransition::Purge(purge)) = &self.transition
- && let PurgePhase::Offsets(kind) = purge.phase
+ && let PurgePhase::Offsets(Some(kind)) = purge.phase
{
allocation_charge = self
.offset_cleanup_charge(kind)
.and_then(|known| allocation_charge.checked_add(known))
.ok_or(IggyError::CannotWriteToFile)?;
}
+ if continuation == crate::PartitionIoContinuation::Purge
+ && matches!(&self.transition, Some(PendingPartitionTransition::Purge(purge)) if matches!(purge.phase, PurgePhase::Cleanup))
+ {
+ allocation_charge = allocation_charge
+ .checked_add(crate::state_transfer::SEGMENT_SWEEP_STACK_BYTES)
+ .ok_or(IggyError::CannotWriteToFile)?;
+ }
if allocation_charge > self.io_bytes_max.get() {
return Err(IggyError::CannotWriteToFile);
}
@@ -5901,7 +5864,10 @@ where
}
};
if let Err(error) = &outcome {
- self.fail_io_work(error);
+ self.fail_io_work(
+ error,
+ identity.continuation == crate::PartitionIoContinuation::Checkpoint,
+ );
}
self.notify_io();
outcome
@@ -5936,7 +5902,10 @@ where
}
}
crate::PartitionIoContinuation::Retention => self.maintenance.retention = None,
- _ => self.fail_io_work(&error),
+ _ => self.fail_io_work(
+ &error,
+ continuation == crate::PartitionIoContinuation::Checkpoint,
+ ),
}
self.notify_io();
}
@@ -5973,8 +5942,15 @@ where
.is_none_or(|persistence| persistence.is_quiescent())))
}
- fn fail_io_work(&mut self, error: &IggyError) {
- if self.pending_commit.is_none() && self.maintenance.transfer_flush {
+ fn fail_io_work(&mut self, error: &IggyError, checkpoint_failure: bool) {
+ if self.pending_commit.is_none()
+ && self.maintenance.transfer_flush
+ && !checkpoint_failure
+ && !matches!(
+ self.maintenance.checkpoint,
+ Some(PendingCheckpoint::Flush(_))
+ )
+ {
self.materialization = None;
self.finish_flush();
self.maintenance.flush_completed_through = None;
@@ -6130,10 +6106,7 @@ where
}
match self.maintenance.shutdown {
Some(PartitionShutdownPhase::DrainAccepted) => {
- if self.pending_commit.is_some()
- || !self.no_ack.is_empty()
- || self.consensus.request_queue_len() > 0
- {
+ if !self.no_ack.is_empty() || self.consensus.request_queue_len() > 0 {
return crate::PartitionIoStep::Pending;
}
// A stopping backup may never ack the uncommitted pipeline tail.
@@ -6216,9 +6189,13 @@ where
// Completion relies on shutdown rejecting new and repaired journal entries.
fn shutdown_flush_target(&self) -> u64 {
- self.consensus
+ let committed = self
+ .consensus
.commit_max()
- .min(self.log.journal().inner.last_op().unwrap_or(0))
+ .min(self.log.journal().inner.last_op().unwrap_or(0));
+ self.persistence
+ .as_ref()
+ .map_or(committed, |persistence| committed.min(persistence.head()))
}
async fn resume_pending_prepare_ack(&self) -> bool {
@@ -6243,13 +6220,6 @@ where
self.pending_persisted_acks.borrow_mut().remove(&header.op);
return true;
}
- if self
- .persistence
- .as_ref()
- .is_some_and(|persistence| !persistence.is_durable(&header))
- {
- return false;
- }
if !self.send_prepare_ok(&header).await {
return false;
}
@@ -6259,6 +6229,7 @@ where
fn queue_committed_replies(&mut self) {
if let Some(committed) = self.publish_pending_commit() {
+ debug_assert!(self.io_replies.is_empty());
self.io_promotions += committed.len();
self.io_replies = committed.into();
}
@@ -7154,14 +7125,7 @@ where
fn no_ack_ready(&self) -> bool {
self.no_ack.front().is_some_and(|head| match head.phase {
NoAckPhase::Validate => {
- head.poll_history != self.poll_history
- || self.fatal.is_some()
- || !self.consensus.is_primary()
- || !self.consensus.is_normal()
- || self.consensus.is_transferring()
- || !self
- .consumer_offset_capacity_for(head.pending.kind)
- .owns(&head.reservation)
+ self.no_ack_refusal(head).is_some()
|| (self.pending_commit.is_none()
&& (self.maintenance.shutdown.is_some()
|| !self.consensus.with_pipeline(|pipeline| {
@@ -7173,6 +7137,20 @@ where
})
}
+ fn no_ack_refusal(&self, head: &PendingNoAck) -> Option<IggyError> {
+ if head.poll_history != self.poll_history {
+ return Some(head.pending.history_reset_refusal());
+ }
+ (self.fatal.is_some()
+ || !self.consensus.is_primary()
+ || !self.consensus.is_normal()
+ || self.consensus.is_transferring()
+ || !self
+ .consumer_offset_capacity_for(head.pending.kind)
+ .owns(&head.reservation))
+ .then_some(IggyError::TransientNotAccepted)
+ }
+
/// Shutdown promotes nothing, so answer every queued request now. None
/// holds an op yet, so its client can safely retry it, unless it is an
/// offset mutation that a history reset made stale.
@@ -8085,8 +8063,8 @@ where
})
}
- /// Drop an uncommitted view-divergent suffix and restore every append cursor
- /// from the retained prefix as one write-locked operation.
+ /// Drop an uncommitted view-divergent suffix inline and restore append cursors
+ /// from the retained prefix. Mounted partitions defer this work to the I/O lane.
///
/// # Errors
///
@@ -8197,7 +8175,6 @@ where
config: &PartitionsConfig,
through_op: u64,
) -> Result<bool, IggyError> {
- #[cfg(any(test, feature = "fault-injection"))]
if std::mem::take(&mut self.injected_commit_failure) {
return Err(IggyError::CannotSaveMessagesToSegment);
}
@@ -8292,14 +8269,15 @@ where
>= self.effective_messages_required_to_save(config) as usize
}
- /// Returns false while the captured prefix is still pending in the WAL.
+ /// Returns false while the captured prefix is pending in the WAL or has been
+ /// handed to the dispatcher. Partitions without a dispatcher run it inline.
async fn commit_messages_inner(
&mut self,
config: &PartitionsConfig,
force: bool,
through_op: u64,
) -> Result<bool, IggyError> {
- if force && self.pending_commit.is_none() {
+ if force && self.pending_commit.is_none() && self.has_io_dispatcher() {
self.defer_flush(through_op);
}
if self.materialization.is_none() {
@@ -8315,7 +8293,7 @@ where
return Ok(true);
}
}
- if self.io_notifier.borrow().is_some() {
+ if self.has_io_dispatcher() {
self.notify_io();
return Ok(false);
}
@@ -8775,7 +8753,7 @@ where
}
/// Evict the committed prefix (the `count` front entries read by
- /// `committed_prefix`) and reset `journal.info` to reflect only the
+ /// `committed_prefix_len`) and reset `journal.info` to reflect only the
/// uncommitted tail left resident, so the next persist threshold counts that
/// tail alone. Call once the prefix is durable, or when there is nothing to
/// persist. The retained tail's accounting is folded from the meta
@@ -8879,65 +8857,72 @@ where
if !self.pending_commit_is_current() {
return Err(IggyError::CannotWriteToFile);
}
- let pending = self
- .pending_commit
- .as_ref()
- .ok_or(IggyError::CannotWriteToFile)?;
- match pending.phase {
- CommitPhase::Start => self.start_commit_materialization(),
- CommitPhase::Materialize => match self.drive_materialization()? {
- MaterializationStep::Progress => Ok(CommitStep::Progress),
- MaterializationStep::Pending => Ok(CommitStep::Pending),
- MaterializationStep::Ready => Ok(CommitStep::Ready),
- MaterializationStep::Complete => {
- let pending = self
- .pending_commit
- .as_mut()
- .expect("selected commit exists");
+ // Selection bounds the commit run. Revalidate after yielding or I/O,
+ // while CPU-only phases can share this validation in the same turn.
+ loop {
+ let pending = self
+ .pending_commit
+ .as_ref()
+ .ok_or(IggyError::CannotWriteToFile)?;
+ let step = match pending.phase {
+ CommitPhase::Start => self.start_commit_materialization(),
+ CommitPhase::Materialize => match self.drive_materialization()? {
+ MaterializationStep::Progress => Ok(CommitStep::Progress),
+ MaterializationStep::Pending => Ok(CommitStep::Pending),
+ MaterializationStep::Ready => Ok(CommitStep::Ready),
+ MaterializationStep::Complete => {
+ let pending = self
+ .pending_commit
+ .as_mut()
+ .expect("selected commit exists");
- pending.phase = CommitPhase::Offsets;
- Ok(CommitStep::Progress)
+ pending.phase = CommitPhase::Offsets;
+ Ok(CommitStep::Progress)
+ }
+ },
+ CommitPhase::Offsets => self.drive_committed_offsets(),
+ CommitPhase::OffsetReady(_) if self.offset_writes_wait_for_checkpoint() => {
+ Ok(CommitStep::Pending)
}
- },
- CommitPhase::Offsets => self.drive_committed_offsets(),
- CommitPhase::OffsetReady(_) if self.offset_writes_wait_for_checkpoint() => {
- Ok(CommitStep::Pending)
- }
- CommitPhase::OffsetReady(_) => Ok(CommitStep::Ready),
- CommitPhase::SegmentDirectory => {
- if (pending.has_sends() || pending.materialize)
- && self.consensus.replica_count() == 1
- && self.durability().is_persisted()
- && self.segment_names_dirty.get()
- && self.partition_dir.is_some()
- {
- return Ok(CommitStep::Ready);
+ CommitPhase::OffsetReady(_) => Ok(CommitStep::Ready),
+ CommitPhase::SegmentDirectory => {
+ if (pending.has_sends() || pending.materialize)
+ && self.consensus.replica_count() == 1
+ && self.durability().is_persisted()
+ && self.segment_names_dirty.get()
+ && self.partition_dir.is_some()
+ {
+ return Ok(CommitStep::Ready);
+ }
+ self.pending_commit
+ .as_mut()
+ .expect("selected commit exists")
+ .phase = CommitPhase::OffsetDirectories;
+ Ok(CommitStep::Progress)
}
- self.pending_commit
- .as_mut()
- .expect("selected commit exists")
- .phase = CommitPhase::OffsetDirectories;
- Ok(CommitStep::Progress)
- }
- CommitPhase::OffsetDirectories => {
- let dirty = self
- .consumer_offset_dirs_touched
- .iter()
- .zip(&self.consumer_offset_dirs_dirty)
- .any(|(touched, dirty)| touched.get().is_some() && dirty.get());
- if dirty {
- return Ok(CommitStep::Ready);
+ CommitPhase::OffsetDirectories => {
+ let dirty = self
+ .consumer_offset_dirs_touched
+ .iter()
+ .zip(&self.consumer_offset_dirs_dirty)
+ .any(|(touched, dirty)| touched.get().is_some() && dirty.get());
+ if dirty {
+ return Ok(CommitStep::Ready);
+ }
+ self.pending_commit
+ .as_mut()
+ .expect("selected commit exists")
+ .phase = CommitPhase::Publish;
+ Ok(CommitStep::Publish)
}
- self.pending_commit
- .as_mut()
- .expect("selected commit exists")
- .phase = CommitPhase::Publish;
- Ok(CommitStep::Publish)
+ CommitPhase::Publish => Ok(CommitStep::Publish),
+ CommitPhase::OffsetSubmitted(_)
+ | CommitPhase::SegmentDirectorySubmitted
+ | CommitPhase::OffsetDirectoriesSubmitted => Ok(CommitStep::Pending),
+ }?;
+ if !matches!(step, CommitStep::Progress) {
+ return Ok(step);
}
- CommitPhase::Publish => Ok(CommitStep::Publish),
- CommitPhase::OffsetSubmitted(_)
- | CommitPhase::SegmentDirectorySubmitted
- | CommitPhase::OffsetDirectoriesSubmitted => Ok(CommitStep::Pending),
}
}
@@ -9044,9 +9029,6 @@ where
&mut self,
config: &PartitionsConfig,
) -> Result<Option<crate::io::PartitionIoJob<SB>>, IggyError> {
- if !self.pending_commit_is_current() {
- return Err(IggyError::CannotWriteToFile);
- }
let phase = self
.pending_commit
.as_ref()
@@ -9213,10 +9195,11 @@ where
return false;
}
let from_pipeline = self.consensus.pipeline_head_header().as_ref() == headers.first();
- let journal_headers = headers
- .iter()
- .map(|header| self.log.journal().inner.header_by_op(header.op))
- .collect();
+ let journal_headers = self
+ .log
+ .journal()
+ .inner
+ .headers_for_commit(headers[0].op, headers.len());
let batch_stats = self.resolve_committed_batch_stats(&headers);
let Some(retirements) = self.validate_committed_entries(&headers, &batch_stats) else {
return false;
@@ -9274,11 +9257,11 @@ where
{
return false;
}
- if pending
- .headers
- .iter()
- .zip(&pending.journal_headers)
- .any(|(header, captured)| self.log.journal().inner.header_by_op(header.op) != *captured)
+ if !self
+ .log
+ .journal()
+ .inner
+ .commit_headers_match(pending.headers[0].op, &pending.journal_headers)
{
return false;
}
@@ -9307,10 +9290,6 @@ where
#[allow(clippy::too_many_lines)]
fn publish_pending_commit(&mut self) -> Option<Vec<(PipelineEntry, Option<CachedReply>)>> {
- if !self.pending_commit_is_current() {
- self.pending_commit = None;
- return None;
- }
let pending = self.pending_commit.take()?;
let entries = if pending.from_pipeline {
self.consensus.with_pipeline_mut(|pipeline| {
@@ -10213,9 +10192,9 @@ where
/// re-stages a committed delete watermark on every pass while the first
/// local segment still starts below it.
///
- /// Holds `write_lock` to serialize against the commit/rotate path, which
- /// runs on the separate consensus-tick loop.
- #[allow(clippy::too_many_lines)]
+ /// Mounted or busy partitions retain the watermark for the I/O lane and
+ /// return an empty result. Only an idle partition without a dispatcher removes
+ /// files inline and returns their counts.
pub async fn remove_sealed_segments_up_to(&mut self, up_to_offset: u64) -> SegmentRemoval {
self.maintenance.retention = Some(
self.maintenance
@@ -10395,7 +10374,9 @@ where
return Err(IggyError::CannotSyncFile);
}
}
- self.flush_committed_messages(config).await?;
+ if !self.flush_committed_messages(config).await? {
+ return Err(IggyError::CannotSyncFile);
+ }
persistence.request_checkpoint();
self.checkpoint_persistence(config).await;
persistence
@@ -10601,56 +10582,6 @@ where
Ok(())
}
- /// Record the purge's frontier reset BEFORE the purge touches anything.
- ///
- /// The unlinks are made durable by their own directory fsync, so a crash
- /// between them and a reset written afterwards boots a purged directory
- /// whose record still names the pre-purge offset space:
- /// `restore_offset_frontier` re-seeds the counter to it while every peer
- /// restarted at 0, and the first append stamps a `base_offset` and
- /// `batch_checksum` no peer shares. Writing 0 first inverts the window into
- /// a harmless one -- the record under-claims while the segments still
- /// exist, and boot takes the max of the record and what the segments prove.
- ///
- /// Spelled out rather than read off the counter, which still holds the
- /// pre-purge frontier at this point.
- ///
- /// # Errors
- /// [`PurgeError::FrontierNotRecorded`]. Refused rather than logged: nothing
- /// has been mutated yet, and a purge that cannot record its reset must not
- /// be the one that erases the data proving the old frontier. The caller
- /// RETRIES; it must not fence, since the chain is still whole and the live
- /// counter still names the pre-purge space.
- #[allow(clippy::future_not_send)]
- #[cfg(test)]
- async fn record_purge_frontier_reset(&mut self, generation: u64) -> Result<(), PurgeError> {
- if self.reset_offset_frontier_at(0).await {
- self.purge_deferred = false;
- return Ok(());
- }
- self.purge_deferred = true;
- // The ONLY operator-visible signal for the withhold: `send_prepare_ok`
- // returns silently, correctly, since it runs per prepare. So this line
- // has to say that the replica is now out of quorum for this group, or
- // the symptom reads as a network fault. The consecutive count
- // correlates it with the superblock writer's own error log, which
- // carries the `ENOSPC` / `EIO` cause but is rate-limited to
- // power-of-two failures, while this deferral repeats per reconciler
- // pass.
- warn!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- namespace_raw = self.namespace().inner(),
- generation,
- superblock_write_failures = self.superblock_write_failures.get(),
- "cannot record the purge's offset-frontier reset; deferring the purge so the \
- durable frontier cannot outlive the data it describes. This replica now \
- withholds PrepareOk for this partition until the purge lands, so it is \
- quorum-invisible there; its other partitions are unaffected"
- );
- 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
@@ -10745,7 +10676,7 @@ where
self.transition = Some(PendingPartitionTransition::Purge(PendingPurge {
generation,
drain: None,
- phase: PurgePhase::Offsets(0),
+ phase: PurgePhase::Offsets(Some(ConsumerKind::Consumer)),
record_only: false,
}));
self.drive_purge_inline(config, storage).await
@@ -10837,8 +10768,7 @@ where
.set_state_transfer_stage(consensus::StateTransferStage::Idle);
}
self.reuse_scan_memo.borrow_mut().take();
- self.maintenance.checkpoint = None;
- self.maintenance.retention = None;
+ self.cancel_history_maintenance();
if let Some(persistence) = &self.persistence {
persistence.retire_offset_files();
}
@@ -10854,16 +10784,16 @@ where
}
PurgePhase::Cleanup if self.partition_dir.is_none() => purge.phase = PurgePhase::Plant,
PurgePhase::PartitionDirectory if self.partition_dir.is_none() => {
- purge.phase = PurgePhase::Offsets(0);
+ purge.phase = PurgePhase::Offsets(Some(ConsumerKind::Consumer));
}
- PurgePhase::Offsets(ConsumerKind::COUNT) => {
+ PurgePhase::Offsets(None) => {
self.publish_purged_state();
purge.phase = PurgePhase::Generation;
}
- PurgePhase::Offsets(kind) | PurgePhase::OffsetDirectory(kind)
- if self.purge_offset_directory(kind).is_none() =>
+ PurgePhase::Offsets(Some(kind)) | PurgePhase::OffsetDirectory(kind)
+ if self.consumer_offsets_dir(kind).is_none() =>
{
- purge.phase = PurgePhase::Offsets(kind + 1);
+ purge.phase = PurgePhase::Offsets(next_consumer_kind(kind));
}
PurgePhase::Generation if self.partition_dir.is_none() => {
self.finish_purge_generation(purge.generation);
@@ -10896,10 +10826,16 @@ where
};
}
PurgePhase::Submitted(_) => step = crate::PartitionIoStep::Pending,
- _ => match self.io_plan(crate::PartitionIoContinuation::Purge) {
- Ok(ready) => step = ready,
- Err(error) => purge.phase = PurgePhase::Done(Err(PurgeError::Unserviceable(error))),
- },
+ _ => {
+ self.transition = Some(PendingPartitionTransition::Purge(purge));
+ return match self.io_plan(crate::PartitionIoContinuation::Purge) {
+ Ok(ready) => ready,
+ Err(error) => {
+ self.reject_io_capture(crate::PartitionIoContinuation::Purge, error);
+ crate::PartitionIoStep::Progress
+ }
+ };
+ }
}
self.transition = Some(PendingPartitionTransition::Purge(purge));
step
@@ -10953,13 +10889,13 @@ where
),
PurgeFilePhase::PartitionDirectory,
),
- PurgePhase::Offsets(kind) => (
+ PurgePhase::Offsets(Some(kind)) => (
self.capture_offset_cleanup(*kind, Vec::new())?,
PurgeFilePhase::Offsets(*kind),
),
PurgePhase::OffsetDirectory(kind) => (
crate::PartitionIoJob::SegmentDirectory(
- self.purge_offset_directory(*kind)
+ self.consumer_offsets_dir(*kind)
.ok_or(IggyError::CannotSyncFile)?
.to_owned(),
),
@@ -11026,20 +10962,18 @@ where
if let Err(error) = outcome {
warn!(%error, "purge could not sync the partition directory");
}
- PurgePhase::Offsets(0)
+ PurgePhase::Offsets(Some(ConsumerKind::Consumer))
}
(
PurgePhase::Submitted(PurgeFilePhase::Offsets(kind)),
crate::PartitionIoResult::PurgeOffsets(outcome),
) => {
- if let Some(consumer_kind) = purge_offset_kind(kind) {
- let capacity = self.consumer_offset_capacity_for(consumer_kind);
- for consumer_id in outcome.released {
- capacity.clear_stranded(consumer_id);
- }
- for consumer_id in outcome.failed {
- capacity.record_stranded(consumer_id);
- }
+ let capacity = self.consumer_offset_capacity_for(kind);
+ for consumer_id in outcome.released {
+ capacity.clear_stranded(consumer_id);
+ }
+ for consumer_id in outcome.failed {
+ capacity.record_stranded(consumer_id);
}
if outcome.scan_error.is_some() {
PurgePhase::Done(Err(PurgeError::Unserviceable(IggyError::CannotDeleteFile)))
@@ -11086,7 +11020,7 @@ where
self.transition = Some(PendingPartitionTransition::Purge(purge));
}
- pub(crate) fn offset_cleanup_charge(&self, kind: usize) -> Option<usize> {
+ pub(crate) fn offset_cleanup_charge(&self, kind: ConsumerKind) -> Option<usize> {
// Bound hash-table load factor and bucket rounding. The file-phase
// control reserve covers the trailing control group and alignment.
const HASH_BUCKETS_PER_ID_MAX: usize = 4;
@@ -11097,22 +11031,21 @@ where
pub(crate) fn capture_offset_cleanup(
&self,
- kind: usize,
+ kind: ConsumerKind,
retained: Vec<u32>,
) -> Result<crate::PartitionIoJob<SB>, IggyError> {
- let consumer_kind = purge_offset_kind(kind).ok_or(IggyError::CannotDeleteFile)?;
let mut known = HashSet::with_capacity(
self.offset_cleanup_known_count(kind)
.ok_or(IggyError::CannotDeleteFile)?,
);
- self.consumer_offset_capacity_for(consumer_kind)
+ self.consumer_offset_capacity_for(kind)
.extend_stranded_ids(&mut known);
// The durable table names every file a committed offset op wrote.
self.durable_consumer_offsets
- .with_entries(consumer_kind, |entries| known.extend(entries.keys()));
+ .with_entries(kind, |entries| known.extend(entries.keys()));
Ok(crate::PartitionIoJob::PurgeOffsets {
directory: self
- .purge_offset_directory(kind)
+ .consumer_offsets_dir(kind)
.ok_or(IggyError::CannotDeleteFile)?
.to_owned(),
known,
@@ -11120,20 +11053,15 @@ where
})
}
- fn offset_cleanup_known_count(&self, kind: usize) -> Option<usize> {
- let consumer_kind = purge_offset_kind(kind)?;
- self.consumer_offset_capacity_for(consumer_kind)
+ fn offset_cleanup_known_count(&self, kind: ConsumerKind) -> Option<usize> {
+ self.consumer_offset_capacity_for(kind)
.stranded_count()
.checked_add(
self.durable_consumer_offsets
- .with_entries(consumer_kind, HashMap::len),
+ .with_entries(kind, HashMap::len),
)
}
- pub(crate) fn purge_offset_directory(&self, kind: usize) -> Option<&str> {
- self.consumer_offsets_dir(purge_offset_kind(kind)?)
- }
-
fn publish_purged_state(&mut self) {
self.recovered_durable_offset = None;
self.installed_frontier = None;
@@ -11173,21 +11101,21 @@ where
fn accept_purged_offset_directory(
&self,
generation: u64,
- kind: usize,
+ kind: ConsumerKind,
outcome: std::io::Result<()>,
) -> PurgePhase {
#[cfg(test)]
- if self.consumer_offset_dir_sync_fault.get() == Some(kind) {
+ if self.consumer_offset_dir_sync_fault.get() == Some(kind.index()) {
return PurgePhase::Done(Err(PurgeError::OffsetsNotDurable(
IggyError::CannotSyncFile,
)));
}
match outcome {
- Ok(()) => PurgePhase::Offsets(kind + 1),
+ Ok(()) => PurgePhase::Offsets(next_consumer_kind(kind)),
// Recovery accepts a missing directory, and a directory that is gone
// holds no unlink to make durable.
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
- PurgePhase::Offsets(kind + 1)
+ PurgePhase::Offsets(next_consumer_kind(kind))
}
Err(error) => {
warn!(
@@ -11195,7 +11123,7 @@ where
plane = "partitions",
namespace_raw = self.namespace().inner(),
generation,
- dir = self.purge_offset_directory(kind).unwrap_or_default(),
+ dir = self.consumer_offsets_dir(kind).unwrap_or_default(),
%error,
"purge could not fsync an offsets dir; its unlinks may not be durable, \
so the partition must be fenced"
@@ -11636,9 +11564,30 @@ where
}
}
-/// `PurgePhase::Offsets` sweeps the consumer kinds in `ConsumerKind::ALL` order.
-pub fn purge_offset_kind(kind: usize) -> Option<ConsumerKind> {
- ConsumerKind::ALL.get(kind).copied()
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+enum DeferredViewPhase {
+ StartViewChange,
+ DoViewChange,
+ StartView,
+}
+
+fn deferred_view_order(message: &server_common::MessageBag) -> Option<(u32, DeferredViewPhase)> {
+ match message {
+ server_common::MessageBag::StartViewChange(message) => {
+ Some((message.header().view, DeferredViewPhase::StartViewChange))
+ }
+ server_common::MessageBag::DoViewChange(message) => {
+ Some((message.header().view, DeferredViewPhase::DoViewChange))
+ }
+ server_common::MessageBag::StartView(message) => {
+ Some((message.header().view, DeferredViewPhase::StartView))
+ }
+ _ => None,
+ }
+}
+
+pub fn next_consumer_kind(kind: ConsumerKind) -> Option<ConsumerKind> {
+ ConsumerKind::ALL.get(kind.index() + 1).copied()
}
/// Automatic commits remain monotone because an earlier poll can commit after
@@ -11946,14 +11895,7 @@ pub struct SegmentRemoval {
pub budget_spent: bool,
}
-/// What a cleanup rollback could not take out of the partition counters,
-/// summed over one [`IggyPartition::remove_sealed_segments_up_to`] call.
-///
-/// Accumulated rather than reported per segment: `UnderflowSite::report` in
-/// `iggy_common` already counts every clamp and power-of-two throttles its own
-/// line, so a per-segment `warn!` here is an unthrottled second copy of it. One
-/// line per call, carrying the offset range the call removed, says which
-/// partition and which segments without that.
+/// Counter shortfalls from removing one segment, reported with its offset range.
#[derive(Debug, Default, Clone, Copy)]
struct CleanupShortfall {
size_bytes: u64,
@@ -11962,8 +11904,8 @@ struct CleanupShortfall {
}
impl CleanupShortfall {
- /// One line for the whole call, naming the partition and the offset range
- /// it removed. Silent when the counters covered every rollback.
+ /// Report one removed segment's shortfall and offset range. Silent when the
+ /// counters covered its rollback.
fn report(self, namespace: IggyNamespace, removed_offsets: Option<(u64, u64)>) {
if self.size_bytes == 0 && self.segments == 0 && self.messages == 0 {
return;
@@ -12355,12 +12297,7 @@ mod tests {
for _ in 0..STEPS_MAX {
match partition.resume_io(config).await {
crate::PartitionIoStep::Ready(plan) => {
- let captured = partition.capture_io(plan, config).unwrap().unwrap();
- let result = captured.job.execute().await;
- partition.accept_io(captured.identity, result).unwrap();
- if let Some(gate) = captured.gate {
- gate.release();
- }
+ execute_partition_io_plan(partition, config, plan).await;
}
crate::PartitionIoStep::Progress => {}
crate::PartitionIoStep::Pending => return,
@@ -12370,6 +12307,19 @@ mod tests {
panic!("partition I/O did not settle in {STEPS_MAX} steps");
}
+ async fn execute_partition_io_plan<B: MessageBus, SB: SuperblockStore>(
+ partition: &mut IggyPartition<B, SB>,
+ config: &PartitionsConfig,
+ plan: crate::PartitionIoPlan,
+ ) {
+ let captured = partition.capture_io(plan, config).unwrap().unwrap();
+ let result = captured.job.execute().await;
+ partition.accept_io(captured.identity, result).unwrap();
+ if let Some(gate) = captured.gate {
+ gate.release();
+ }
+ }
+
/// A backup with a partition directory and offset stores, ready to install
/// what an origin offers. Keep the directory alive while the receiver runs.
fn transfer_receiver() -> (tempfile::TempDir, IggyPartition<RecordingBus>) {
@@ -12596,6 +12546,88 @@ mod tests {
assert!(persistence.is_durable_through(1));
}
+ #[cfg(target_os = "linux")]
+ #[compio::test]
+ async fn checkpoint_index_sync_failure_fences_during_transfer_flush() {
+ let directory = tempfile::tempdir().unwrap();
+ let config = repair_config();
+ let (mut partition, _) = recording_partition_at(0, 3);
+ partition.set_partition_dir(directory.path().to_string_lossy().into_owned());
+ partition.runtime_options.durability = iggy_common::Durability::Persisted;
+ partition.open_persistence().await.unwrap();
+ let persistence = Rc::clone(partition.persistence.as_ref().unwrap());
+ let prepare = checksummed_segment_prepare(1, 0, 0, b"durable");
+ persistence.append(prepare.into_frozen(), true).unwrap();
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ partition.consensus.restore_commit_state(1, 1);
+ // fsync on /dev/null fails, standing in for a writeback error.
+ let writer = IggyIndexWriter::new("/dev/null", Rc::new(AtomicU64::new(0)), true, false)
+ .await
+ .unwrap();
+ let active = partition.log.index_writers().len() - 1;
+ partition.log.index_writers_mut()[active] = Some(Rc::new(writer));
+ partition.set_io_notifier(
+ Rc::new(|_, _| {}),
+ crate::largest_legal_job_charge().unwrap(),
+ );
+ // What `drive_transfer_plan` leaves behind when the flush block then
+ // falls through to `drive_checkpoint_io`.
+ partition.maintenance.transfer_flush = true;
+ partition.request_flush(1);
+ partition.maintenance.checkpoint = Some(PendingCheckpoint::Index(1));
+ let crate::PartitionIoStep::Ready(plan) = partition
+ .io_plan(crate::PartitionIoContinuation::Checkpoint)
+ .unwrap()
+ else {
+ panic!("the checkpoint index sync must plan");
+ };
+ let captured = partition.capture_io(plan, &config).unwrap().unwrap();
+ assert!(matches!(captured.job, crate::PartitionIoJob::IndexSync(_)));
+ let result = captured.job.execute().await;
+ assert!(partition.accept_io(captured.identity, result).is_err());
+ assert!(
+ partition.fatal().is_some(),
+ "a failed checkpoint index sync must fence inside a transfer flush too"
+ );
+ }
+
+ #[compio::test]
+ async fn checkpoint_planning_and_capture_failures_fence_during_transfer_flush() {
+ for fail_capture in [false, true] {
+ let (mut partition, _) = recording_partition();
+ partition.set_io_notifier(
+ Rc::new(|_, _| {}),
+ crate::largest_legal_job_charge().unwrap(),
+ );
+ partition.maintenance.transfer_flush = true;
+ partition.maintenance.checkpoint = Some(PendingCheckpoint::Receipts(1, vec![1, 2, 3]));
+ if !fail_capture {
+ partition.io_bytes_max.set(0);
+ }
+ let config = repair_config();
+ let step = partition.resume_io(&config).await;
+ if fail_capture {
+ let crate::PartitionIoStep::Ready(plan) = step else {
+ panic!("receipt checkpoint must reach capture");
+ };
+ let Err(error) = partition.capture_io(plan, &config) else {
+ panic!("capture must fail without a partition directory");
+ };
+ assert!(
+ matches!(&partition.maintenance.checkpoint, Some(PendingCheckpoint::Receipts(1, bytes)) if bytes == &[1, 2, 3])
+ );
+ partition.reject_io_capture(plan.continuation, error);
+ } else {
+ assert!(matches!(step, crate::PartitionIoStep::Pending));
+ }
+ assert!(
+ partition.fatal().is_some(),
+ "checkpoint failure must fence before transfer-flush recovery; fail_capture: {fail_capture}"
+ );
+ }
+ }
+
#[compio::test]
async fn pending_wal_prefix_keeps_pipeline_replies_and_does_not_partially_flush() {
for durability in [
@@ -13724,22 +13756,19 @@ mod tests {
}
#[compio::test]
- async fn retire_session_fence_follows_partition_durability() {
- for (durability, consumer_offset_durability, durable) in [
+ async fn retire_session_fence_is_durable_before_metadata_can_finalize() {
+ for (durability, consumer_offset_durability) in [
(
iggy_common::Durability::Replicated,
iggy_common::Durability::Replicated,
- false,
),
(
iggy_common::Durability::Persisted,
iggy_common::Durability::Replicated,
- true,
),
(
iggy_common::Durability::Replicated,
iggy_common::Durability::Persisted,
- true,
),
] {
let directory = tempfile::tempdir().unwrap();
@@ -13766,20 +13795,37 @@ mod tests {
let header = partition.consensus.pipeline_head_header().unwrap();
assert_eq!(header.operation, Operation::RetireSession);
commit_recorded_loopback(&mut partition).await;
- assert_eq!(
+ assert!(
partition
.pending_persisted_acks
.borrow()
.contains_key(&header.op),
- durable,
"{durability:?} messages, {consumer_offset_durability:?} offsets"
);
- assert_eq!(
- partition.consensus.commit_min(),
- if durable { 0 } else { header.op }
- );
+ assert_eq!(partition.consensus.commit_min(), 0);
+ assert!(!partition.session_retired(identity));
Rc::clone(&persistence).run().await;
- assert_eq!(persistence.durable_op() == header.op, durable);
+ assert!(persistence.is_durable(&header));
+ partition.drive_persistence().await;
+ commit_recorded_loopback(&mut partition).await;
+ assert!(partition.session_retired(identity));
+ let namespace = partition.namespace();
+ drop(partition);
+ drop(persistence);
+ let (recovered, prepares) =
+ crate::persistence::PartitionPersistence::open_with_storage(
+ &directory.path().join("prepares-0"),
+ namespace.inner(),
+ 0,
+ journal::durable_storage::DiskStorage,
+ )
+ .await
+ .unwrap();
+ assert!(recovered.is_durable(&header));
+ assert!(
+ prepares.iter().any(|prepare| *prepare.header() == header),
+ "recovery must retain the barrier metadata finalization relies on"
+ );
}
}
@@ -14269,6 +14315,79 @@ mod tests {
assert!(partition.pending_persisted_acks.borrow().is_empty());
}
+ #[compio::test]
+ async fn pending_prepare_ack_uses_topic_durability_after_superblock_completion() {
+ for durability in [
+ iggy_common::Durability::Replicated,
+ iggy_common::Durability::Persisted,
+ ] {
+ for dispatched in [false, true] {
+ let directory = tempfile::tempdir().unwrap();
+ let mut partition = partition_at_view(0, 0);
+ partition.set_partition_dir(directory.path().to_string_lossy().into_owned());
+ partition.runtime_options.durability = durability;
+ partition.open_persistence().await.unwrap();
+ let persistence = Rc::clone(partition.persistence.as_ref().unwrap());
+ let prepare = checksummed_segment_prepare(1, 0, 0, b"pending");
+ let header = *prepare.header();
+ partition.consensus.sequencer().set_sequence(header.op);
+ partition
+ .consensus
+ .set_last_prepare_checksum(header.checksum);
+ partition
+ .log
+ .journal()
+ .inner
+ .append(prepare.clone().into_frozen())
+ .await
+ .unwrap();
+ persistence.append(prepare.into_frozen(), false).unwrap();
+ assert!(persistence.start());
+ Rc::clone(&persistence).run().await;
+ assert!(!persistence.is_durable(&header));
+ partition.consensus.set_view(3);
+ let store = Rc::new(RecordingSuperblock::default());
+ partition.set_superblock(store, None);
+ partition.set_io_notifier(
+ Rc::new(|_, _| {}),
+ crate::largest_legal_job_charge().unwrap(),
+ );
+ assert!(!partition.send_prepare_ok(&header).await);
+ assert!(
+ partition
+ .pending_persisted_acks
+ .borrow()
+ .contains_key(&header.op)
+ );
+ assert!(!partition.persist_superblock_if_needed().await);
+ let config = repair_config();
+ let crate::PartitionIoStep::Ready(plan) = partition.resume_io(&config).await else {
+ panic!("the superblock write must be ready");
+ };
+ assert_eq!(
+ plan.continuation,
+ crate::PartitionIoContinuation::Superblock
+ );
+ execute_partition_io_plan(&mut partition, &config, plan).await;
+ assert!(!partition.consensus.needs_superblock_persist());
+ assert!(!persistence.is_durable(&header));
+ if dispatched {
+ partition.resume_pending_prepare_ack().await;
+ } else {
+ partition.drive_persistence().await;
+ }
+ assert_eq!(
+ partition
+ .pending_persisted_acks
+ .borrow()
+ .contains_key(&header.op),
+ durability == iggy_common::Durability::Persisted,
+ "{durability:?}, dispatched: {dispatched}"
+ );
+ }
+ }
+ }
+
#[compio::test]
async fn persisted_ack_survives_recovery_transfer_and_a_rewound_head() {
let (_directory, mut partition, header) = partition_with_pending_durable_ack().await;
@@ -14396,6 +14515,15 @@ mod tests {
);
assert_eq!(persistence.checkpoint_op(), header.op);
let checkpoint_op = if install_before_completion {
+ partition.maintenance.checkpoint = Some(PendingCheckpoint::Wal(header.op));
+ assert!(matches!(
+ partition.drive_checkpoint_io(&config).unwrap(),
+ Some(crate::PartitionIoStep::Progress)
+ ));
+ assert!(matches!(
+ partition.maintenance.checkpoint,
+ Some(PendingCheckpoint::Reclaim(1))
+ ));
let transferred =
checksummed_segment_prepare(2, header.checksum, 1, b"transferred");
let mut segment_body = prepare.as_slice()[size_of::<PrepareHeader>()..].to_vec();
@@ -14426,6 +14554,7 @@ mod tests {
.install_state_transfer(&config, 2, vec![staged], &offsets.encode(), 0)
.await
.unwrap();
+ settle_partition_io(&mut partition, &config).await;
2
} else {
header.op
@@ -15936,71 +16065,9 @@ mod tests {
);
}
- /// A purge records its reset before it unlinks anything, so a write it
- /// cannot make has to stop the purge while the data proving the old
- /// frontier is still on disk.
#[compio::test]
- async fn given_failing_store_when_purge_records_its_reset_should_refuse_before_mutating() {
- let mut partition = partition_at_view(1, 1);
- let store = Rc::new(RecordingSuperblock::default());
- partition.set_superblock(store.clone(), None);
- partition.offset.store(9_000, Ordering::Release);
- partition.set_offset_space_used(true);
-
- store.fail_writes.set(true);
- assert!(
- matches!(
- partition.record_purge_frontier_reset(7).await,
- Err(PurgeError::FrontierNotRecorded)
- ),
- "the pre-mutation refusal must be distinguishable from a post-drain \
- failure: the caller retries this one and fences the other"
- );
-
- assert!(
- partition.purge_deferred,
- "a deferred purge must fence the ack path: the counter still names the \
- pre-purge offset space, and the view-change persist gate cannot see \
- this because a stable view attempts no write at all"
- );
-
- store.fail_writes.set(false);
- assert!(
- matches!(
- partition.record_purge_frontier_reset(7).await,
- Err(PurgeError::FrontierNotRecorded)
- ),
- "the failed write armed a backoff, and the retry must respect it rather \
- than re-running a full atomic_replace against a disk that just refused one"
- );
- assert_eq!(
- store.attempts.get(),
- 1,
- "the backed-off retry must not reach the store at all"
- );
-
- // Backoff expiry, without a controllable clock in this fixture.
- partition.superblock_retry_after_micros.set(0);
- partition
- .record_purge_frontier_reset(7)
- .await
- .expect("a working store records the reset once the backoff elapses");
- assert!(
- !partition.purge_deferred,
- "recording the reset releases the fence"
- );
- assert_eq!(
- last_recorded_frontier(&store),
- 0,
- "the reset is spelled out, not read off a counter still holding the \
- pre-purge frontier"
- );
- }
-
- #[test]
- fn purge_offset_cleanup_reserves_bytes_for_durable_and_stranded_ids() {
+ async fn purge_offset_cleanup_reserves_bytes_for_durable_and_stranded_ids() {
for consumer_kind in ConsumerKind::ALL {
- let kind = consumer_kind.index();
let (mut partition, _) = recording_partition();
partition.consumer_offset_dirs =
["consumers", "groups", "external_groups"].map(|dir| Some(dir.to_owned()));
@@ -16011,7 +16078,7 @@ mod tests {
partition.transition = Some(PendingPartitionTransition::Purge(PendingPurge {
generation: 1,
drain: None,
- phase: PurgePhase::Offsets(kind),
+ phase: PurgePhase::Offsets(Some(consumer_kind)),
record_only: false,
}));
let crate::PartitionIoStep::Ready(empty_plan) = partition
@@ -16036,10 +16103,7 @@ mod tests {
partition
.io_bytes_max
.set(crate::largest_legal_job_charge().unwrap());
- let crate::PartitionIoStep::Ready(plan) = partition
- .io_plan(crate::PartitionIoContinuation::Purge)
- .unwrap()
- else {
+ let crate::PartitionIoStep::Ready(plan) = partition.drive_purge_io().await else {
panic!("the full budget must admit offset cleanup");
};
let captured = partition
@@ -16082,15 +16146,14 @@ mod tests {
for _ in 0..STEPS_MAX {
match partition.resume_io(&config).await {
crate::PartitionIoStep::Ready(plan) => {
- let captured = partition.capture_io(plan, &config).unwrap().unwrap();
- let result = captured.job.execute().await;
- partition.accept_io(captured.identity, result).unwrap();
- if let Some(gate) = captured.gate {
- gate.release();
- }
+ execute_partition_io_plan(&mut partition, &config, plan).await;
}
crate::PartitionIoStep::PurgeFinished { outcome, .. } => {
assert!(matches!(outcome, Err(PurgeError::FrontierNotRecorded)));
+ assert!(
+ partition.purge_deferred,
+ "the failed reset must withhold prepare acknowledgments"
+ );
aborted = true;
break;
}
@@ -16150,12 +16213,7 @@ mod tests {
for _ in 0..STEPS_MAX {
match partition.resume_io(&config).await {
crate::PartitionIoStep::Ready(plan) => {
- let captured = partition.capture_io(plan, &config).unwrap().unwrap();
- let result = captured.job.execute().await;
- partition.accept_io(captured.identity, result).unwrap();
- if let Some(gate) = captured.gate {
- gate.release();
- }
+ execute_partition_io_plan(&mut partition, &config, plan).await;
}
crate::PartitionIoStep::QuarantineFinished(outcome) => {
assert!(outcome.is_ok());
@@ -17009,7 +17067,76 @@ mod tests {
}
#[compio::test]
- async fn missing_materialization_resumes_deferred_recovery_control_after_writer_settles() {
+ async fn deferred_views_keep_newer_progress_without_replacing_history_transitions() {
+ let (mut partition, _) = recording_partition();
+ let start_change = |view| {
+ server_common::MessageBag::StartViewChange(
+ Message::<iggy_binary_protocol::StartViewChangeHeader>::new(size_of::<
+ iggy_binary_protocol::StartViewChangeHeader,
+ >())
+ .transmute_header(
+ |_, header: &mut iggy_binary_protocol::StartViewChangeHeader| {
+ header.command = Command::StartViewChange;
+ header.size =
+ u32::try_from(size_of::<iggy_binary_protocol::StartViewChangeHeader>())
+ .unwrap();
+ header.view = view;
+ },
+ ),
+ )
+ };
+ let start_view = |view| {
+ server_common::MessageBag::StartView(
+ Message::<iggy_binary_protocol::StartViewHeader>::new(size_of::<
+ iggy_binary_protocol::StartViewHeader,
+ >())
+ .transmute_header(
+ |_, header: &mut iggy_binary_protocol::StartViewHeader| {
+ header.command = Command::StartView;
+ header.size =
+ u32::try_from(size_of::<iggy_binary_protocol::StartViewHeader>())
+ .unwrap();
+ header.view = view;
+ },
+ ),
+ )
+ };
+ partition.defer_view_transition(start_change(1));
+ partition.defer_view_transition(start_view(1));
+ partition.defer_view_transition(start_change(1));
+ partition.defer_view_transition(start_view(0));
+ assert!(matches!(partition.resume_io(&repair_config()).await,
+ crate::PartitionIoStep::Transition(server_common::MessageBag::StartView(message)) if message.header().view == 1));
+ partition.defer_view_transition(start_view(1));
+ partition.defer_view_transition(start_change(2));
+ assert!(matches!(partition.resume_io(&repair_config()).await,
+ crate::PartitionIoStep::Transition(server_common::MessageBag::StartViewChange(message)) if message.header().view == 2));
+ partition.transition = Some(PendingPartitionTransition::Truncate {
+ from_op: 3,
+ actions: Vec::new(),
+ peer: None,
+ drain: None,
+ });
+ partition.defer_view_transition(start_view(3));
+ assert!(matches!(
+ partition.transition,
+ Some(PendingPartitionTransition::Truncate { from_op: 3, .. })
+ ));
+ partition.transition = Some(PendingPartitionTransition::Purge(PendingPurge {
+ generation: 1,
+ drain: None,
+ phase: PurgePhase::Frontier,
+ record_only: false,
+ }));
+ partition.defer_view_transition(start_view(4));
+ assert!(matches!(
+ partition.transition,
+ Some(PendingPartitionTransition::Purge(_))
+ ));
+ }
+
+ #[compio::test]
+ async fn missing_materialization_allows_deferred_recovery_control() {
let (mut partition, _) = recording_partition();
partition.materialization_missing = true;
partition.set_io_notifier(
@@ -17031,7 +17158,6 @@ mod tests {
#[compio::test]
async fn transfer_install_waits_for_captured_writer_before_replacing_history() {
- const INSTALL_STEPS_MAX: usize = 64;
let directory = tempfile::tempdir().unwrap();
let config = repair_config();
let mut partition = partition_at_view(1, 1);
@@ -17088,34 +17214,17 @@ mod tests {
partition.accept_io(captured.identity, result).unwrap();
captured.gate.unwrap().release();
- for _ in 0..INSTALL_STEPS_MAX {
- match partition.resume_io(&config).await {
- crate::PartitionIoStep::Ready(plan) => {
- let captured = partition.capture_io(plan, &config).unwrap().unwrap();
- let result = captured.job.execute().await;
- partition.accept_io(captured.identity, result).unwrap();
- if let Some(gate) = captured.gate {
- gate.release();
- }
- }
- crate::PartitionIoStep::Progress | crate::PartitionIoStep::Pending => {}
- crate::PartitionIoStep::InstallFinished { outcome, .. } => {
- assert_eq!(outcome.unwrap().applied_commit_op, 1);
- assert_eq!(partition.consensus().commit_min(), 1);
- assert_ne!(partition.poll_history, old_history);
- assert!(partition.fatal().is_none());
- assert_eq!(
- std::fs::metadata(directory.path().join("00000000000000000000.log"))
- .unwrap()
- .len(),
- 0
- );
- return;
- }
- _ => panic!("install produced an unrelated continuation"),
- }
- }
- panic!("install did not finish after its old writer settled");
+ let outcome = finish_install_io(&mut partition).await.unwrap();
+ assert_eq!(outcome.applied_commit_op, 1);
+ assert_eq!(partition.consensus().commit_min(), 1);
+ assert_ne!(partition.poll_history, old_history);
+ assert!(partition.fatal().is_none());
+ assert_eq!(
+ std::fs::metadata(directory.path().join("00000000000000000000.log"))
+ .unwrap()
+ .len(),
+ 0
+ );
}
/// Runs the tick's superblock step between install steps, as the shard does.
@@ -17131,12 +17240,7 @@ mod tests {
let _ = partition.persist_superblock_if_needed().await;
match partition.resume_io(&config).await {
crate::PartitionIoStep::Ready(plan) => {
- let captured = partition.capture_io(plan, &config).unwrap().unwrap();
- let result = captured.job.execute().await;
- partition.accept_io(captured.identity, result).unwrap();
- if let Some(gate) = captured.gate {
- gate.release();
- }
+ execute_partition_io_plan(partition, &config, plan).await;
}
crate::PartitionIoStep::Progress => {}
crate::PartitionIoStep::Pending => {
@@ -19071,7 +19175,7 @@ mod tests {
);
partition.begin_shutdown_io();
- for commit_max in 1..=2 {
+ for commit_max in 1..=3 {
partition.consensus.advance_commit_max(commit_max);
for _ in 0..STEPS_MAX {
settle_partition_io(&mut partition, &config).await;
@@ -19080,7 +19184,10 @@ mod tests {
}
compio::time::sleep(std::time::Duration::from_millis(1)).await;
}
- assert!(partition.shutdown_io_complete());
+ assert!(
+ partition.shutdown_io_complete(),
+ "shutdown must finish the flush of commit {commit_max}"
+ );
assert!(persistence.is_quiescent());
assert_eq!(persistence.head(), 2);
partition.acknowledge_prepare(3).await;
@@ -19668,6 +19775,13 @@ mod tests {
let auto_commit = true;
let consumer = PollingConsumer::ConsumerGroup(group_id, member_id);
let old_result = poll_read_result(&partition, consumer, auto_commit, Some(4));
+ for kind in ConsumerKind::ALL {
+ let path = partition.persisted_offset_path(kind, 42).unwrap();
+ persist_offset(&path, 0, true).await.unwrap();
+ partition
+ .consumer_offset_capacity_for(kind)
+ .record_stranded(42);
+ }
// Describe staged files without creating them. The swap fails after
// preflight, forcing recovery to replace the served log with an empty one.
@@ -19714,6 +19828,19 @@ mod tests {
"unexpected failure: {install_error:?}"
);
+ for kind in ConsumerKind::ALL {
+ assert!(
+ !std::path::Path::new(&partition.persisted_offset_path(kind, 42).unwrap()).exists()
+ );
+ assert_eq!(
+ partition
+ .consumer_offset_capacity_for(kind)
+ .stranded_count(),
+ 0,
+ "convergence must release removed offset files for {kind:?}"
+ );
+ }
+
// Failure did not preserve the old history: recovery left an empty log,
// and its owner must reject the result captured before installation.
assert_eq!(partition.log.segments().len(), 1);
@@ -20199,7 +20326,7 @@ mod tests {
partition.pending_consumer_offset_commits.clear();
partition
.consumer_offset_capacity_for(ConsumerKind::Consumer)
- .release_reservation(7);
+ .set_pending_count(7, 0);
partition.consensus.set_view(1);
partition.resynchronize_consumer_offset_reservations();
diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs
index 3a00309f0..027d58157 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -195,7 +195,8 @@ where
*self.io_notifier.borrow_mut() = Some((notifier, bytes_max));
}
- /// Only physical result settlement may look through the tombstone gate.
+ /// Access the physical owner through the tombstone gate for I/O settlement,
+ /// shutdown, fault scans, and ticks. Admission must use the live lookup.
#[allow(clippy::mut_from_ref)]
pub fn get_io_owner(&self, namespace: &IggyNamespace) -> Option<&mut IggyPartition<B, SB>> {
let local = self.namespace_map().get(namespace).copied()?;
@@ -586,9 +587,6 @@ where
/// `ReconcileOp::ConfirmRemove` after the partition is dropped.
pub fn untombstone(&self, namespace: &IggyNamespace) {
self.tombstoned.borrow_mut().remove(namespace);
- if let Some(partition) = self.get_by_ns(namespace) {
- self.install_loopback_notifier(*namespace, partition);
- }
}
/// Snapshot read resources under a synchronous borrow on the owning pump.
@@ -995,12 +993,8 @@ mod tests {
partitions.tombstone(namespace);
assert!(!partitions.has_ready_loopbacks());
- partitions.untombstone(&namespace);
- assert!(
- partitions.has_ready_loopbacks(),
- "unmount cancellation restores queued work"
- );
let retired = partitions.remove(&namespace).unwrap();
+ partitions.untombstone(&namespace);
assert!(!partitions.has_ready_loopbacks());
let mut messages = Vec::new();
retired.consensus().drain_loopback_into(&mut messages);
diff --git a/core/partitions/src/io.rs b/core/partitions/src/io.rs
index ec964e106..5ecb4acdd 100644
--- a/core/partitions/src/io.rs
+++ b/core/partitions/src/io.rs
@@ -413,6 +413,8 @@ pub enum TransferFileJob {
Converge {
directory: String,
offset_directories: [Option<String>; ConsumerKind::COUNT],
+ // Keep recovery-only snapshots out of every file job's inline state.
+ stranded_offsets: Box<[HashSet<u32>; ConsumerKind::COUNT]>,
segment: SegmentIoJob,
},
}
@@ -423,7 +425,8 @@ pub enum TransferFileResult {
Opened(Result<InstalledSegment, crate::state_transfer::PartitionInstallError>),
Converged {
segment: Result<InstalledSegment, IggyError>,
- stranded: Option<(usize, u32)>,
+ stranded: Option<(ConsumerKind, u32)>,
+ released: Vec<(ConsumerKind, u32)>,
},
}
@@ -997,6 +1000,8 @@ impl TransferFileJob {
};
let mut storage = match open().await {
Ok(storage) => storage,
+ // A transient open failure need not fence durable history
+ // or force a complete pull for a partition without a WAL.
Err(_) => open().await?,
};
// Only the tail takes writes, as after a rotation.
@@ -1078,9 +1083,12 @@ impl TransferFileJob {
Self::Converge {
directory,
offset_directories,
+ mut stranded_offsets,
segment,
} => {
let mut stranded = None;
+ let mut released =
+ Vec::with_capacity(stranded_offsets.iter().map(HashSet::len).sum());
let outcome = async {
for (kind, path) in offset_directories.into_iter().enumerate() {
let Some(path) = path else {
@@ -1104,13 +1112,16 @@ impl TransferFileJob {
})
.await
{
- stranded = Some((kind, id));
+ stranded = Some((ConsumerKind::ALL[kind], id));
return Err(error);
}
+ if stranded_offsets[kind].remove(&id) {
+ released.push((ConsumerKind::ALL[kind], id));
+ }
} else if entry.file_name().to_str().is_some_and(|name| {
crate::offset_storage::offset_replacement_id(name).is_some()
- }) {
- let _ = compio::fs::remove_file(path).await;
+ }) && let Err(error) = compio::fs::remove_file(path).await {
+ warn!(%path, %error, "cannot remove abandoned offset replacement during convergence");
}
}
match crate::state_transfer::fsync_dir(&path).await {
@@ -1150,6 +1161,7 @@ impl TransferFileJob {
TransferFileResult::Converged {
segment: outcome,
stranded,
+ released,
}
}
}
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index f5c7bd6f1..2ebd04c28 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -21,6 +21,7 @@ use server_common::{
iobuf::{Frozen, Owned},
send_messages::{self, BatchRef, COMMAND_HEADER_SIZE, decode_prepare_slice_trusted},
};
+use smallvec::SmallVec;
use std::io;
use std::{
cell::{Cell, UnsafeCell},
@@ -29,7 +30,7 @@ use std::{
};
use tracing::warn;
-use crate::{Fragment, PollFragments, PollQueryResult};
+use crate::{COMMIT_WALK_OPS_MAX, Fragment, PollFragments, PollQueryResult};
const ZERO_LEN: usize = 0;
const PREPARE_HEADER_SIZE: usize = std::mem::size_of::<PrepareHeader>();
@@ -544,7 +545,8 @@ impl PartitionJournal<PartitionJournalMemStorage> {
/// Pin the contiguous resident prefix through `commit_max`, stopping at gaps.
/// Entries remain resident until their physical writes have been accepted.
- pub fn committed_prefix(&self, commit_max: u64) -> Vec<JournalBuffer> {
+ #[cfg(test)]
+ pub(crate) fn committed_prefix(&self, commit_max: u64) -> Vec<JournalBuffer> {
let count = self.committed_prefix_len(commit_max);
let inner = unsafe { &*self.inner.get() };
let entries = unsafe { &*inner.storage.entries.get() };
@@ -648,7 +650,7 @@ impl PartitionJournal<PartitionJournalMemStorage> {
///
/// INVARIANT (length-lock): the header is pushed before `storage.write_at`,
/// so `headers[i]` and the entry at storage index `i` stay positionally
- /// paired - `committed_prefix`'s zip relies on that. `MemStorage::write_at`
+ /// paired - materialization reads entries by that position. `MemStorage::write_at`
/// is infallible, so the push never runs ahead of a failed write. A future
/// fallible `Storage` MUST roll the header push back on a write error (or
/// write before pushing the header) or the zip desyncs.
@@ -814,6 +816,59 @@ where
headers.iter().find(|header| header.op == op).copied()
}
+ pub(crate) fn headers_for_commit(
+ &self,
+ from_op: u64,
+ count: usize,
+ ) -> Vec<Option<PrepareHeader>> {
+ if count == 1 {
+ return vec![self.header_by_op(from_op)];
+ }
+ let mut selected = vec![None; count];
+ let headers = unsafe { &*self.headers.get() };
+ for header in headers {
+ if let Some(index) = header
+ .op
+ .checked_sub(from_op)
+ .and_then(|index| usize::try_from(index).ok())
+ && let Some(slot) = selected.get_mut(index)
+ {
+ slot.get_or_insert(*header);
+ }
+ }
+ selected
+ }
+
+ pub(crate) fn commit_headers_match(
+ &self,
+ from_op: u64,
+ expected: &[Option<PrepareHeader>],
+ ) -> bool {
+ if let [header] = expected {
+ return self.header_by_op(from_op) == *header;
+ }
+ let mut seen = SmallVec::<[bool; COMMIT_WALK_OPS_MAX]>::new();
+ seen.resize(expected.len(), false);
+ let headers = unsafe { &*self.headers.get() };
+ for header in headers {
+ if let Some(index) = header
+ .op
+ .checked_sub(from_op)
+ .and_then(|index| usize::try_from(index).ok())
+ && let Some(seen) = seen.get_mut(index)
+ && !*seen
+ {
+ if expected[index] != Some(*header) {
+ return false;
+ }
+ *seen = true;
+ }
+ }
+ seen.iter()
+ .zip(expected)
+ .all(|(seen, header)| *seen == header.is_some())
+ }
+
/// Test residency without scanning the header vector.
///
/// Synchronous insertion and eviction keep the storage map and headers
@@ -1102,7 +1157,7 @@ impl Journal for PartitionJournal<PartitionJournalMemStorage> {
/// drain-and-re-append shape as `evict_prefix`, from the other end and retaining
/// nothing: `append` has no slot-collision check here, so a superseded entry left
/// in place sits beside the new view's prepare at the same op and
- /// `committed_prefix`, which walks positionally, flushes the stale one.
+ /// positional materialization flushes the stale one.
///
/// Dropped entries do NOT enter the evicted repair ring: it answers repair for
/// committed ops, and these are ones the view just decided against.
@@ -1605,6 +1660,54 @@ mod tests {
);
}
+ #[compio::test]
+ async fn commit_header_snapshot_handles_repair_order_gaps_and_replacements() {
+ let journal = PartitionJournal::<PartitionJournalMemStorage>::default();
+ for op in [3, 1, 4] {
+ journal
+ .append(build_prepare(op, HEADER_SIZE + 16).into_frozen())
+ .await
+ .unwrap();
+ }
+ let selected = journal.headers_for_commit(1, 4);
+ assert_eq!(
+ selected
+ .iter()
+ .map(|header| header.map(|header| header.op))
+ .collect::<Vec<_>>(),
+ vec![Some(1), None, Some(3), Some(4)]
+ );
+ assert!(journal.commit_headers_match(1, &selected));
+ journal
+ .append(build_prepare(2, HEADER_SIZE + 16).into_frozen())
+ .await
+ .unwrap();
+ assert!(
+ !journal.commit_headers_match(1, &selected),
+ "repair filled a previously missing header"
+ );
+ let selected = journal.headers_for_commit(1, 4);
+ assert!(journal.commit_headers_match(1, &selected));
+ let duplicate = build_prepare(1, HEADER_SIZE + 16).transmute_header(
+ |mut old, header: &mut PrepareHeader| {
+ old.checksum ^= 1;
+ *header = old;
+ },
+ );
+ journal.append(duplicate.into_frozen()).await.unwrap();
+ assert_eq!(
+ journal.headers_for_commit(1, 4),
+ selected,
+ "the snapshot must agree with header_by_op's first resident match"
+ );
+ assert!(journal.commit_headers_match(1, &selected));
+ journal.truncate_from(3).await.unwrap();
+ assert!(
+ !journal.commit_headers_match(1, &selected),
+ "a truncated selection must not publish"
+ );
+ }
+
#[compio::test]
async fn committed_prefix_reads_then_evict_retains_uncommitted_tail() {
// A backup journals ops ahead of the commit frontier. Reading the
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 30800400f..dc515d2c3 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -35,9 +35,9 @@ mod persistence;
mod poll_plan;
#[cfg(feature = "simulator")]
pub use persistence::CheckpointBarrier;
+pub(crate) use persistence::PersistenceDrain;
pub use persistence::{
- PartitionPersistence, PersistenceCompletion, PersistenceDrain, PersistenceMetrics,
- PersistenceNotifier,
+ PartitionPersistence, PersistenceCompletion, PersistenceMetrics, PersistenceNotifier,
};
mod segment;
pub mod segment_anchor;
@@ -53,11 +53,12 @@ pub use iggy_index_reader::IggyIndexReader;
pub use iggy_index_writer::IggyIndexWriter;
pub use iggy_partition::{IggyPartition, PurgeError, SegmentRemoval};
pub use iggy_partitions::IggyPartitions;
+pub(crate) use io::PartitionIoVerdict;
pub use io::{
CapturedPartitionIo, MaterializationIoJob, MaterializationIoResult, PartitionIncarnation,
PartitionIoContinuation, PartitionIoIdentity, PartitionIoJob, PartitionIoNotifier,
PartitionIoPlan, PartitionIoQuiescence, PartitionIoResources, PartitionIoResult,
- PartitionIoStep, PartitionIoVerdict, PartitionTeardown, largest_legal_job_charge,
+ PartitionIoStep, PartitionTeardown, largest_legal_job_charge,
};
pub use journal::{EVICTED_RING_BYTES_MAX, EVICTED_RING_CAPACITY};
diff --git a/core/partitions/src/offset_storage.rs b/core/partitions/src/offset_storage.rs
index 63fd28c95..fc4bb63fc 100644
--- a/core/partitions/src/offset_storage.rs
+++ b/core/partitions/src/offset_storage.rs
@@ -200,10 +200,6 @@ impl<F> RetainedOffsetFiles<F> {
self.retired.borrow_mut().clear();
}
- pub(crate) fn retired_count(&self) -> usize {
- self.retired.borrow().len()
- }
-
pub(crate) fn take_checkpoint(&self) -> (Vec<RetainedOffsetFile<F>>, BTreeSet<PathBuf>) {
let paths = self
.files
@@ -825,7 +821,7 @@ mod tests {
let (_, replacement) = cache.checkout("offset").unwrap();
cache.put("offset", 9, replacement);
cache.put("offset", 7, original);
- assert_eq!(cache.retired_count(), 1);
+ assert_eq!(cache.retired.borrow().len(), 1);
let replacement = cache.take("offset").unwrap();
assert_eq!(replacement.file, 9);
cache.clear();
diff --git a/core/partitions/src/persistence.rs b/core/partitions/src/persistence.rs
index 1bb173e88..ef0964ef5 100644
--- a/core/partitions/src/persistence.rs
+++ b/core/partitions/src/persistence.rs
@@ -1028,7 +1028,6 @@ impl<S: DurableStorage> PartitionPersistence<S> {
&& (self.checkpoint_needed.get()
|| self.retained_bytes.get() + self.queued_bytes.get() + self.in_flight_bytes.get()
>= self.capacity / 2
- || self.offset_files.retired_count() >= CHECKPOINT_DIRTY_FILES_MAX
|| self.dirty_segments.borrow().len() * 2
+ self
.dirty_offsets
@@ -1107,7 +1106,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
self.queued_bytes.set(0);
}
- pub fn begin_drain(&self) -> PersistenceDrain {
+ pub(crate) fn begin_drain(&self) -> PersistenceDrain {
self.enqueue_paused.set(true);
PersistenceDrain {
instance: self.instance,
@@ -1122,7 +1121,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
/// # Errors
/// Preserves storage/interruption failures and the original drain deadline.
- pub fn observe_drain(&self, drain: &PersistenceDrain) -> io::Result<bool> {
+ pub(crate) fn observe_drain(&self, drain: &PersistenceDrain) -> io::Result<bool> {
if drain.instance != self.instance || drain.epoch != self.epoch.get() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
@@ -1146,7 +1145,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
Ok(false)
}
- pub fn finish_drain(&self, drain: &PersistenceDrain) {
+ pub(crate) fn finish_drain(&self, drain: &PersistenceDrain) {
if drain.instance == self.instance && drain.epoch == self.epoch.get() {
self.enqueue_paused.set(false);
}
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index b45590bec..c03bbdb96 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -41,6 +41,7 @@ use consensus::state_manifest::artifact_kind;
use consensus::{
ArtifactProgress, DedupWatermark, Sequencer as _, StateArtifactHasher, state_artifact_checksum,
};
+use futures::channel::oneshot;
use iggy_binary_protocol::responses::messages::send_messages::CONFIRMATION_SIZE;
use iggy_binary_protocol::{Operation, PrepareHeader, ReplyHeader};
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset};
@@ -64,6 +65,7 @@ use std::sync::atomic::Ordering;
/// Current state-transfer offsets format, including the prepare-chain anchor.
pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO1";
pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 4;
+pub(crate) const SEGMENT_SWEEP_STACK_BYTES: usize = 2 * 1024 * 1024;
const RETRY_CHECKPOINT_MAGIC: [u8; 4] = *b"IRP2";
const RETRY_CHECKPOINT_PREFIX: &str = "receipts-";
@@ -1880,7 +1882,22 @@ fn segment_dir_entries(partition_dir: &str) -> std::io::Result<impl Iterator<Ite
/// Remove physical tails outside the logical segment list after draining the WAL.
/// The owner has settled its writers and published a purge or install backup.
pub(crate) async fn remove_public_segment_files(partition_dir: &str) -> std::io::Result<()> {
- let directory = Path::new(partition_dir);
+ let directory = PathBuf::from(partition_dir);
+ let (completed, completion) = oneshot::channel();
+ // Directory iteration and metadata probes must not block the shard. The
+ // admitted job owns this worker, with one directory entry live at a time.
+ std::thread::Builder::new()
+ .name("iggy-segment-sweep".to_owned())
+ .stack_size(SEGMENT_SWEEP_STACK_BYTES)
+ .spawn(move || {
+ let _ = completed.send(remove_public_segment_files_blocking(&directory));
+ })?;
+ completion
+ .await
+ .map_err(|_| std::io::Error::other("segment sweep worker stopped"))?
+}
+
+fn remove_public_segment_files_blocking(directory: &Path) -> std::io::Result<()> {
for entry in std::fs::read_dir(directory)? {
let entry = entry?;
let name = entry.path();
@@ -1893,7 +1910,7 @@ pub(crate) async fn remove_public_segment_files(partition_dir: &str) -> std::io:
.and_then(|stem| stem.to_str())
.is_some_and(|stem| stem.parse::<u64>().is_ok())
{
- match DiskStorage.remove_file(&name).await {
+ match std::fs::remove_file(&name) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error),
@@ -2136,7 +2153,7 @@ enum InstallPhase {
RenameLogs(usize),
OpenSegments(usize),
EmptyDirectory,
- OldOffsets(usize),
+ OldOffsets(Option<ConsumerKind>),
CommitOffsets(usize),
OffsetDirectories,
Publish,
@@ -2973,22 +2990,37 @@ where
InstallPhase::StageOffsets(_) | InstallPhase::DiscardOffsets(_) => {
base.checked_mul(OFFSET_PERSIST_CONCURRENCY)
}
- InstallPhase::Sweep => pending.staged.iter().try_fold(base, |charge, segment| {
- charge
- .checked_add(segment.log_staging.capacity())?
- .checked_add(segment.index_staging.capacity())?
- .checked_add(2 * (size_of::<PathBuf>() + 4 * size_of::<&Path>() + 4))
- }),
- InstallPhase::OldOffsets(kind) => {
+ InstallPhase::Sweep => pending.staged.iter().try_fold(
+ base.checked_add(SEGMENT_SWEEP_STACK_BYTES)?,
+ |charge, segment| {
+ charge
+ .checked_add(segment.log_staging.capacity())?
+ .checked_add(segment.index_staging.capacity())?
+ .checked_add(2 * (size_of::<PathBuf>() + 4 * size_of::<&Path>() + 4))
+ },
+ ),
+ InstallPhase::OldOffsets(Some(kind)) => {
let retained = pending
.offsets_wire
.offset_sections()
- .get(kind)
+ .get(kind.index())
.map_or(0, |entries| entries.len());
base.checked_add(self.offset_cleanup_charge(kind)?)?
.checked_add(retained.checked_mul(size_of::<u32>())?)
}
InstallPhase::RetryCheckpoint => base.checked_add(pending.retry_checkpoint.capacity()),
+ InstallPhase::Converge => ConsumerKind::ALL.into_iter().try_fold(
+ base.checked_add(size_of::<[HashSet<u32>; ConsumerKind::COUNT]>())?,
+ |charge, kind| {
+ charge
+ .checked_add(self.offset_cleanup_charge(kind)?)?
+ .checked_add(
+ self.consumer_offset_capacity_for(kind)
+ .stranded_count()
+ .checked_mul(size_of::<(ConsumerKind, u32)>())?,
+ )
+ },
+ ),
_ => Some(base),
}
}
@@ -3044,6 +3076,7 @@ where
),
crate::PartitionIoVerdict::Ready => {
install.disposition = InstallDisposition::Mutating;
+ self.cancel_history_maintenance();
self.invalidate_poll_history();
self.log.invalidate_sealed_read_state();
self.segment_checksum_cache.borrow_mut().clear();
@@ -3067,13 +3100,14 @@ where
}
InstallPhase::OpenSegments(cursor) if cursor > 0 && cursor >= install.staged.len() => {
self.clear_install_offsets();
- install.phase = InstallPhase::OldOffsets(0);
+ install.phase = InstallPhase::OldOffsets(Some(ConsumerKind::Consumer));
}
- InstallPhase::OldOffsets(ConsumerKind::COUNT) => {
+ InstallPhase::OldOffsets(None) => {
install.phase = InstallPhase::CommitOffsets(0);
}
- InstallPhase::OldOffsets(kind) if self.purge_offset_directory(kind).is_none() => {
- install.phase = InstallPhase::OldOffsets(kind + 1);
+ InstallPhase::OldOffsets(Some(kind)) if self.consumer_offsets_dir(kind).is_none() => {
+ install.phase =
+ InstallPhase::OldOffsets(crate::iggy_partition::next_consumer_kind(kind));
}
InstallPhase::CommitOffsets(cursor) if cursor == install.planned_offsets.len() => {
install.phase = InstallPhase::OffsetDirectories;
@@ -3387,9 +3421,8 @@ where
crate::PartitionIoJob::SegmentDirectory(install.partition_dir.clone()),
InstallFilePhase::EmptyDirectory,
),
- InstallPhase::OldOffsets(kind) => {
- let consumer_kind = crate::iggy_partition::purge_offset_kind(*kind)
- .ok_or(iggy_common::IggyError::CannotDeleteFile)?;
+ InstallPhase::OldOffsets(Some(kind)) => {
+ let consumer_kind = *kind;
// Replacements atomically overwrite their old files, so only keys
// absent from the plan are unlinked.
let mut retained: Vec<u32> = install
@@ -3471,6 +3504,12 @@ where
directory: install.partition_dir.clone(),
offset_directories: ConsumerKind::ALL
.map(|kind| self.consumer_offsets_dir(kind).map(str::to_owned)),
+ stranded_offsets: Box::new(ConsumerKind::ALL.map(|kind| {
+ let capacity = self.consumer_offset_capacity_for(kind);
+ let mut known = HashSet::with_capacity(capacity.stranded_count());
+ capacity.extend_stranded_ids(&mut known);
+ known
+ })),
segment: self.prepare_empty_segment(config, install.offsets_wire.next_offset),
}),
InstallFilePhase::Converge,
@@ -3579,7 +3618,7 @@ where
) => outcome
.map(|()| {
self.clear_install_offsets();
- InstallPhase::OldOffsets(0)
+ InstallPhase::OldOffsets(Some(ConsumerKind::Consumer))
})
.map_err(|source| PartitionInstallError::SwapIo {
path: install.partition_dir.clone(),
@@ -3614,7 +3653,9 @@ where
source: iggy_common::IggyError::CannotDeleteFile,
})
} else {
- Ok(InstallPhase::OldOffsets(kind + 1))
+ Ok(InstallPhase::OldOffsets(
+ crate::iggy_partition::next_consumer_kind(consumer_kind),
+ ))
}
}
(
@@ -3668,12 +3709,14 @@ where
crate::PartitionIoResult::Transfer(crate::io::TransferFileResult::Converged {
segment,
stranded,
+ released,
}),
) => {
self.clear_converged_state(&install);
- if let Some((kind, id)) = stranded
- && let Some(kind) = crate::iggy_partition::purge_offset_kind(kind)
- {
+ for (kind, id) in released {
+ self.consumer_offset_capacity_for(kind).clear_stranded(id);
+ }
+ if let Some((kind, id)) = stranded {
self.consumer_offset_capacity_for(kind).record_stranded(id);
}
match segment {
diff --git a/core/server/config.toml b/core/server/config.toml
index 5a826f939..5e60c3fde 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -760,9 +760,11 @@ reply_inbox_capacity = 1024
# separate capacity. This limits operation count, not retained message bytes.
poll_completion_capacity = 1024
-# Active partition file jobs plus results awaiting owner acceptance per shard.
+# Active partition file jobs plus results awaiting owner acceptance per shard (1..=256).
partition_io_capacity = 16
-# Omit partition_io_bytes_max to resolve max(256 MiB, the largest legal job).
+# Omission selects the largest legal job charge, slightly above 4 GiB per shard.
+# This is an admission ceiling, not an eager allocation. A smaller budget fails boot.
+# partition_io_bytes_max = "8 GiB"
# Charges include retained backing allocations, even for small visible slices.
# Wall-clock budget for a single shard's bus drain on shutdown. Drives
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index 571ac42ea..30ec34589 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -292,7 +292,12 @@ pub(in crate::boot) async fn build_shard_for_thread(
)
.with_partition_io_limits(shard::PartitionIoLimits::new(
config.sharding.partition_io_capacity,
- config.sharding.partition_io_bytes_max,
+ config
+ .sharding
+ .partition_io_bytes_max
+ .map(|bytes| usize::try_from(bytes.as_bytes_u64()))
+ .transpose()
+ .map_err(|_| shard::PartitionIoLimitsError::Overflow)?,
)?)
.build()
.map_err(ServerError::ShardConstruction)?;
diff --git a/core/server/src/boot/threads.rs b/core/server/src/boot/threads.rs
index 74540578f..56a285863 100644
--- a/core/server/src/boot/threads.rs
+++ b/core/server/src/boot/threads.rs
@@ -569,7 +569,11 @@ pub(in crate::boot) fn validate_sharding_runtime_knobs(
let inbox_capacity = sharding.inbox_capacity;
shard::PartitionIoLimits::new(
sharding.partition_io_capacity,
- sharding.partition_io_bytes_max,
+ sharding
+ .partition_io_bytes_max
+ .map(|bytes| usize::try_from(bytes.as_bytes_u64()))
+ .transpose()
+ .map_err(|_| shard::PartitionIoLimitsError::Overflow)?,
)?;
if inbox_capacity == 0 || inbox_capacity > INBOX_CAPACITY_MAX {
return Err(ServerError::InvalidInboxCapacity {
diff --git a/core/server/src/consumer_group.rs b/core/server/src/consumer_group.rs
index f7b45b3a1..50513cbc7 100644
--- a/core/server/src/consumer_group.rs
+++ b/core/server/src/consumer_group.rs
@@ -799,6 +799,7 @@ mod tests {
inbox,
replies,
POLL_COMPLETION_CAPACITY,
+ None,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, replica_count), bus),
None,
diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs
index 21733e720..1a421e326 100644
--- a/core/server/src/dispatch/mod.rs
+++ b/core/server/src/dispatch/mod.rs
@@ -1145,6 +1145,7 @@ mod tests {
inbox_rx,
reply_inbox_rx,
POLL_COMPLETION_CAPACITY,
+ None,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
None,
diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs
index d67b106b9..69d9ce1bf 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -2375,6 +2375,7 @@ mod tests {
inbox_rx,
reply_inbox_rx,
POLL_COMPLETION_CAPACITY,
+ None,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
None,
diff --git a/core/server_common/src/sharding/mod.rs b/core/server_common/src/sharding/mod.rs
index 6e938c595..00c70beed 100644
--- a/core/server_common/src/sharding/mod.rs
+++ b/core/server_common/src/sharding/mod.rs
@@ -29,6 +29,9 @@ pub use namespace::{
pub use partition_location::PartitionLocation;
pub use shard_id::ShardId;
+/// Slot lookup is linear on the shard pump; bound its worst-case scan.
+pub const PARTITION_IO_CAPACITY_MAX: usize = 256;
+
/// Maximum time a client-list gather waits for all shard replies.
pub const LIST_CLIENTS_GATHER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
diff --git a/core/shard/src/builder.rs b/core/shard/src/builder.rs
index 8796378a2..d13b681a8 100644
--- a/core/shard/src/builder.rs
+++ b/core/shard/src/builder.rs
@@ -222,7 +222,7 @@ where
None
};
- let mut shard = IggyShard::new(
+ let shard = IggyShard::new(
self.identity,
self.bus,
self.host,
@@ -232,16 +232,13 @@ where
self.inbox,
self.reply_inbox,
self.poll_completion_capacity,
+ self.partition_io_limits,
self.shards_table,
self.partition_consensus,
coordinator,
self.metrics,
)?;
- if let Some(limits) = self.partition_io_limits {
- shard.partition_io = crate::partition_io::PartitionIoLane::new(limits);
- }
-
Ok(BuiltShard { shard })
}
}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 1b64b8a1d..bed9f7086 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1746,6 +1746,7 @@ where
/// pump (client `Reply` forwards only; see [`TaggedSender::reply_sender`]).
/// * `poll_completion_capacity` - separate limit on running disk polls plus
/// results awaiting dequeue by this shard's owner pump. Must be nonzero.
+ /// * `partition_io_limits` - validated file-job limits, or the default limits.
/// * `shards_table` - namespace -> shard routing table.
/// * `coordinator` - `Some` on shard 0 (supplied by the builder when
/// `is_shard_zero`), `None` everywhere else. Immutable post-ctor:
@@ -1777,6 +1778,7 @@ where
inbox: Receiver<ShardFrame>,
reply_inbox: Receiver<ShardFrame>,
poll_completion_capacity: usize,
+ partition_io_limits: Option<PartitionIoLimits>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
coordinator: Option<Rc<crate::coordinator::ShardZeroCoordinator>>,
@@ -1792,10 +1794,10 @@ where
let ShardIdentity { id, name } = identity;
let poll_completions =
poll::completion::PollCompletionLane::new(poll_completion_capacity, &metrics);
- let partition_io = partition_io::PartitionIoLane::new(PartitionIoLimits::new(
- partition_io::DEFAULT_PARTITION_IO_CAPACITY,
- None,
- )?);
+ let partition_io = partition_io::PartitionIoLane::new(match partition_io_limits {
+ Some(limits) => limits,
+ None => PartitionIoLimits::new(partition_io::DEFAULT_PARTITION_IO_CAPACITY, None)?,
+ });
Ok(Self {
id,
name,
@@ -7705,7 +7707,7 @@ where
let _ = partition.persist_superblock_if_needed().await;
}
if let Some(partition) = partitions.get_io_owner(namespace) {
- partition.notify_io();
+ partition.retry_io();
}
}
@@ -7733,6 +7735,7 @@ where
}
}
self.metrics.record_persistence(&persistence_metrics);
+ self.partition_io.record_metrics(&self.metrics);
self.metrics
.record_replica_reads(&self.bus.take_replica_read_stats());
self.metrics
@@ -8879,7 +8882,6 @@ where
/// advancing write would stamp that stale counter over the reset the
/// install just made, then quarantine the segments that would have
/// contradicted it. `None` where the counter is authoritative.
- #[allow(clippy::future_not_send)]
fn fence_partition_for_rebuild(
&self,
namespace: IggyNamespace,
@@ -11350,10 +11352,9 @@ fn build_dvc_suffix(
///
/// Worse to skip here than on the metadata plane, which is why this exists.
/// Partition `append` has no slot-collision check, so a re-prepared op pushes a
-/// duplicate header and rewrites `op_to_storage_offset`, and `committed_prefix` walks
-/// positionally, so the stale entry is what `evict_prefix` flushes to the segment:
+/// duplicate header and rewrites `op_to_storage_offset`. Materialization walks
+/// positionally, so it would flush the stale entry to the segment:
/// durable divergent bytes, no error anywhere.
-#[allow(clippy::future_not_send)]
fn partition_view_divergence_from<B, SB>(
shard: u16,
partition: &IggyPartition<B, SB>,
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index ebbe46a98..12ede8719 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -804,27 +804,27 @@ impl ShardMetrics {
self.register_persistence(registry);
registry.register(
"partition_io_active_jobs",
- "partition I/O active jobs",
+ "partition file jobs whose physical work has not completed",
self.partition_io_active_jobs.clone(),
);
registry.register(
"partition_io_queued_results",
- "partition I/O queued results",
+ "completed partition file jobs awaiting acceptance by their owner",
self.partition_io_queued_results.clone(),
);
registry.register(
"partition_io_charged_bytes",
- "partition I/O charged bytes",
+ "bytes reserved for partition file jobs and retained continuations",
self.partition_io_charged_bytes.clone(),
);
registry.register(
"partition_io_wait_depth",
- "partition I/O wait depth",
+ "partition continuations ready to run or waiting for file-job capacity",
self.partition_io_wait_depth.clone(),
);
registry.register(
"partition_io_quarantined_jobs",
- "partition I/O quarantined jobs",
+ "interrupted partition file jobs retaining their reservation and resource fence",
self.partition_io_quarantined_jobs.clone(),
);
diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs
index 70acdadda..1ebe19c06 100644
--- a/core/shard/src/partition_io.rs
+++ b/core/shard/src/partition_io.rs
@@ -18,10 +18,10 @@
use std::cell::{Cell, RefCell};
use std::collections::{BTreeSet, VecDeque};
use std::rc::Rc;
-use std::task::Waker;
+use std::task::{Poll, Waker};
use consensus::PartitionsHandle;
-use crossfire::{RecvError, TryRecvError};
+use futures::future::poll_fn;
use journal::local_gate::OwnedLocalGateGuard;
use journal::superblock::SuperblockStore;
use message_bus::MessageBus;
@@ -30,13 +30,12 @@ use partitions::{
PartitionIoResult,
};
use server_common::sharding::IggyNamespace;
+pub use server_common::sharding::PARTITION_IO_CAPACITY_MAX;
use thiserror::Error;
-use crate::{IggyShard, Receiver, Sender, channel};
+use crate::IggyShard;
pub const DEFAULT_PARTITION_IO_CAPACITY: usize = 16;
-pub const DEFAULT_PARTITION_IO_BYTES: usize = 256 * 1024 * 1024;
-pub const PARTITION_IO_CAPACITY_MAX: usize = 1 << 20;
#[derive(Clone, Copy, Debug)]
pub struct PartitionIoLimits {
@@ -68,7 +67,7 @@ impl PartitionIoLimits {
let minimum = partitions::largest_legal_job_charge()
.filter(|charge| isize::try_from(*charge).is_ok())
.ok_or(PartitionIoLimitsError::Overflow)?;
- let bytes_max = bytes_max.unwrap_or_else(|| DEFAULT_PARTITION_IO_BYTES.max(minimum));
+ let bytes_max = bytes_max.unwrap_or(minimum);
if bytes_max < minimum || bytes_max > isize::MAX as usize {
return Err(PartitionIoLimitsError::Bytes {
value: bytes_max,
@@ -81,11 +80,6 @@ impl PartitionIoLimits {
})
}
- #[must_use]
- pub const fn capacity(self) -> usize {
- self.capacity
- }
-
#[must_use]
pub const fn bytes_max(self) -> usize {
self.bytes_max
@@ -122,6 +116,10 @@ struct PartitionIoSlot<SB> {
#[derive(Default)]
struct ReadyPartitions {
queue: RefCell<VecDeque<(IggyNamespace, PartitionIncarnation)>>,
+ waiting: RefCell<VecDeque<(IggyNamespace, PartitionIncarnation)>>,
+ // Only the oldest capacity waiter may also be in the runnable queue.
+ waiting_ready: Cell<bool>,
+ completions: RefCell<VecDeque<PartitionIoToken>>,
present: RefCell<BTreeSet<(IggyNamespace, PartitionIncarnation)>>,
waker: RefCell<Option<Waker>>,
}
@@ -151,13 +149,9 @@ pub struct PartitionIoLane<SB> {
pub(crate) limits: PartitionIoLimits,
slots: RefCell<Vec<Option<Rc<PartitionIoSlot<SB>>>>>,
charged: Cell<usize>,
- sender: Sender<PartitionIoToken>,
- receiver: Receiver<PartitionIoToken>,
ready: Rc<ReadyPartitions>,
interrupted: Rc<Cell<bool>>,
- undelivered: Rc<Cell<bool>>,
closed: Cell<bool>,
- capacity_blocked: Cell<bool>,
counters: Rc<IoCounters>,
#[cfg(test)]
execution_gate: RefCell<Option<futures::channel::oneshot::Receiver<()>>>,
@@ -165,18 +159,16 @@ pub struct PartitionIoLane<SB> {
impl<SB: SuperblockStore> PartitionIoLane<SB> {
pub(crate) fn new(limits: PartitionIoLimits) -> Self {
- let (sender, receiver) = channel(limits.capacity);
Self {
limits,
slots: RefCell::new((0..limits.capacity).map(|_| None).collect()),
charged: Cell::new(0),
- sender,
- receiver,
- ready: Rc::default(),
+ ready: Rc::new(ReadyPartitions {
+ completions: RefCell::new(VecDeque::with_capacity(limits.capacity)),
+ ..ReadyPartitions::default()
+ }),
interrupted: Rc::new(Cell::new(false)),
- undelivered: Rc::new(Cell::new(false)),
closed: Cell::new(false),
- capacity_blocked: Cell::new(false),
counters: Rc::default(),
#[cfg(test)]
execution_gate: RefCell::new(None),
@@ -199,30 +191,56 @@ impl<SB: SuperblockStore> PartitionIoLane<SB> {
}
pub(crate) fn has_ready(&self) -> bool {
- self.head().is_some() || self.undelivered.get() || self.interrupted.get()
+ self.head().is_some()
+ || !self.ready.completions.borrow().is_empty()
+ || self.interrupted.get()
}
pub(crate) fn head(&self) -> Option<(IggyNamespace, PartitionIncarnation)> {
- if !self.capacity_blocked.get() {
- return self.ready.queue.borrow().front().copied();
- }
- let present = self.ready.present.borrow();
- self.slots.borrow().iter().flatten().find_map(|slot| {
- (slot.state.get() == SlotState::Settled
- && present.contains(&(slot.namespace, slot.incarnation)))
- .then_some((slot.namespace, slot.incarnation))
- })
+ self.ready.queue.borrow().front().copied()
}
pub(crate) fn pop_ready(&self, namespace: IggyNamespace, incarnation: PartitionIncarnation) {
- self.ready
- .queue
- .borrow_mut()
- .retain(|queued| *queued != (namespace, incarnation));
+ let mut queue = self.ready.queue.borrow_mut();
+ if queue.front() == Some(&(namespace, incarnation)) {
+ queue.pop_front();
+ } else {
+ queue.retain(|queued| *queued != (namespace, incarnation));
+ }
self.ready
.present
.borrow_mut()
.remove(&(namespace, incarnation));
+ drop(queue);
+ let mut waiting = self.ready.waiting.borrow_mut();
+ if waiting.front() == Some(&(namespace, incarnation)) {
+ waiting.pop_front();
+ self.ready.waiting_ready.set(false);
+ drop(waiting);
+ self.wake_capacity_waiter();
+ }
+ }
+
+ fn wait_for_capacity(&self, namespace: IggyNamespace, incarnation: PartitionIncarnation) {
+ let owner = (namespace, incarnation);
+ debug_assert_eq!(self.ready.queue.borrow().front(), Some(&owner));
+ self.ready.queue.borrow_mut().pop_front();
+ let mut waiting = self.ready.waiting.borrow_mut();
+ if waiting.front() == Some(&owner) {
+ self.ready.waiting_ready.set(false);
+ } else {
+ waiting.push_back(owner);
+ }
+ }
+
+ fn wake_capacity_waiter(&self) {
+ if self.ready.waiting_ready.get() {
+ return;
+ }
+ if let Some(owner) = self.ready.waiting.borrow().front().copied() {
+ self.ready.queue.borrow_mut().push_back(owner);
+ self.ready.waiting_ready.set(true);
+ }
}
pub(crate) fn reschedule(&self, namespace: IggyNamespace, incarnation: PartitionIncarnation) {
@@ -247,6 +265,15 @@ impl<SB: SuperblockStore> PartitionIoLane<SB> {
return (existing.state.get() == SlotState::Settled && charge <= existing.charge)
.then_some(index);
}
+ if self
+ .ready
+ .waiting
+ .borrow()
+ .front()
+ .is_some_and(|owner| *owner != (namespace, incarnation))
+ {
+ return None;
+ }
let charged = self.charged.get().checked_add(charge)?;
if charged > self.limits.bytes_max {
return None;
@@ -298,8 +325,6 @@ impl<SB: SuperblockStore> PartitionIoLane<SB> {
quiescence,
counters: Rc::clone(&self.counters),
};
- let sender = self.sender.clone();
- let undelivered = Rc::clone(&self.undelivered);
let ready = Rc::clone(&self.ready);
let counters = Rc::clone(&self.counters);
#[cfg(test)]
@@ -316,48 +341,28 @@ impl<SB: SuperblockStore> PartitionIoLane<SB> {
slot.state.set(SlotState::Queued);
counters.active.set(counters.active.get() - 1);
counters.queued.set(counters.queued.get() + 1);
- if sender
- .try_send(PartitionIoToken {
- slot: index,
- identity,
- })
- .is_err()
- {
- undelivered.set(true);
- if let Some(waker) = ready.waker.borrow().as_ref() {
- waker.wake_by_ref();
- }
+ ready.completions.borrow_mut().push_back(PartitionIoToken {
+ slot: index,
+ identity,
+ });
+ if let Some(waker) = ready.waker.borrow().as_ref() {
+ waker.wake_by_ref();
}
drop(marker);
});
}
#[allow(clippy::future_not_send)]
- pub(crate) async fn recv(&self) -> Result<PartitionIoToken, RecvError> {
- self.receiver.recv().await
+ pub(crate) async fn recv(&self) -> PartitionIoToken {
+ poll_fn(|context| {
+ self.register_waker(context.waker());
+ self.try_recv().map_or(Poll::Pending, Poll::Ready)
+ })
+ .await
}
- pub(crate) fn try_recv(&self) -> Result<PartitionIoToken, TryRecvError> {
- self.receiver.try_recv().or_else(|error| {
- if !self.undelivered.get() {
- return Err(error);
- }
- self.slots
- .borrow()
- .iter()
- .enumerate()
- .find_map(|(index, slot)| {
- let slot = slot.as_ref()?;
- (slot.state.get() == SlotState::Queued).then(|| PartitionIoToken {
- slot: index,
- identity: slot.identity.get().expect("queued slot has an identity"),
- })
- })
- .ok_or_else(|| {
- self.undelivered.set(false);
- error
- })
- })
+ pub(crate) fn try_recv(&self) -> Option<PartitionIoToken> {
+ self.ready.completions.borrow_mut().pop_front()
}
pub(crate) fn take_result(&self, token: PartitionIoToken) -> Option<PartitionIoResult> {
@@ -398,7 +403,7 @@ impl<SB: SuperblockStore> PartitionIoLane<SB> {
}) {
let slot = slots[index].take().expect("settled slot exists");
self.charged.set(self.charged.get() - slot.charge);
- self.capacity_blocked.set(false);
+ self.wake_capacity_waiter();
}
}
@@ -453,7 +458,7 @@ impl<SB: SuperblockStore> PartitionIoLane<SB> {
self.ready.waker.borrow_mut().take();
}
- fn record_metrics(&self, metrics: &crate::metrics::ShardMetrics) {
+ pub(crate) fn record_metrics(&self, metrics: &crate::metrics::ShardMetrics) {
metrics.set_partition_io(
self.counters.active.get(),
self.counters.queued.get(),
@@ -501,15 +506,11 @@ where
/// Bounded completion and continuation service after each ordinary pump event.
#[allow(clippy::future_not_send, clippy::too_many_lines)]
- pub(crate) async fn service_partition_io(&self) -> bool {
- self.partition_io.record_metrics(&self.metrics);
- let progressed = if let Ok(token) = self.partition_io.try_recv() {
+ pub(crate) async fn service_partition_io(&self) {
+ if let Some(token) = self.partition_io.try_recv() {
self.accept_partition_io_completion(token);
self.cooperate().await;
- true
- } else {
- false
- };
+ }
for identity in self.partition_io.interrupted() {
if let Some(partition) = self.plane.partitions().get_io_owner(&identity.namespace) {
partition.interrupt_io(identity);
@@ -517,7 +518,7 @@ where
self.cooperate().await;
}
let Some((namespace, incarnation)) = self.partition_io.head() else {
- return progressed;
+ return;
};
let partitions = self.plane.partitions();
let Some(partition) = partitions
@@ -528,7 +529,7 @@ where
if let Some(token) = self.partition_io.retained(namespace, incarnation) {
self.partition_io.release(token.slot);
}
- return true;
+ return;
};
let step = partition.resume_io(partitions.config()).await;
if let Some(token) = self.partition_io.retained(namespace, incarnation)
@@ -583,12 +584,15 @@ where
self.drop_partition_transfer_state(namespace, partition);
match outcome {
Ok(()) => self.partition_io.reschedule(namespace, incarnation),
- Err(partitions::PurgeError::Unserviceable(error)) => {
+ Err(
+ partitions::PurgeError::Unserviceable(error)
+ | partitions::PurgeError::OffsetsNotDurable(error),
+ ) => {
tracing::error!(namespace_raw = namespace.inner(), generation, %error, "partition purge failed after mutation");
self.fence_partition_for_rebuild(namespace, partition, Some(0));
}
Err(error) => {
- tracing::warn!(namespace_raw = namespace.inner(), generation, %error, "partition purge remains unapplied; reconciler will retry");
+ tracing::warn!(namespace_raw = namespace.inner(), generation, %error, "partition purge remains incomplete; reconciler will retry");
}
}
}
@@ -617,8 +621,8 @@ where
self.partition_io
.try_reserve(namespace, incarnation, plan.allocation_charge)
else {
- self.partition_io.capacity_blocked.set(true);
- return progressed;
+ self.partition_io.wait_for_capacity(namespace, incarnation);
+ return;
};
self.partition_io.pop_ready(namespace, incarnation);
match partition.capture_io(plan, partitions.config()) {
@@ -646,7 +650,6 @@ where
}
}
self.cooperate().await;
- true
}
}
@@ -723,6 +726,236 @@ mod tests {
const REPLY_DEADLINE: Duration = Duration::from_secs(1);
const TASK_POLL_INTERVAL: Duration = Duration::from_millis(1);
+ #[compio::test]
+ async fn failed_purge_offset_sync_fences_before_readmission() {
+ const STEPS_MAX: usize = 64;
+ let directory = tempfile::tempdir().unwrap();
+ let bus = Rc::new(IggyMessageBus::new(0));
+ let (owner, _sender) = test_owner(&bus, None);
+ let namespace = IggyNamespace::new(0, 0, 0);
+ let partitions = owner.plane.partitions();
+ partitions.set_io_notifier(
+ owner.partition_io.notifier(),
+ owner.partition_io.limits.bytes_max(),
+ );
+ let partition = partitions.get_mut_by_ns(&namespace).unwrap();
+ partition.set_partition_dir(directory.path().to_str().unwrap().to_owned());
+ let directories = ["consumers", "groups", "external"]
+ .map(|name| directory.path().join(name).to_str().unwrap().to_owned());
+ for path in &directories {
+ std::fs::create_dir(path).unwrap();
+ }
+ let failed_directory = directories[0].clone();
+ partition.configure_consumer_offset_storage(
+ directories,
+ ConsumerOffsets::with_capacity(1),
+ ConsumerGroupOffsets::with_capacity(1),
+ );
+ assert!(matches!(
+ partition.purge(partitions.config(), 1).await,
+ Err(partitions::PurgeError::Pending)
+ ));
+ let mut failed = false;
+ for _ in 0..STEPS_MAX {
+ if let partitions::PartitionIoStep::Ready(plan) =
+ partition.resume_io(partitions.config()).await
+ {
+ let captured = partition
+ .capture_io(plan, partitions.config())
+ .unwrap()
+ .unwrap();
+ let failure = matches!(&captured.job, partitions::PartitionIoJob::SegmentDirectory(path) if *path == failed_directory);
+ let result = if failure {
+ partitions::PartitionIoResult::SegmentDirectory(Err(io::Error::other(
+ "offset directory writeback failed",
+ )))
+ } else {
+ captured.job.execute().await
+ };
+ partition.accept_io(captured.identity, result).unwrap();
+ if let Some(gate) = captured.gate {
+ gate.release();
+ }
+ captured.quiescence.settle(captured.identity);
+ if failure {
+ failed = true;
+ break;
+ }
+ }
+ }
+ assert!(failed, "purge reached the offset directory barrier");
+ owner.service_partition_io().await;
+ assert!(
+ partitions.get_by_ns(&namespace).is_none(),
+ "failed purge must stop admission before the next frame"
+ );
+ assert!(owner.shards_table.shard_for(namespace).is_none());
+ assert_eq!(
+ partitions
+ .get_io_owner(&namespace)
+ .unwrap()
+ .applied_purge_generation(),
+ 0
+ );
+ }
+
+ #[test]
+ fn idle_partitions_do_not_enter_the_io_ready_queue_on_retry() {
+ let bus = Rc::new(IggyMessageBus::new(0));
+ let (owner, _) = test_owner(&bus, None);
+ let partitions = owner.plane.partitions();
+ partitions.set_io_notifier(
+ owner.partition_io.notifier(),
+ owner.partition_io.limits.bytes_max(),
+ );
+ while let Some((namespace, incarnation)) = owner.partition_io.head() {
+ owner.partition_io.pop_ready(namespace, incarnation);
+ }
+ for namespace in partitions.namespaces() {
+ partitions.get_io_owner(namespace).unwrap().retry_io();
+ }
+ assert!(!owner.partition_io.has_ready());
+ }
+
+ #[compio::test]
+ async fn completed_commit_publishes_while_one_slot_is_held_and_another_job_waits() {
+ let directory = tempfile::tempdir().unwrap();
+ let store = Rc::new(HeldSuperblock {
+ entered: RefCell::new(None),
+ held: RefCell::new(None),
+ });
+ let bus = Rc::new(IggyMessageBus::new(0));
+ let (mut owner, _sender) = test_owner(&bus, Some(&store));
+ owner.partition_io =
+ super::PartitionIoLane::new(super::PartitionIoLimits::new(1, None).unwrap());
+ let lane = &owner.partition_io;
+ let partitions = owner.plane.partitions();
+ partitions.set_io_notifier(lane.notifier(), lane.limits.bytes_max());
+ let completed = IggyNamespace::new(0, 0, 1);
+ let partition = partitions.get_mut_by_ns(&completed).unwrap();
+ let dirs = ["consumers", "groups", "external_groups"]
+ .map(|name| directory.path().join(name).to_str().unwrap().to_owned());
+ for dir in &dirs {
+ std::fs::create_dir(dir).unwrap();
+ }
+ partition.configure_consumer_offset_storage(
+ dirs,
+ ConsumerOffsets::with_capacity(1),
+ ConsumerGroupOffsets::with_capacity(1),
+ );
+ partition.stats.increment_messages_count(1);
+ partition.set_runtime_options(TopicRuntimeOptions {
+ consumer_offset_durability: Durability::Persisted,
+ ..Default::default()
+ });
+ let completed_incarnation = partition.incarnation();
+ let (reply, result) = consensus::oneshot_channel();
+ partitions
+ .on_request_with_reply(store_request(completed, AckLevel::Quorum), Some(reply))
+ .await;
+ let partition = partitions.get_mut_by_ns(&completed).unwrap();
+ let mut acknowledgments = Vec::new();
+ partition
+ .consensus()
+ .drain_loopback_into(&mut acknowledgments);
+ for ack in acknowledgments {
+ partition
+ .on_ack(ack.try_into_typed().unwrap(), partitions.config())
+ .await;
+ }
+ let captured_bus = crate::poll::timeout_tests::PollTestBus::default();
+ let mut files_complete = false;
+ for _ in 0..crate::router::COOPERATIVE_EVENT_BUDGET {
+ let partition = partitions.get_mut_by_ns(&completed).unwrap();
+ match partition.resume_io(partitions.config()).await {
+ partitions::PartitionIoStep::Ready(plan) => {
+ let slot = lane
+ .try_reserve(completed, completed_incarnation, plan.allocation_charge)
+ .unwrap();
+ let captured = partition
+ .capture_io(plan, partitions.config())
+ .unwrap()
+ .unwrap();
+ let last = matches!(
+ &captured.job,
+ partitions::PartitionIoJob::OffsetDirectories(_)
+ );
+ lane.dispatch(slot, captured, &captured_bus);
+ let task = captured_bus.spawned_tasks.borrow_mut().pop().unwrap();
+ task.await;
+ owner.accept_partition_io_completion(lane.try_recv().unwrap());
+ if last {
+ files_complete = true;
+ break;
+ }
+ }
+ partitions::PartitionIoStep::Progress => {}
+ _ => panic!("offset commit must reach its directory sync"),
+ }
+ }
+ assert!(files_complete);
+ assert_eq!(
+ partitions
+ .get_by_ns(&completed)
+ .unwrap()
+ .consensus()
+ .commit_min(),
+ 0
+ );
+ assert_eq!(lane.charged.get(), 0);
+ let mut result = std::pin::pin!(result);
+ assert!(futures::poll!(&mut result).is_pending());
+
+ let (slot, captured) = capture_first_write(&owner).await;
+ lane.dispatch(slot, captured, &captured_bus);
+ let waiting = IggyNamespace::new(0, 0, 2);
+ let partition = partitions.get_mut_by_ns(&waiting).unwrap();
+ partition.set_superblock(Rc::clone(&store), None);
+ assert!(!partition.reserve_offsets_through(0).await);
+ let waiting_incarnation = partition.incarnation();
+ while let Some((namespace, incarnation)) = lane.head() {
+ lane.pop_ready(namespace, incarnation);
+ }
+ lane.reschedule(waiting, waiting_incarnation);
+ lane.reschedule(completed, completed_incarnation);
+ owner.service_partition_io().await;
+ assert_eq!(
+ lane.ready.waiting.borrow().front(),
+ Some(&(waiting, waiting_incarnation))
+ );
+ for _ in 0..crate::router::COOPERATIVE_EVENT_BUDGET {
+ owner.service_partition_io().await;
+ if let std::task::Poll::Ready(reply) = futures::poll!(&mut result) {
+ assert_eq!(reply.unwrap().header().status, 0);
+ assert_eq!(
+ partitions
+ .get_by_ns(&completed)
+ .unwrap()
+ .consensus()
+ .commit_min(),
+ 1
+ );
+ assert_eq!(
+ lane.counters.active.get(),
+ 1,
+ "the unrelated file job must still be held"
+ );
+ let task = captured_bus.spawned_tasks.borrow_mut().pop().unwrap();
+ task.await;
+ owner.accept_partition_io_completion(lane.try_recv().unwrap());
+ assert!(
+ lane.ready
+ .queue
+ .borrow()
+ .contains(&(waiting, waiting_incarnation)),
+ "the freed slot must wake the waiting file job"
+ );
+ return;
+ }
+ }
+ panic!("a completed commit was blocked by unrelated file-job capacity");
+ }
+
#[compio::test]
async fn given_held_superblock_when_other_partition_receives_send_should_ack_before_release() {
let (entered, started) = oneshot::channel();
@@ -859,7 +1092,7 @@ mod tests {
ConsumerGroupOffsets::with_capacity(1),
);
partition.stats.increment_messages_count(1);
- store_request(namespace)
+ store_request(namespace, AckLevel::NoAck)
} else {
send_request(namespace, 1)
};
@@ -1610,17 +1843,17 @@ mod tests {
}
#[compio::test]
- async fn disconnected_completion_channel_preserves_local_result_for_acceptance() {
+ async fn closing_admission_preserves_local_result_for_acceptance() {
let store = Rc::new(HeldSuperblock {
entered: RefCell::new(None),
held: RefCell::new(None),
});
let bus = Rc::new(IggyMessageBus::new(0));
- let (mut owner, _sender) = test_owner(&bus, Some(&store));
+ let (owner, _sender) = test_owner(&bus, Some(&store));
let captured_bus = crate::poll::timeout_tests::PollTestBus::default();
let (slot, captured) = capture_first_write(&owner).await;
- owner.partition_io.receiver = channel(1).1;
owner.partition_io.dispatch(slot, captured, &captured_bus);
+ owner.partition_io.close();
let task = captured_bus.spawned_tasks.borrow_mut().pop().unwrap();
task.await;
assert!(owner.partition_io.has_ready());
@@ -1655,7 +1888,6 @@ mod tests {
.set(super::SlotState::Settled);
lane.reschedule(namespace, old);
lane.reschedule(namespace, replacement);
- lane.capacity_blocked.set(true);
assert_eq!(
lane.head(),
Some((namespace, old)),
@@ -1700,13 +1932,25 @@ mod tests {
lane.try_reserve(identities[1].0, identities[1].1, limits.bytes_max())
.is_none()
);
- lane.capacity_blocked.set(true);
+ lane.wait_for_capacity(identities[1].0, identities[1].1);
+ assert_eq!(
+ lane.head(),
+ Some(identities[2]),
+ "CPU continuations remain runnable"
+ );
assert!(
- lane.head().is_none(),
- "new reservations wait for the oldest eligible job"
+ lane.try_reserve(identities[2].0, identities[2].1, half)
+ .is_none()
);
+ lane.wait_for_capacity(identities[2].0, identities[2].1);
+ assert!(lane.head().is_none(), "file jobs wait without spinning");
lane.release(first);
assert_eq!(lane.head(), Some(identities[1]));
+ assert_eq!(
+ lane.ready.queue.borrow().len(),
+ 1,
+ "one capacity release retries only the oldest file job"
+ );
let large = lane
.try_reserve(identities[1].0, identities[1].1, limits.bytes_max())
.unwrap();
@@ -1850,6 +2094,7 @@ mod tests {
inbox,
replies,
2,
+ None,
routes,
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
None,
@@ -1901,7 +2146,7 @@ mod tests {
.unwrap()
}
- fn store_request(namespace: IggyNamespace) -> Message<RoutedRequestHeader> {
+ fn store_request(namespace: IggyNamespace, ack: AckLevel) -> Message<RoutedRequestHeader> {
let body = StoreConsumerOffsetRequest {
consumer: WireConsumer {
kind: ConsumerKind::Consumer.as_code(),
@@ -1911,7 +2156,7 @@ mod tests {
topic_id: WireIdentifier::Numeric(namespace.topic_id().try_into().unwrap()),
partition_id: Some(namespace.partition_id().try_into().unwrap()),
offset: 0,
- ack: AckLevel::NoAck,
+ ack,
}
.to_bytes();
let header_size = size_of::<RoutedRequestHeader>();
diff --git a/core/shard/src/poll/completion_tests.rs b/core/shard/src/poll/completion_tests.rs
index 42a6d0c2f..e70f925f9 100644
--- a/core/shard/src/poll/completion_tests.rs
+++ b/core/shard/src/poll/completion_tests.rs
@@ -1089,6 +1089,7 @@ fn owner_with_metadata_plane(
inbox,
replies,
2,
+ None,
routes,
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 3), bus.clone()),
None,
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 94c3dcb24..8e1b58d39 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -481,19 +481,14 @@ where
}
}
completion = self.partition_io.recv().fuse() => {
- match completion {
- Ok(token) => {
- self.accept_partition_io_completion(token);
- self.cooperate().await;
- }
- Err(_) => break,
- }
+ self.accept_partition_io_completion(completion);
+ self.cooperate().await;
}
completion = self.poll_completions.recv().fuse() => {
match completion {
Ok(completion) => {
self.on_poll_completed(*completion).await;
- self.cooperate().await;
+ self.cooperate().await;
self.process_loopback(&mut loopback_buf).await;
self.apply_reconcile_ops();
}
@@ -586,8 +581,8 @@ where
let rearm = || self.bus.sleep(CONSENSUS_TICK_INTERVAL).fuse();
let mut tick = std::pin::pin!(rearm());
let mut namespace_scratch = Vec::with_capacity(namespaces.len());
- let mut inbox_closed = false;
- let mut replies_closed = false;
+ let mut inbox = std::pin::pin!(self.inbox.recv().fuse());
+ let mut replies = std::pin::pin!(self.reply_inbox.recv().fuse());
loop {
self.process_one_poll_completion(loopback).await;
self.service_partition_io().await;
@@ -605,7 +600,7 @@ where
for namespace in &namespaces {
if let Some(partition) = partitions.get_io_owner(namespace) {
partition.drive_persistence().await;
- partition.notify_io();
+ partition.retry_io();
}
}
self.tick_metadata().await;
@@ -613,23 +608,23 @@ where
tick.set(rearm());
}
completion = self.partition_io.recv().fuse() => {
- if let Ok(token) = completion {
- self.accept_partition_io_completion(token);
- self.cooperate().await;
- }
+ self.accept_partition_io_completion(completion);
+ self.cooperate().await;
}
- frame = (if inbox_closed { futures::future::Either::Left(futures::future::pending()) } else { futures::future::Either::Right(self.inbox.recv()) }).fuse() => {
- match frame {
- Ok(frame) if self.accept_frame_for_self(&frame) => self.process_frame(frame).await,
- Ok(_) => {},
- Err(_) => inbox_closed = true,
+ frame = inbox.as_mut() => {
+ if let Ok(frame) = frame {
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ }
+ inbox.set(self.inbox.recv().fuse());
}
}
- frame = (if replies_closed { futures::future::Either::Left(futures::future::pending()) } else { futures::future::Either::Right(self.reply_inbox.recv()) }).fuse() => {
- match frame {
- Ok(frame) if self.accept_frame_for_self(&frame) => self.process_frame(frame).await,
- Ok(_) => {},
- Err(_) => replies_closed = true,
+ frame = replies.as_mut() => {
+ if let Ok(frame) = frame {
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ }
+ replies.set(self.reply_inbox.recv().fuse());
}
}
() = poll_fn(|context| {
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index d403e1dd1..0923d7aa2 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -436,6 +436,7 @@ pub fn new_shard(
inbox,
reply_inbox,
ServerConfig::default().sharding.poll_completion_capacity,
+ None,
PapayaShardsTable::new(),
shard::PartitionConsensusConfig::with_clock(
CLUSTER_ID,