hubcio commented on code in PR #3828:
URL: https://github.com/apache/iggy/pull/3828#discussion_r3735345600
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3153,6 +3227,33 @@ where
self.stats.zero_out_all();
self.stats.increment_segments_count(1);
+ // Fence the resident journal instead of clearing it: entries are
+ // consensus history (backup commit walks, repair, retransmission), so
+ // they stay, but every journal-apply path no-ops ops at or below this
+ // floor (see `purge_floor_op`). The write lock held here is the same
+ // one appends take, and the pump is single-threaded, so no op can be
+ // assigned between reading the sequence and installing the floor.
+ self.purge_floor_op = self.consensus.sequencer().current_sequence();
Review Comment:
agreed - the floor is read off the local sequencer at local apply time, so
replicas fence different in-flight sets and an ack in the apply window can die.
a real fix needs the purge ordered against the partition plane (checkpoint
barrier or a replicated purge op), which is a protocol change, not something to
smuggle into this PR. the TODO at the reconciler now names both skew modes;
tracking it as a follow-up.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -3394,30 +3381,10 @@ where
&body,
))
}
- Operation::UpdateTopic => {
- let mut request = WireUpdateTopicRequest::decode_from(body)
- .map_err(|_| IggyError::InvalidCommand)?;
- // Same `ServerDefault` resolution as `CreateTopic` above;
rebuild
- // the prepare only if a sentinel actually needs stamping, else
- // project the untouched buffer zero-copy.
- let needs_rewrite = request.max_topic_size == 0 ||
request.message_expiry == 0;
- if request.max_topic_size == 0 {
- request.max_topic_size = self.default_max_topic_size.get();
- }
- if request.message_expiry == 0 {
- request.message_expiry = self.default_message_expiry.get();
- }
- if needs_rewrite {
- let body = request.to_bytes();
- return Ok(build_prepare_message(
- consensus,
- &header,
- Operation::UpdateTopic,
- &body,
- ));
- }
- Ok(message.project(consensus))
- }
+ // `UpdateTopic` deliberately takes the default arm: unlike create,
Review Comment:
fixed. the cleaner now resolves `ServerDefault` against the node default at
enforcement time (new `default_max_topic_size()` accessor + an extracted
`per_partition_size_budget` so it is unit-testable). explicit sizes ignore the
default; a node default of unlimited still means no cap, so it cannot turn into
a trim-everything budget.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3153,6 +3227,33 @@ where
self.stats.zero_out_all();
self.stats.increment_segments_count(1);
+ // Fence the resident journal instead of clearing it: entries are
+ // consensus history (backup commit walks, repair, retransmission), so
+ // they stay, but every journal-apply path no-ops ops at or below this
+ // floor (see `purge_floor_op`). The write lock held here is the same
+ // one appends take, and the pump is single-threaded, so no op can be
+ // assigned between reading the sequence and installing the floor.
+ self.purge_floor_op = self.consensus.sequencer().current_sequence();
+ // The journal's flush accounting and resident poll indexes describe
+ // pre-purge bytes; reset them so the flush threshold counts only
+ // post-purge appends and polls fall back to the (fresh, empty)
+ // segments instead of resolving purged resident entries.
+ self.log.journal_mut().info = JournalInfo::default();
+ self.log
+ .journal()
+ .inner
+ .clear_poll_index(self.purge_floor_op);
+
+ // Last durable step: record the applied generation before the
+ // in-memory marker advances. On a write failure the marker stays old,
+ // the error propagates, and the reconciler retries the whole purge
+ // (idempotent). The reverse order would ack a purge that a crash
+ // then silently undoes: restart would hydrate the old generation,
+ // yet the reconciler believes the purge applied.
+ if let Some(dir) = self.partition_dir() {
+ let path = format!("{dir}/{PURGE_GENERATION_FILE}");
+ persist_purge_generation(&path, generation).await?;
Review Comment:
both halves are gone after the rebase onto the state-transfer machinery. the
frontier reset is recorded before anything is mutated, and a failed generation
record now sets `purge_deferred`, which withholds `PrepareOk` for the group
until the re-issued purge lands - so nothing gets acked at offset 0 inside the
retry window (single node included, the commit rides the loopback ack).
ordering: the partition dir is fsynced after the unlinks and the empty plant,
before the generation write, so `purge.gen` can never be durable while the
deletions are not.
##########
core/shard/src/lib.rs:
##########
@@ -3311,8 +3312,47 @@ where
}
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
+ // Purge convergence gate: while a committed purge is not yet locally
+ // applied, this journal still holds pre-purge entries with NO floor
+ // to fence them (the floor is installed by the purge itself), so
+ // serving now would hand a rejoiner batches the cluster purged.
+ // Defer instead: no RepairDone is sent, the rejoiner's stall retry
+ // re-asks, and the local purge (one reconciler wake away) installs
+ // the floor the fence below serves behind.
+ let namespace = IggyNamespace::from_raw(header.namespace);
+ let committed_purge = self
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .partition_purge_generation(
+ namespace.stream_id(),
+ namespace.topic_id(),
+ namespace.partition_id(),
+ );
+ if committed_purge > partition.applied_purge_generation() {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ committed_purge,
+ applied_purge = partition.applied_purge_generation(),
+ "deferring repair serve until the committed purge applies
locally"
+ );
+ return;
+ }
let to_op = header.to_op.min(partition.consensus().commit_max());
- let retained_from =
partition.log.journal().inner.repair_retained_from();
+ // Purge fence: never serve entries at or below this replica's purge
+ // floor. The journal keeps them (own commit walk), but a rejoiner's
+ // floor died with its process, so served pre-purge batches would
+ // flush right back into its freshly reset segments. Reporting the
+ // floor as the retention start rides the normal `RangeEvicted` path:
+ // the rejoiner moves its commit floor to the purge point instead.
+ let purge_floor = partition.purge_floor_op();
+ let retained_from = match
partition.log.journal().inner.repair_retained_from() {
+ Some(op) => Some(op.max(purge_floor + 1)),
Review Comment:
resolved by the rebase. an empty repaired window (floor >= to_op) now counts
as complete, refusal is definitive (session dropped, `FloorRefused`), and the
shard converts it to a partition state transfer instead of retrying forever.
the serve side also defers while a committed purge is not yet locally applied,
so the window you describe cannot even be served pre-purge.
##########
core/server-ng/src/responses.rs:
##########
@@ -1193,13 +1176,18 @@ fn partition_response(
// across all shards and both left-right buffers), populated when the
// owning shard materializes the partition; `None` only in the window
// before that first materialization.
+ //
+ // A committed partition always materializes with exactly one empty
+ // segment, so before the owning shard gets there (registry miss, or
+ // registered but not yet segmented) the reply reports that deterministic
+ // initial state instead of a zero a client would read as "no storage".
let stats = streams
.stats_registry
.partition_get(stream_id, topic_id, partition.id);
let (segments_count, current_offset, size_bytes, messages_count) =
- stats.map_or((0, 0, 0, 0), |stats| {
+ stats.map_or((1, 0, 0, 0), |stats| {
(
- stats.segments_count_inconsistent(),
+ stats.segments_count_inconsistent().max(1),
Review Comment:
looked hard at this one and the premise does not hold: the registry entry is
minted get-or-create by the reconciler before `build_partition_fresh` runs, so
"registered with zero segments" is the ordinary mid-build window, not the
fenced state. i tried dropping the clamp - it deterministically breaks the
get-topic-right-after-create contract (one empty segment, no convergence loop,
which the system scenario pins). a fenced partition also reading one empty
segment is the accepted cost and the comment now says so explicitly. separating
the two needs a materialization signal the stats registry does not carry today,
so that is a design change rather than a tweak here.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -724,6 +730,80 @@ fn pop_next_client_request(
message
}
+/// Per-request partitions-count cap shared by create-topic, create-partitions
+/// and delete-partitions admission. Runs pre-consensus like
+/// [`validate_topic_bounds`]: an oversized count must not burn a replicated
+/// log entry (create-partitions admission would also allocate that many
+/// consensus-group ids before replicating).
+pub(crate) const fn validate_partitions_count(partitions_count: u32) ->
Result<(), IggyError> {
+ if partitions_count > MAX_PARTITIONS_PER_REQUEST {
Review Comment:
checked legacy first and it splits: create_topic admits zero (`0..=MAX` in
the shared validator), while create/delete partitions reject it - so the zero
gate went into a separate `validate_partitions_change_count` used only by the
two change paths, keeping create_topic parity. also took the two typed-error
gaps while in there: tcp `get_consumer_offset` now denies typed on
`PartitionNotFound` (other resolver errors keep the empty body so "no offset
yet" still reads as none), and the http poll passes `PartitionNotFound` through
instead of a generic 404.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -1077,6 +1094,7 @@ fn reconcile_partition_purges(ctx: &ReconcilerCtx) {
.map_or(0, partitions::IggyPartition::applied_purge_generation);
if committed > applied {
ctx.shard.request_purge_partition(namespace, committed);
+ counters.purges_pending += 1;
Review Comment:
took the observability half: `purges_staged` and `trims_pending` are in the
pass-complete debug line now, and both defer sites (serve + receive) bump a new
`partition_repair_serves_deferred_purge_total` counter, plus a per-attempt warn
on each deferred purge. left the backoff out deliberately - staging is a
`try_send` of one frame and the pump re-checks the generation, so a restage per
pass is the pacing, and the frontier-record failure path already respects the
superblock writer's own backoff and mutates nothing when it refuses.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1984,6 +2025,13 @@ where
}
continue;
}
+ // Purge floor: a pre-purge batch committing after the
+ // purge must not flush its (purged) bytes into the fresh
+ // segment. It still counts into `chunk_len`, so it joins
+ // the evictable prefix and commit_min advances normally.
+ if peek_op(&entry) <= self.purge_floor_op {
Review Comment:
fixed. the receive side now mirrors the serve gate: `on_repair_range_reply`
defers the whole reply (floor install included) while the committed purge
generation is ahead of the local applied one, so the stale durable line is
cleared by the local purge before any floor can be accepted. deferring both
arms beats gating just completion - a deferred `RangeEvicted` is re-derived by
the stall retry.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -3465,6 +3475,9 @@ const fn operation_triggers_partition_reconcile(op:
Operation) -> bool {
| Operation::DeleteTopic
| Operation::DeleteStream
| Operation::DeletePartitions
+ | Operation::PurgeStream
+ | Operation::PurgeTopic
+ | Operation::TruncatePartition
Review Comment:
fixed. both purge applies (`PurgeStream` and `PurgeTopic`) now zero
`deleted_up_to_offset` - replicated state, so every replica converges on the
same reset and the reconciler has nothing stale to re-stage.
`PartitionSnapshot` already carried the field, so the snapshot shape is
unchanged.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2302,6 +2356,14 @@ where
if entry.header.operation != Operation::SendMessages {
return None;
}
+ // Purge floor: a pre-purge send committing after the purge
+ // reports no visible offsets ("send without confirmation", the
+ // established degradation), which also keeps
+ // `commit_partition_entry` from re-advancing the reset offset
+ // and stats with pre-purge values.
+ if entry.header.op <= self.purge_floor_op {
Review Comment:
kept the empty-confirmations shape, and documented why at the site: the op
did commit cluster-wide (the fence only stops the local flush of purged bytes),
so a typed transient status would invite a client retry that duplicates the
send into the post-purge offset space. the reply is byte-identical to a send
without confirmation, which is the established degradation for
offsets-unavailable.
##########
core/partitions/src/journal.rs:
##########
@@ -662,6 +679,25 @@ where
offset_to_op.keys().next().copied()
}
+ /// Seal the resident poll tier: clear the offset and timestamp poll
+ /// indexes ONLY, so `oldest_resident_offset` reads `None` and every poll
+ /// falls back to the on-disk segments. Called by a partition purge, which
+ /// wipes the segments but must KEEP the journal entries themselves:
+ /// headers, storage, `op_to_storage_offset` and the evicted ring are
+ /// consensus history that backups, repair and retransmission still walk.
+ /// Clearing those would wedge `commit_min` until a view change.
+ ///
+ /// `floor` (the purge's fence op) makes the seal survive eviction:
+ /// `evict_prefix` re-appends the retained tail, and without the floor
+ /// that re-append would re-index the pre-purge entries just cleared.
+ pub fn clear_poll_index(&self, floor: u64) {
Review Comment:
fixed on both halves. `resident_entries()` (the snapshot accessor the poll
walk splices on; the index path is fenced separately by the same floor at
insert) filters ops at or below the floor, so a post-purge re-index cannot
re-expose fenced entries. and the purge now hands the fenced prefix to
`evict_committed_prefix` bounded at `commit_min` - not `commit_max`, evicting
an op the commit walk has not reached would delete its header and wedge the
walk on the hole. the ring still serves repair, and the serve path clamps
`retained_from` above the floor anyway.
##########
core/server-ng/src/partition_helpers.rs:
##########
@@ -536,6 +536,11 @@ pub async fn build_partition_fresh(
topic_id,
partition_id,
));
+ // Fresh dirs read generation 0; a dir surviving from a crashed process
+ // (this "fresh" build races repair re-materialization) reads the last
+ // durably-applied purge so the reconciler does not re-wipe messages
+ // appended after it.
+ partition.hydrate_applied_purge_generation().await?;
Review Comment:
the incarnation hole is fixed: `purge.gen` now records
`[generation][created_revision]` (the STM row's epoch), and hydration treats a
mismatched incarnation as 0, so a file surviving a failed teardown cannot
swallow the next topic's purges. left the `committed > applied` gate alone -
`!=` would restage forever whenever a stale applied ran ahead of committed,
which pins the reconciler scan. on the first problem: a fresh build carries no
segments and the resident journal dies with the process, so there is nothing
pre-purge for the hydrated generation to mis-fence - the debug_assert on the
build path pins that.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -143,6 +146,15 @@ where
/// generation against this and resets only when it advances, so a
redundant
/// reconcile pass never re-wipes a partition already at this generation.
applied_purge_generation: u64,
+ /// Highest consensus op assigned when the last purge ran. INVARIANT: every
Review Comment:
fixed - the guard is in. the entry is still journaled (dropping it would
leave a hole and wedge `commit_min` on the walk), but a repaired send at or
below the floor no longer touches `dirty_offset`, segment position or
`journal.info`, and returns no offset so it cannot anchor `first_batch_offset`
in the floor-connect check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
