You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/09/03 09:43:18 UTC
[PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
numinnex opened a new pull request, #4046:
URL: https://github.com/apache/iggy/pull/4046
Update the `Stats` from metadata layer when partitions get removed.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4046:
URL: https://github.com/apache/iggy/pull/4046#issuecomment-5551030323
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4046:
URL: https://github.com/apache/iggy/pull/4046#issuecomment-5523788254
please add integration test (either enhance current one if it exists or create new one)
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4046:
URL: https://github.com/apache/iggy/pull/4046#issuecomment-5524250505
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4046?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 84.74%. Comparing base ([`bad22c0`](https://app.codecov.io/gh/apache/iggy/commit/bad22c03c64ae0b1b843a0c3f8375f26c6d38228?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`cd885e2`](https://app.codecov.io/gh/apache/iggy/commit/cd885e22051b61a43008d8dc6ba888f7defadc45?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4046 +/- ##
============================================
- Coverage 85.21% 84.74% -0.48%
Complexity 1402 1402
============================================
Files 1236 1236
Lines 181543 181533 -10
Branches 147847 147837 -10
============================================
- Hits 154699 153834 -865
- Misses 22823 23720 +897
+ Partials 4021 3979 -42
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.10% <ø> (-0.01%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `67.29% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.33% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.06% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.65% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.24% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4046/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.05% <ø> (-13.27%)` | :arrow_down: |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4046?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/metadata/src/stm/stream.rs](https://app.codecov.io/gh/apache/iggy/pull/4046?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fstm%2Fstream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvc3RtL3N0cmVhbS5ycw==) | `91.50% <ø> (-0.04%)` | :arrow_down: |
... and [40 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4046/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4046:
URL: https://github.com/apache/iggy/pull/4046#discussion_r3938153922
##########
core/server/src/partition_reconciler.rs:
##########
@@ -928,10 +926,21 @@ async fn tear_down_owned_partition(
// on_replicate / on_ack frames that haven't observed the queued
// tombstone yet. Idempotent on retry: already-tombstoned namespace
// stays tombstoned; already-removed shards_table row is a no-op.
+ // Take the counters handle BEFORE the tombstone (every partition accessor
+ // is tombstone-gated) and settle it AFTER, once the tombstone has stopped
+ // new frames from resolving the partition. The metadata apply already
+ // rolled it out at commit time, so what is left here is whatever landed in
+ // the window between the two: in-flight appends adding to parents that
+ // outlive the partition, and a retention tick subtracting from counters the
+ // apply had already zeroed. A frame already past the gate can still land
+ // after this settle, which is why the rollback clamps rather than trusting
+ // the amount it is handed.
+ let live_stats = partitions.with_partition(&ns, |partition| Arc::clone(&partition.stats));
if !partitions.is_tombstoned(&ns) {
partitions.tombstone(ns);
}
shards_table.remove(&ns);
+ settle_partition_stats(ctx, ns, live_stats.as_ref());
Review Comment:
warning: a pump handler suspended mid-append resumes after this settle, and its increment lands in the parent totals with nothing left to roll it back. settle off the removed partition in the `ConfirmRemove` arm instead.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1137,18 +1146,68 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
.read(|inner| inner.revision)
}
-/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
-/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_partition_stats(
+/// Roll a torn-down partition's counters out of its parent topic and stream,
+/// and drop its registry entry.
+///
+/// Two handles, because they can differ. The registry entry is usually gone
+/// already (the metadata apply evicts it on the commit that acked the delete),
+/// but a stale incarnation being torn down after a slab-key reuse still has
+/// one, and it must not survive into the rebuild: `partition` is a
+/// get-or-create, so the rebuild would inherit the dead incarnation's counters
+/// and its `purged_generation`. `live` is the mounted partition's own handle,
+/// which is the only place the post-commit residue lives. When both name the
+/// same `Arc` the second zeroing rolls back 0.
+fn settle_partition_stats(
+ ctx: &ReconcilerCtx,
+ ns: IggyNamespace,
+ live: Option<&Arc<iggy_common::PartitionStats>>,
+) {
+ // Registry cloned out rather than mutated inside the read closure: the
+ // rollback cascades into parent totals, which the STM read has no part in.
+ let registry = ctx
+ .shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|inner| Arc::clone(&inner.stats_registry));
+ registry.remove_partition(ns.stream_id(), ns.topic_id(), ns.partition_id());
Review Comment:
warning: evicting the whole entry drops `purged_generation`, so the rebuild mints 0 and a still-pending purge then wipes everything appended since. seed the gate from the committed `Partition::purge_generation` - keeping the entry would break the recreate test.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -2360,6 +2553,10 @@ impl StreamsInner {
// that slot next.
registry.retain_from_snapshot(&snapshot);
*self = Self::inner_from_snapshot(snapshot, registry);
+ // Last, over the totals `inner_from_snapshot` just stored: the donor's
+ // aggregates describe the donor's partitions, and this node kept its
+ // own the whole time.
+ self.stats_registry.rebuild_parent_totals(&self.items);
Review Comment:
warning: `retain_from_snapshot` keeps entries by slab key with no identity check, so behind a donor that recycled a key the receiver keeps the dead occupant's stats and this rebuild makes them authoritative. compare `created_revision` per key first.
##########
core/server/src/http/metrics.rs:
##########
@@ -45,6 +45,16 @@ pub(in crate::http) struct HttpMetrics {
pub(in crate::http) messages: Gauge,
pub(in crate::http) users: Gauge,
pub(in crate::http) clients: Gauge,
+ /// Not a legacy-parity metric: the running count of rollup decrements that
+ /// had to be clamped at zero. Non-zero means a topic or stream total was
+ /// asked to give back more than it held, so the aggregate it now reports is
+ /// low. Alert on any increase.
+ ///
+ /// A `Counter`, not a `Gauge`: it only ever climbs within a process, so
+ /// `rate()` and `increase()` are the queries an operator wants, and both are
+ /// counter-only. The scrape handler samples the source and advances this by
+ /// the difference, which is why the source is monotonic per process.
+ stats_rollup_underflows: Counter,
Review Comment:
warning: this only advances inside the `/metrics` scrape, which needs both `http.enabled` and `http.metrics.enabled`. a tcp-only or quic-only deployment gets no counter at all, leaving one throttled warn line as the whole signal.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,86 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts. Clamping keeps the total merely low, which the
Review Comment:
warning: "the next write corrects" is wrong - increments are `fetch_add`, so they stack on the clamped base and the shortfall sticks until a rebuild or restart. this is what tells an operator whether the alarm needs a response.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while 7 `metadata::stm::stream` unit tests fail), and with only the
+//! reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead — `metadata::stm::stream` for the
+//! rollback and eviction, `server::partition_reconciler` for the teardown
+//! settle and the membership gate — because the integration layer cannot hold
+//! the reconciler off by configuration: the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! * `given_counted_partition_when_apply_delete_partitions_should_roll_it_out_of_the_parents`
+//! * `given_counted_topic_when_apply_delete_topic_should_roll_it_out_of_the_stream`
+//! * `given_many_counted_partitions_when_apply_delete_topic_should_roll_all_of_them_out`
+//! * `given_non_zero_base_partition_ids_when_apply_delete_partitions_should_keep_the_survivor`
+//! * `given_replayed_delete_partitions_when_applied_twice_should_not_double_roll_back`
+//! * `given_topic_residue_no_partition_entry_covers_when_apply_delete_topic_should_settle_it`
+//!
+//! The reads after each delete are deliberately single-shot. Retrying until the
+//! numbers converge, the way the pre-delete assertions do for the async send
+//! folding, would also wait out a broken rollback and pass regardless.
+
+use crate::server::scenarios::PARTITIONS_COUNT;
+use bytes::Bytes;
+use iggy::prelude::*;
+use integration::harness::{TestHarness, assert_clean_system, login_root};
+use std::str::FromStr;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// Committed partition ops fold into the shared stats on the owning shard, so a
+// read can race the apply window. Retry until the expectation holds, then make
+// the terminal assertion for a real mismatch.
+const STATS_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
Review Comment:
simplification: the consts, `create_messages` and the three validators are copies of `stream_size_validation_scenario.rs` - `validate_system_stats` byte for byte. hoist them into `scenarios/mod.rs`, which already owns the shared consts and helpers.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,86 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts. Clamping keeps the total merely low, which the
+/// next write corrects; this counter plus the `warn!` below is what tells an
+/// operator the divergence happened at all.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+fn report_rollup_underflow(scope: &'static str, counter: &'static str, shortfall: u64) {
+ let total = ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed) + 1;
+ // One line for the first, then at powers of two. A skewed tree emits up to
+ // three of these per counter per rollback, and a bulk delete would turn a
+ // diagnostic into a log flood that buries the first one. The counter above
+ // stays exact, and it is what an alert reads.
+ if total.is_power_of_two() {
+ warn!(
+ scope,
+ counter,
+ shortfall,
+ total,
+ "rollup decrement exceeded the total it was subtracted from; clamped at zero"
+ );
+ }
+}
+
+/// Subtract `amount`, clamping at zero instead of wrapping. Returns what was
+/// actually taken, which is what the caller passes on to its parent.
+///
+/// `fetch_update` rather than a load followed by a subtract: the counters are
+/// written from the metadata shard and from whichever shard owns the partition,
+/// so a separate load leaves a window where the clamp reads one value and
+/// subtracts from another.
+fn clamped_sub_u64(
+ counter: &AtomicU64,
+ amount: u64,
+ scope: &'static str,
+ name: &'static str,
+) -> u64 {
+ // The closure always yields `Some`, so `Err` carries the same previous
Review Comment:
simplification: the `unwrap_or_else` arm can't run - the closure always returns `Some`, so `fetch_update` never gives `Err`. four lines of comment justify a dead branch, twice.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,86 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts. Clamping keeps the total merely low, which the
+/// next write corrects; this counter plus the `warn!` below is what tells an
+/// operator the divergence happened at all.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+fn report_rollup_underflow(scope: &'static str, counter: &'static str, shortfall: u64) {
+ let total = ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed) + 1;
+ // One line for the first, then at powers of two. A skewed tree emits up to
+ // three of these per counter per rollback, and a bulk delete would turn a
+ // diagnostic into a log flood that buries the first one. The counter above
+ // stays exact, and it is what an alert reads.
+ if total.is_power_of_two() {
+ warn!(
Review Comment:
warning: no stream, topic or partition id here, so an operator alerting on the counter can't locate the divergence. return the shortfall and log where the ids are known - retention and purge are the likeliest source.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while 7 `metadata::stm::stream` unit tests fail), and with only the
+//! reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead — `metadata::stm::stream` for the
+//! rollback and eviction, `server::partition_reconciler` for the teardown
+//! settle and the membership gate — because the integration layer cannot hold
+//! the reconciler off by configuration: the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! * `given_counted_partition_when_apply_delete_partitions_should_roll_it_out_of_the_parents`
Review Comment:
nit: this hand-kept list already misses three of the nine tests this commit adds (`stream.rs:3542`, `:3656`, `:3759`), and it's split from the paragraph it belongs to. name the module instead.
##########
core/server/src/boot/mod.rs:
##########
@@ -552,9 +552,7 @@ async fn shard_main(
// and decrement the parent `StreamStats` by it.
let () = recovered.mux_stm.streams().read(|inner| {
Review Comment:
warning: this runs after the wal replay, so a replayed `DeleteTopic` over a torn checkpoint clamps against the stream and trips the new underflow counter on every such boot. zero the aggregates before replay.
##########
core/server/src/http/metrics.rs:
##########
@@ -45,6 +45,16 @@ pub(in crate::http) struct HttpMetrics {
pub(in crate::http) messages: Gauge,
pub(in crate::http) users: Gauge,
pub(in crate::http) clients: Gauge,
+ /// Not a legacy-parity metric: the running count of rollup decrements that
Review Comment:
warning: "alert on any increase" isn't actionable - a retention decrement against a settled partition trips it, and the new unit test drives exactly that case. restate as a bounded rate outside delete, purge, teardown and restore windows.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1137,18 +1146,68 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
.read(|inner| inner.revision)
}
-/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
-/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_partition_stats(
+/// Roll a torn-down partition's counters out of its parent topic and stream,
+/// and drop its registry entry.
+///
+/// Two handles, because they can differ. The registry entry is usually gone
+/// already (the metadata apply evicts it on the commit that acked the delete),
+/// but a stale incarnation being torn down after a slab-key reuse still has
+/// one, and it must not survive into the rebuild: `partition` is a
+/// get-or-create, so the rebuild would inherit the dead incarnation's counters
+/// and its `purged_generation`. `live` is the mounted partition's own handle,
+/// which is the only place the post-commit residue lives. When both name the
+/// same `Arc` the second zeroing rolls back 0.
+fn settle_partition_stats(
+ ctx: &ReconcilerCtx,
+ ns: IggyNamespace,
+ live: Option<&Arc<iggy_common::PartitionStats>>,
+) {
+ // Registry cloned out rather than mutated inside the read closure: the
+ // rollback cascades into parent totals, which the STM read has no part in.
+ let registry = ctx
+ .shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|inner| Arc::clone(&inner.stats_registry));
+ registry.remove_partition(ns.stream_id(), ns.topic_id(), ns.partition_id());
+ if let Some(stats) = live {
+ stats.zero_out_all();
+ }
+}
+
+/// Everything one namespace's build needs out of committed metadata: the shared
+/// `Arc<PartitionStats>`, the topic's runtime options, and the committed
+/// [`Partition`] record the disk loader reads `created_at` / `created_revision`
+/// / `created_view` from.
+///
+/// One read for all three. The `Partition` lookup doubles as the membership
+/// check: the committed topic has to still list this partition, because a pass
+/// captures its targets once and then awaits disk work per namespace, so
+/// without it a target captured before a `DeletePartitions` re-creates the
+/// entry that delete just evicted -- and the apply on the second left-right
+/// buffer, deferred to a later publish, then zeroes a live partition's counters
+/// and drops its entry again.
+///
+/// `None` if the stream, the topic, or the partition vanished between the
+/// target snapshot and this read.
+fn fetch_partition_build_inputs(
Review Comment:
nit: three references still name the old `fetch_partition_stats` - `stream.rs:411`, `responses.rs:1385`, and the dead intra-doc link at line 1107. that comment also still claims stats are fetched lazily, which this merge made false.
##########
core/server/src/boot/mod.rs:
##########
@@ -1121,4 +1140,55 @@ mod tests {
"ops with no partition-shape effect must stay off the broadcast"
);
}
+
+ /// A checkpoint loads a stream's total and its topics' as separate reads
+ /// while the partition plane keeps counting, so the two can disagree in
+ /// either direction. Boot has to drop both, or the surviving residue sits
+ /// under a total nothing on disk backs.
+ #[test]
+ fn clear_snapshot_totals_drops_both_levels_of_a_torn_checkpoint() {
+ use iggy_common::{
+ CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize, StreamStats,
+ };
+ use metadata::stm::stream::{Stream, Topic};
+
+ let stream_stats = Arc::new(StreamStats::default());
+ let mut stream = Stream::with_stats(
+ Arc::from("stream"),
+ IggyTimestamp::now(),
+ Arc::clone(&stream_stats),
+ );
+ let topic = Topic::new(
+ Arc::from("topic"),
+ IggyTimestamp::now(),
+ IggyExpiry::NeverExpire,
+ CompressionAlgorithm::default(),
+ MaxTopicSize::Unlimited,
+ Arc::clone(&stream_stats),
+ );
+ let topic_stats = Arc::clone(&topic.stats);
+ stream.topics.insert(topic);
+
+ // Topics summing above their stream is the torn shape: the stream was
+ // read first, and the topic kept counting before its own read.
+ stream_stats.store_from_snapshot(100, 1, 1);
+ topic_stats.store_from_snapshot(200, 2, 2);
+ let underflows_before = iggy_common::rollup_underflows();
Review Comment:
nit: `rollup_underflows()` is a process-global static, so this exact-equality assert breaks at a distance under `cargo test -p server`, where the whole crate shares one process. assert on the totals instead.
##########
core/server/src/boot/mod.rs:
##########
@@ -1032,6 +1030,27 @@ async fn shard_main(
/// and dropped (the periodic tick recovers). Installed via
/// [`metadata::IggyMetadata::set_commit_notifier`] on shard 0 only, the
/// sole writer of the metadata state machine.
Review Comment:
nit: `clear_snapshot_totals` landed between this doc block and the fn it describes, so both blocks now attach to the stats helper and `make_metadata_commit_notifier` ships undocumented. move it above line 1024 or below the notifier.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1137,18 +1146,68 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
.read(|inner| inner.revision)
}
-/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
-/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_partition_stats(
+/// Roll a torn-down partition's counters out of its parent topic and stream,
+/// and drop its registry entry.
+///
+/// Two handles, because they can differ. The registry entry is usually gone
+/// already (the metadata apply evicts it on the commit that acked the delete),
+/// but a stale incarnation being torn down after a slab-key reuse still has
+/// one, and it must not survive into the rebuild: `partition` is a
+/// get-or-create, so the rebuild would inherit the dead incarnation's counters
+/// and its `purged_generation`. `live` is the mounted partition's own handle,
+/// which is the only place the post-commit residue lives. When both name the
+/// same `Arc` the second zeroing rolls back 0.
+fn settle_partition_stats(
+ ctx: &ReconcilerCtx,
+ ns: IggyNamespace,
+ live: Option<&Arc<iggy_common::PartitionStats>>,
Review Comment:
nit: `&Arc<PartitionStats>` buys nothing over `&PartitionStats` here - the body only calls `zero_out_all()`. the call site becomes `live_stats.as_deref()`.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,86 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts. Clamping keeps the total merely low, which the
+/// next write corrects; this counter plus the `warn!` below is what tells an
+/// operator the divergence happened at all.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+#[must_use]
+pub fn rollup_underflows() -> u64 {
Review Comment:
nit: this reaches the root of the published `iggy_common` through the glob re-export, so a server-only diagnostic becomes semver-locked api. `#[doc(hidden)]` it.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -384,10 +389,29 @@ impl Stream {
/// read.
///
/// Shared across buffers and reader shards via `Arc` (a `StreamsInner` clone
-/// shares it). Only shard 0's writer mutates the maps, under the
-/// single-threaded Absorb; the `Mutex` is for `Sync` (uncontended), not
-/// concurrency. Ids are deterministic across replicas (same op order), so both
-/// buffers resolve the same key.
+/// shares it). Shard 0's writer owns the map mutations that ride the op-log,
+/// but it is NOT the only writer: [`Self::partition`] is a get-or-create the
+/// owning shard's reconciler calls when it materializes a namespace, and boot
+/// recovery calls it from every shard at once. The `Mutex` is load-bearing for
+/// that concurrency, not merely for `Sync`. Ids are deterministic across
+/// replicas (same op order), so both buffers resolve the same key.
+///
+/// Every eviction here therefore runs twice, once per buffer, and the second
+/// run is deferred. Two properties keep that from doing damage, and neither is
+/// guaranteed by the `Absorb` trait:
+///
+/// * `WriteCell::apply` is `append(cmd).publish()`, so a batch is one op. Batch
+/// two and a delete could be absorbed on the second buffer AFTER a create
+/// that re-minted the same id, and the delete would then evict the new
+/// partition's entry.
+/// * `left_right` drains the previous batch's `absorb_second` before the
+/// current batch's `absorb_first` (0.11.8 `write.rs`), which is an
+/// implementation detail, not a contract.
+///
+/// Independently of both, `fetch_partition_stats` refuses to register a
Review Comment:
warning: this doesn't close the live route - the check reads the left-right read side, but the eviction happens in `absorb_first` before `publish()`. a reader in that window recreates the entry the apply just dropped.
##########
core/server/src/boot/mod.rs:
##########
@@ -1121,4 +1140,55 @@ mod tests {
"ops with no partition-shape effect must stay off the broadcast"
);
}
+
+ /// A checkpoint loads a stream's total and its topics' as separate reads
+ /// while the partition plane keeps counting, so the two can disagree in
+ /// either direction. Boot has to drop both, or the surviving residue sits
+ /// under a total nothing on disk backs.
+ #[test]
+ fn clear_snapshot_totals_drops_both_levels_of_a_torn_checkpoint() {
+ use iggy_common::{
Review Comment:
nit: imports go at the top of the module - the sibling test above has none.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,86 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts. Clamping keeps the total merely low, which the
+/// next write corrects; this counter plus the `warn!` below is what tells an
+/// operator the divergence happened at all.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+fn report_rollup_underflow(scope: &'static str, counter: &'static str, shortfall: u64) {
+ let total = ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed) + 1;
+ // One line for the first, then at powers of two. A skewed tree emits up to
+ // three of these per counter per rollback, and a bulk delete would turn a
+ // diagnostic into a log flood that buries the first one. The counter above
+ // stays exact, and it is what an alert reads.
+ if total.is_power_of_two() {
Review Comment:
nit: the throttle keys on one global count across all six scope/counter pairs, so a first-ever divergence in a fresh scope stays silent for up to 1023 events. key it per pair, or on a time window.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1137,18 +1146,68 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
.read(|inner| inner.revision)
}
-/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
-/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_partition_stats(
+/// Roll a torn-down partition's counters out of its parent topic and stream,
+/// and drop its registry entry.
+///
+/// Two handles, because they can differ. The registry entry is usually gone
+/// already (the metadata apply evicts it on the commit that acked the delete),
+/// but a stale incarnation being torn down after a slab-key reuse still has
+/// one, and it must not survive into the rebuild: `partition` is a
+/// get-or-create, so the rebuild would inherit the dead incarnation's counters
+/// and its `purged_generation`. `live` is the mounted partition's own handle,
+/// which is the only place the post-commit residue lives. When both name the
+/// same `Arc` the second zeroing rolls back 0.
+fn settle_partition_stats(
+ ctx: &ReconcilerCtx,
+ ns: IggyNamespace,
+ live: Option<&Arc<iggy_common::PartitionStats>>,
+) {
+ // Registry cloned out rather than mutated inside the read closure: the
+ // rollback cascades into parent totals, which the STM read has no part in.
+ let registry = ctx
+ .shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|inner| Arc::clone(&inner.stats_registry));
+ registry.remove_partition(ns.stream_id(), ns.topic_id(), ns.partition_id());
+ if let Some(stats) = live {
+ stats.zero_out_all();
+ }
+}
+
+/// Everything one namespace's build needs out of committed metadata: the shared
+/// `Arc<PartitionStats>`, the topic's runtime options, and the committed
+/// [`Partition`] record the disk loader reads `created_at` / `created_revision`
+/// / `created_view` from.
+///
+/// One read for all three. The `Partition` lookup doubles as the membership
+/// check: the committed topic has to still list this partition, because a pass
+/// captures its targets once and then awaits disk work per namespace, so
+/// without it a target captured before a `DeletePartitions` re-creates the
+/// entry that delete just evicted -- and the apply on the second left-right
+/// buffer, deferred to a later publish, then zeroes a live partition's counters
+/// and drops its entry again.
+///
+/// `None` if the stream, the topic, or the partition vanished between the
+/// target snapshot and this read.
+fn fetch_partition_build_inputs(
ctx: &ReconcilerCtx,
ns: IggyNamespace,
) -> Option<(
Arc<iggy_common::PartitionStats>,
iggy_common::TopicRuntimeOptions,
+ Partition,
)> {
ctx.shard.plane.metadata().mux_stm.streams().read(|inner| {
let stream = inner.items.get(ns.stream_id())?;
let topic = stream.topics.get(ns.topic_id())?;
+ let partition = topic
+ .partitions
+ .iter()
+ .find(|partition| partition.id == ns.partition_id())?
Review Comment:
nit: this `find` now runs on every build, not just the disk-load branch, so materialising a k-partition topic is quadratic under the read guard. cheap next to the mkdir, but `iggy_partitions.rs:365` is the same shape.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -928,10 +926,21 @@ async fn tear_down_owned_partition(
// on_replicate / on_ack frames that haven't observed the queued
// tombstone yet. Idempotent on retry: already-tombstoned namespace
// stays tombstoned; already-removed shards_table row is a no-op.
+ // Take the counters handle BEFORE the tombstone (every partition accessor
+ // is tombstone-gated) and settle it AFTER, once the tombstone has stopped
+ // new frames from resolving the partition. The metadata apply already
+ // rolled it out at commit time, so what is left here is whatever landed in
+ // the window between the two: in-flight appends adding to parents that
+ // outlive the partition, and a retention tick subtracting from counters the
+ // apply had already zeroed. A frame already past the gate can still land
+ // after this settle, which is why the rollback clamps rather than trusting
+ // the amount it is handed.
+ let live_stats = partitions.with_partition(&ns, |partition| Arc::clone(&partition.stats));
Review Comment:
nit: "take the counters handle BEFORE the tombstone" reads like the retry path is covered, but `with_partition` is tombstone-gated, so on retry this is always `None` and the settle is a no-op. say so.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1817,6 +1862,96 @@ mod tests {
);
}
+ /// A pass captures its targets once and then awaits disk work per
+ /// namespace, so a delete committed mid-pass leaves a stale target behind.
+ /// Resolving stats for it would get-or-CREATE the registry entry the delete
+ /// just evicted, and the apply on the second left-right buffer would then
+ /// zero and drop a live partition's counters.
+ #[compio::test]
+ async fn stats_are_refused_for_a_partition_the_committed_topic_no_longer_lists() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-a");
+ seed_topic(&mux, 2, 0, "topic-a", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ let listed = IggyNamespace::new(0, 0, 0);
+ assert!(
+ fetch_partition_build_inputs(&ctx, listed).is_some(),
+ "a partition the topic lists must still resolve"
+ );
+
+ let unlisted = IggyNamespace::new(0, 0, 7);
+ assert!(
+ fetch_partition_build_inputs(&ctx, unlisted).is_none(),
+ "a partition the committed topic does not list must not resolve"
+ );
+ assert!(
+ registry(&ctx).partition_get(0, 0, 7).is_none(),
+ "and the refusal must not have created its entry on the way"
+ );
+ }
+
+ /// The metadata apply rolls a deleted partition out of its parents at
+ /// commit, but the partition stays mounted until the reconciler tears it
+ /// down, and appends in that window land in parents with no entry left to
+ /// account for them. Teardown is where that residue gets settled.
+ #[compio::test]
+ async fn teardown_settles_the_residue_the_metadata_delete_could_not_reach() {
Review Comment:
nit: this still passes with the `registry.remove_partition` at line 1174 deleted, because the metadata apply already evicted the entry in this fixture. the stale-incarnation eviction has no test.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1137,18 +1146,68 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
.read(|inner| inner.revision)
}
-/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
-/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_partition_stats(
+/// Roll a torn-down partition's counters out of its parent topic and stream,
+/// and drop its registry entry.
+///
+/// Two handles, because they can differ. The registry entry is usually gone
+/// already (the metadata apply evicts it on the commit that acked the delete),
+/// but a stale incarnation being torn down after a slab-key reuse still has
+/// one, and it must not survive into the rebuild: `partition` is a
+/// get-or-create, so the rebuild would inherit the dead incarnation's counters
+/// and its `purged_generation`. `live` is the mounted partition's own handle,
+/// which is the only place the post-commit residue lives. When both name the
+/// same `Arc` the second zeroing rolls back 0.
+fn settle_partition_stats(
+ ctx: &ReconcilerCtx,
+ ns: IggyNamespace,
+ live: Option<&Arc<iggy_common::PartitionStats>>,
+) {
+ // Registry cloned out rather than mutated inside the read closure: the
+ // rollback cascades into parent totals, which the STM read has no part in.
+ let registry = ctx
Review Comment:
simplification: one full `streams().read()` per torn-down namespace just to clone the registry `Arc`. build it once into `ReconcilerCtx` - it's never re-minted after boot, and a topic delete opens one reader epoch per partition.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -550,23 +578,109 @@ impl StatsRegistry {
}
fn remove_topic(&self, stream_id: usize, topic_id: usize) {
- self.topics
+ // Partitions first: each one's rollback cascades through its parent
+ // topic into the stream, so zeroing the topic ahead of them would
+ // subtract the same bytes from the stream twice.
+ self.remove_partitions_where(|(sid, tid, _)| *sid == stream_id && *tid == topic_id);
Review Comment:
simplification: this still uses the predicate sweep the comment at line 608 argues against, and `remove_stream` does too. the ids are in hand at `:2234`, and every registry insert takes them from the committed topic, so keying is safe.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
Review Comment:
nit: the literal "7 unit tests fail" is a number nothing maintains - it goes stale on the next test added to that module. keep the conclusion, drop the count.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while 7 `metadata::stm::stream` unit tests fail), and with only the
+//! reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead — `metadata::stm::stream` for the
+//! rollback and eviction, `server::partition_reconciler` for the teardown
+//! settle and the membership gate — because the integration layer cannot hold
+//! the reconciler off by configuration: the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! * `given_counted_partition_when_apply_delete_partitions_should_roll_it_out_of_the_parents`
+//! * `given_counted_topic_when_apply_delete_topic_should_roll_it_out_of_the_stream`
+//! * `given_many_counted_partitions_when_apply_delete_topic_should_roll_all_of_them_out`
+//! * `given_non_zero_base_partition_ids_when_apply_delete_partitions_should_keep_the_survivor`
+//! * `given_replayed_delete_partitions_when_applied_twice_should_not_double_roll_back`
+//! * `given_topic_residue_no_partition_entry_covers_when_apply_delete_topic_should_settle_it`
+//!
+//! The reads after each delete are deliberately single-shot. Retrying until the
+//! numbers converge, the way the pre-delete assertions do for the async send
+//! folding, would also wait out a broken rollback and pass regardless.
+
+use crate::server::scenarios::PARTITIONS_COUNT;
+use bytes::Bytes;
+use iggy::prelude::*;
+use integration::harness::{TestHarness, assert_clean_system, login_root};
+use std::str::FromStr;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// Committed partition ops fold into the shared stats on the owning shard, so a
+// read can race the apply window. Retry until the expectation holds, then make
+// the terminal assertion for a real mismatch.
+const STATS_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
+const STATS_RETRY_INTERVAL: Duration = Duration::from_millis(100);
+
+const STREAM_NAME: &str = "delete-stats-stream";
+const KEPT_TOPIC: &str = "kept-topic";
+const DELETED_TOPIC: &str = "deleted-topic";
+const MESSAGE_PAYLOAD_SIZE_BYTES: u64 = 57;
+const MSGS_COUNT: u64 = 17;
+
+// The server accounts the on-disk batch framing: one command header per append
+// pass plus a per-message header. Mirrors `stream_size_validation_scenario`.
+const NG_BATCH_HEADER_SIZE: u64 = 256;
+const NG_MESSAGE_HEADER_SIZE: u64 = 48;
+const MSGS_SIZE: u64 =
+ NG_BATCH_HEADER_SIZE + (NG_MESSAGE_HEADER_SIZE + MESSAGE_PAYLOAD_SIZE_BYTES) * MSGS_COUNT;
+
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .new_client()
+ .await
+ .expect("Failed to create new client");
+ client.ping().await.unwrap();
+ login_root(&client).await.expect("login failed");
+
+ // Partition half first: its assertions are the tighter pair.
+ delete_partitions_rolls_back_topic_and_stream(&client).await;
+ delete_topic_rolls_back_the_stream(&client).await;
+
+ assert_clean_system(&client).await;
+}
+
+/// Deleting partitions that still hold messages must take their bytes out of
+/// both the topic and the stream. The retained partition's messages must stay
+/// counted, so a blanket zeroing fails here as loudly as no rollback at all.
+async fn delete_partitions_rolls_back_topic_and_stream(client: &IggyClient) {
+ create_stream(client, STREAM_NAME).await;
+ create_topic(client, STREAM_NAME, KEPT_TOPIC).await;
+
+ // One pass per partition, so the delete below removes a known share.
+ for partition_id in 0..PARTITIONS_COUNT {
+ send_one_pass(client, STREAM_NAME, KEPT_TOPIC, partition_id).await;
+ }
+ let all_partitions = MSGS_SIZE * u64::from(PARTITIONS_COUNT);
+ let all_messages = MSGS_COUNT * u64::from(PARTITIONS_COUNT);
+ validate_topic(
+ client,
+ STREAM_NAME,
+ KEPT_TOPIC,
+ all_partitions,
+ all_messages,
+ )
+ .await;
+ validate_stream(client, STREAM_NAME, all_partitions, all_messages).await;
+
+ client
+ .delete_partitions(
+ &Identifier::from_str(STREAM_NAME).unwrap(),
+ &Identifier::from_str(KEPT_TOPIC).unwrap(),
+ 1,
+ )
+ .await
+ .unwrap();
+
+ // Single-shot again: the retained partitions' messages must still be
+ // counted, so this fails both on a missing rollback and on a blanket
+ // zeroing of the parents.
+ let retained = u64::from(PARTITIONS_COUNT - 1);
+ let topic = client
+ .get_topic(
+ &Identifier::from_str(STREAM_NAME).unwrap(),
+ &Identifier::from_str(KEPT_TOPIC).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("Failed to get topic");
+ assert_eq!(
+ topic.size,
+ MSGS_SIZE * retained,
+ "the deleted partitions' bytes must leave the topic total on the delete's commit"
+ );
+ assert_eq!(topic.messages_count, MSGS_COUNT * retained);
+ let stream = client
+ .get_stream(&Identifier::from_str(STREAM_NAME).unwrap())
+ .await
+ .unwrap()
+ .expect("Failed to get stream");
+ assert_eq!(
+ stream.size,
+ MSGS_SIZE * retained,
+ "the deleted partitions' bytes must leave the stream total too"
+ );
+ assert_eq!(stream.messages_count, MSGS_COUNT * retained);
+
+ client
+ .delete_stream(&Identifier::from_str(STREAM_NAME).unwrap())
+ .await
+ .unwrap();
+}
Review Comment:
nit: missing blank line before the next doc comment - every other item boundary in this file has one.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,86 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts. Clamping keeps the total merely low, which the
+/// next write corrects; this counter plus the `warn!` below is what tells an
+/// operator the divergence happened at all.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+fn report_rollup_underflow(scope: &'static str, counter: &'static str, shortfall: u64) {
+ let total = ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed) + 1;
+ // One line for the first, then at powers of two. A skewed tree emits up to
+ // three of these per counter per rollback, and a bulk delete would turn a
+ // diagnostic into a log flood that buries the first one. The counter above
+ // stays exact, and it is what an alert reads.
+ if total.is_power_of_two() {
+ warn!(
+ scope,
+ counter,
+ shortfall,
+ total,
+ "rollup decrement exceeded the total it was subtracted from; clamped at zero"
+ );
+ }
+}
+
+/// Subtract `amount`, clamping at zero instead of wrapping. Returns what was
+/// actually taken, which is what the caller passes on to its parent.
+///
+/// `fetch_update` rather than a load followed by a subtract: the counters are
+/// written from the metadata shard and from whichever shard owns the partition,
+/// so a separate load leaves a window where the clamp reads one value and
+/// subtracts from another.
+fn clamped_sub_u64(
Review Comment:
simplification: `clamped_sub_u64` and `clamped_sub_u32` are the same 16 lines twice, differing only in the atomic type. one `macro_rules!` covers both - a sealed trait would be longer than the duplication.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -550,23 +578,109 @@ impl StatsRegistry {
}
fn remove_topic(&self, stream_id: usize, topic_id: usize) {
- self.topics
+ // Partitions first: each one's rollback cascades through its parent
+ // topic into the stream, so zeroing the topic ahead of them would
+ // subtract the same bytes from the stream twice.
+ self.remove_partitions_where(|(sid, tid, _)| *sid == stream_id && *tid == topic_id);
+ let topic = self
+ .topics
.lock()
.expect("stats registry mutex poisoned")
.remove(&(stream_id, topic_id));
- self.partitions
- .lock()
- .expect("stats registry mutex poisoned")
- .retain(|(sid, tid, _), _| !(*sid == stream_id && *tid == topic_id));
+ // Whatever the topic still counts after the loop above is what its
+ // partitions did not account for: a snapshot restore that stored a
+ // total this node's partition entries never contributed, or bytes a
+ // partition kept adding after its own entry was evicted. Swapping it
+ // out settles that residue on the stream instead of stranding it.
+ if let Some(topic) = topic {
+ topic.zero_out_all();
+ }
}
- fn remove_partitions_from(&self, stream_id: usize, topic_id: usize, first_removed: usize) {
- self.partitions
- .lock()
- .expect("stats registry mutex poisoned")
- .retain(|(sid, tid, pid), _| {
- !(*sid == stream_id && *tid == topic_id && *pid >= first_removed)
+ /// Roll the named partitions out of their parents and drop their entries.
+ ///
+ /// Ids, never positions. `DeletePartitions` truncates the tail of a `Vec`,
+ /// and a count-based predicate (`id >= retained`) only picks the same set
+ /// while ids happen to be dense; when they are not it evicts and zeroes a
+ /// SURVIVING partition, which strips its bytes from the parents and resets
+ /// the offset its clients store against.
+ fn remove_partitions(&self, stream_id: usize, topic_id: usize, removed_ids: &[usize]) {
+ // Keyed removes, not a predicate: the ids are known, and a predicate
+ // makes the map walk every entry on the node to drop a handful of them.
+ // Bulk deletes call this per topic, so that walk squares.
+ let dropped: Vec<Arc<PartitionStats>> = {
+ let mut entries = self
+ .partitions
+ .lock()
+ .expect("stats registry mutex poisoned");
+ removed_ids
+ .iter()
+ .filter_map(|partition_id| entries.remove(&(stream_id, topic_id, *partition_id)))
+ .map(|entry| entry.stats)
+ .collect()
+ };
+ // Guard released first: the rollback cascades into parent totals, which
+ // the partition map has no part in.
+ for stats in dropped {
+ stats.zero_out_all();
+ }
+ }
+
+ /// Drop every partition entry `should_drop` selects, rolling each one's
+ /// counters out of its parent topic and stream on the way.
+ ///
+ /// Both halves are load-bearing. A partition reports by incrementing its
+ /// parents through the `Arc`, so evicting alone strands what it contributed
+ /// in the topic and stream totals. And zeroing alone leaves an entry that
+ /// outlives its ids: a topic's slab key is recycled by the next
+ /// `create_topic`, and `DeletePartitions` truncates the tail so the next
+ /// `CreatePartitions` mints the freed ids again. The survivor would hand
+ /// its successor both its counters and its `purged_generation`, and that
+ /// stale generation makes the next purge's gate skip the reset.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ fn remove_partitions_where(&self, should_drop: impl Fn(&(usize, usize, usize)) -> bool) {
Review Comment:
simplification: generic over a predicate with exactly one caller. inline it as `remove_topic_partitions(stream_id, topic_id)` and drop the single-instantiation generic.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while 7 `metadata::stm::stream` unit tests fail), and with only the
+//! reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead — `metadata::stm::stream` for the
+//! rollback and eviction, `server::partition_reconciler` for the teardown
+//! settle and the membership gate — because the integration layer cannot hold
+//! the reconciler off by configuration: the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! * `given_counted_partition_when_apply_delete_partitions_should_roll_it_out_of_the_parents`
+//! * `given_counted_topic_when_apply_delete_topic_should_roll_it_out_of_the_stream`
+//! * `given_many_counted_partitions_when_apply_delete_topic_should_roll_all_of_them_out`
+//! * `given_non_zero_base_partition_ids_when_apply_delete_partitions_should_keep_the_survivor`
+//! * `given_replayed_delete_partitions_when_applied_twice_should_not_double_roll_back`
+//! * `given_topic_residue_no_partition_entry_covers_when_apply_delete_topic_should_settle_it`
+//!
+//! The reads after each delete are deliberately single-shot. Retrying until the
+//! numbers converge, the way the pre-delete assertions do for the async send
+//! folding, would also wait out a broken rollback and pass regardless.
+
+use crate::server::scenarios::PARTITIONS_COUNT;
+use bytes::Bytes;
+use iggy::prelude::*;
+use integration::harness::{TestHarness, assert_clean_system, login_root};
+use std::str::FromStr;
Review Comment:
nit: `Identifier::named` is what the other scenarios use - `from_str` reinterprets a numeric-looking name as an id. no live bug with these names, just the wrong constructor.
also at the other 15 call sites in this file.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while 7 `metadata::stm::stream` unit tests fail), and with only the
+//! reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead — `metadata::stm::stream` for the
+//! rollback and eviction, `server::partition_reconciler` for the teardown
+//! settle and the membership gate — because the integration layer cannot hold
+//! the reconciler off by configuration: the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! * `given_counted_partition_when_apply_delete_partitions_should_roll_it_out_of_the_parents`
+//! * `given_counted_topic_when_apply_delete_topic_should_roll_it_out_of_the_stream`
+//! * `given_many_counted_partitions_when_apply_delete_topic_should_roll_all_of_them_out`
+//! * `given_non_zero_base_partition_ids_when_apply_delete_partitions_should_keep_the_survivor`
+//! * `given_replayed_delete_partitions_when_applied_twice_should_not_double_roll_back`
+//! * `given_topic_residue_no_partition_entry_covers_when_apply_delete_topic_should_settle_it`
+//!
+//! The reads after each delete are deliberately single-shot. Retrying until the
+//! numbers converge, the way the pre-delete assertions do for the async send
+//! folding, would also wait out a broken rollback and pass regardless.
+
+use crate::server::scenarios::PARTITIONS_COUNT;
+use bytes::Bytes;
+use iggy::prelude::*;
+use integration::harness::{TestHarness, assert_clean_system, login_root};
+use std::str::FromStr;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// Committed partition ops fold into the shared stats on the owning shard, so a
+// read can race the apply window. Retry until the expectation holds, then make
+// the terminal assertion for a real mismatch.
+const STATS_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
+const STATS_RETRY_INTERVAL: Duration = Duration::from_millis(100);
+
+const STREAM_NAME: &str = "delete-stats-stream";
+const KEPT_TOPIC: &str = "kept-topic";
+const DELETED_TOPIC: &str = "deleted-topic";
+const MESSAGE_PAYLOAD_SIZE_BYTES: u64 = 57;
+const MSGS_COUNT: u64 = 17;
+
+// The server accounts the on-disk batch framing: one command header per append
+// pass plus a per-message header. Mirrors `stream_size_validation_scenario`.
+const NG_BATCH_HEADER_SIZE: u64 = 256;
+const NG_MESSAGE_HEADER_SIZE: u64 = 48;
+const MSGS_SIZE: u64 =
+ NG_BATCH_HEADER_SIZE + (NG_MESSAGE_HEADER_SIZE + MESSAGE_PAYLOAD_SIZE_BYTES) * MSGS_COUNT;
+
+pub async fn run(harness: &TestHarness) {
+ let client = harness
Review Comment:
simplification: this re-implements `scenarios::create_client` verbatim, panic string included. import it instead.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,346 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while 7 `metadata::stm::stream` unit tests fail), and with only the
+//! reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead — `metadata::stm::stream` for the
+//! rollback and eviction, `server::partition_reconciler` for the teardown
+//! settle and the membership gate — because the integration layer cannot hold
+//! the reconciler off by configuration: the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! * `given_counted_partition_when_apply_delete_partitions_should_roll_it_out_of_the_parents`
+//! * `given_counted_topic_when_apply_delete_topic_should_roll_it_out_of_the_stream`
+//! * `given_many_counted_partitions_when_apply_delete_topic_should_roll_all_of_them_out`
+//! * `given_non_zero_base_partition_ids_when_apply_delete_partitions_should_keep_the_survivor`
+//! * `given_replayed_delete_partitions_when_applied_twice_should_not_double_roll_back`
+//! * `given_topic_residue_no_partition_entry_covers_when_apply_delete_topic_should_settle_it`
+//!
+//! The reads after each delete are deliberately single-shot. Retrying until the
+//! numbers converge, the way the pre-delete assertions do for the async send
+//! folding, would also wait out a broken rollback and pass regardless.
+
+use crate::server::scenarios::PARTITIONS_COUNT;
+use bytes::Bytes;
+use iggy::prelude::*;
+use integration::harness::{TestHarness, assert_clean_system, login_root};
+use std::str::FromStr;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// Committed partition ops fold into the shared stats on the owning shard, so a
+// read can race the apply window. Retry until the expectation holds, then make
+// the terminal assertion for a real mismatch.
+const STATS_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
+const STATS_RETRY_INTERVAL: Duration = Duration::from_millis(100);
+
+const STREAM_NAME: &str = "delete-stats-stream";
+const KEPT_TOPIC: &str = "kept-topic";
+const DELETED_TOPIC: &str = "deleted-topic";
+const MESSAGE_PAYLOAD_SIZE_BYTES: u64 = 57;
+const MSGS_COUNT: u64 = 17;
+
+// The server accounts the on-disk batch framing: one command header per append
+// pass plus a per-message header. Mirrors `stream_size_validation_scenario`.
+const NG_BATCH_HEADER_SIZE: u64 = 256;
+const NG_MESSAGE_HEADER_SIZE: u64 = 48;
+const MSGS_SIZE: u64 =
+ NG_BATCH_HEADER_SIZE + (NG_MESSAGE_HEADER_SIZE + MESSAGE_PAYLOAD_SIZE_BYTES) * MSGS_COUNT;
+
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .new_client()
+ .await
+ .expect("Failed to create new client");
+ client.ping().await.unwrap();
+ login_root(&client).await.expect("login failed");
+
+ // Partition half first: its assertions are the tighter pair.
+ delete_partitions_rolls_back_topic_and_stream(&client).await;
+ delete_topic_rolls_back_the_stream(&client).await;
+
+ assert_clean_system(&client).await;
+}
+
+/// Deleting partitions that still hold messages must take their bytes out of
+/// both the topic and the stream. The retained partition's messages must stay
+/// counted, so a blanket zeroing fails here as loudly as no rollback at all.
+async fn delete_partitions_rolls_back_topic_and_stream(client: &IggyClient) {
+ create_stream(client, STREAM_NAME).await;
+ create_topic(client, STREAM_NAME, KEPT_TOPIC).await;
+
+ // One pass per partition, so the delete below removes a known share.
+ for partition_id in 0..PARTITIONS_COUNT {
+ send_one_pass(client, STREAM_NAME, KEPT_TOPIC, partition_id).await;
+ }
+ let all_partitions = MSGS_SIZE * u64::from(PARTITIONS_COUNT);
+ let all_messages = MSGS_COUNT * u64::from(PARTITIONS_COUNT);
+ validate_topic(
+ client,
+ STREAM_NAME,
+ KEPT_TOPIC,
+ all_partitions,
+ all_messages,
+ )
+ .await;
+ validate_stream(client, STREAM_NAME, all_partitions, all_messages).await;
+
+ client
+ .delete_partitions(
+ &Identifier::from_str(STREAM_NAME).unwrap(),
+ &Identifier::from_str(KEPT_TOPIC).unwrap(),
+ 1,
+ )
+ .await
+ .unwrap();
+
+ // Single-shot again: the retained partitions' messages must still be
+ // counted, so this fails both on a missing rollback and on a blanket
+ // zeroing of the parents.
+ let retained = u64::from(PARTITIONS_COUNT - 1);
+ let topic = client
+ .get_topic(
+ &Identifier::from_str(STREAM_NAME).unwrap(),
+ &Identifier::from_str(KEPT_TOPIC).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("Failed to get topic");
+ assert_eq!(
+ topic.size,
+ MSGS_SIZE * retained,
+ "the deleted partitions' bytes must leave the topic total on the delete's commit"
+ );
+ assert_eq!(topic.messages_count, MSGS_COUNT * retained);
+ let stream = client
+ .get_stream(&Identifier::from_str(STREAM_NAME).unwrap())
+ .await
+ .unwrap()
+ .expect("Failed to get stream");
+ assert_eq!(
+ stream.size,
+ MSGS_SIZE * retained,
+ "the deleted partitions' bytes must leave the stream total too"
+ );
+ assert_eq!(stream.messages_count, MSGS_COUNT * retained);
+
+ client
+ .delete_stream(&Identifier::from_str(STREAM_NAME).unwrap())
+ .await
+ .unwrap();
+}
+/// Deleting a topic that still holds messages must take its bytes out of the
+/// stream total, not just remove the topic.
+async fn delete_topic_rolls_back_the_stream(client: &IggyClient) {
+ create_stream(client, STREAM_NAME).await;
+ create_topic(client, STREAM_NAME, KEPT_TOPIC).await;
+ create_topic(client, STREAM_NAME, DELETED_TOPIC).await;
+
+ send_one_pass(client, STREAM_NAME, KEPT_TOPIC, 0).await;
+ send_one_pass(client, STREAM_NAME, DELETED_TOPIC, 0).await;
+ validate_stream(client, STREAM_NAME, MSGS_SIZE * 2, MSGS_COUNT * 2).await;
+
+ client
+ .delete_topic(
+ &Identifier::from_str(STREAM_NAME).unwrap(),
+ &Identifier::from_str(DELETED_TOPIC).unwrap(),
+ )
+ .await
+ .unwrap();
+
+ // Read once, with no convergence retry: the delete acks on the metadata
+ // commit, so the totals must already be right in the reply the client is
+ // holding. A retry here would wait for the reconciler's on-disk wipe and
+ // hide the window entirely.
+ let stream = client
+ .get_stream(&Identifier::from_str(STREAM_NAME).unwrap())
+ .await
+ .unwrap()
+ .expect("Failed to get stream");
+ assert_eq!(
+ stream.size, MSGS_SIZE,
+ "the deleted topic's bytes must leave the stream total on the commit that acked the delete"
+ );
+ assert_eq!(stream.messages_count, MSGS_COUNT);
+ validate_topic(client, STREAM_NAME, KEPT_TOPIC, MSGS_SIZE, MSGS_COUNT).await;
+ validate_system_stats(client, MSGS_SIZE, MSGS_COUNT).await;
+
+ client
+ .delete_stream(&Identifier::from_str(STREAM_NAME).unwrap())
+ .await
+ .unwrap();
+}
+
+async fn create_stream(client: &IggyClient, stream_name: &str) {
Review Comment:
simplification: one-line passthrough with two call sites. inline it.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -550,23 +578,109 @@ impl StatsRegistry {
}
fn remove_topic(&self, stream_id: usize, topic_id: usize) {
- self.topics
+ // Partitions first: each one's rollback cascades through its parent
+ // topic into the stream, so zeroing the topic ahead of them would
+ // subtract the same bytes from the stream twice.
+ self.remove_partitions_where(|(sid, tid, _)| *sid == stream_id && *tid == topic_id);
+ let topic = self
+ .topics
.lock()
.expect("stats registry mutex poisoned")
.remove(&(stream_id, topic_id));
- self.partitions
- .lock()
- .expect("stats registry mutex poisoned")
- .retain(|(sid, tid, _), _| !(*sid == stream_id && *tid == topic_id));
+ // Whatever the topic still counts after the loop above is what its
+ // partitions did not account for: a snapshot restore that stored a
+ // total this node's partition entries never contributed, or bytes a
+ // partition kept adding after its own entry was evicted. Swapping it
+ // out settles that residue on the stream instead of stranding it.
+ if let Some(topic) = topic {
+ topic.zero_out_all();
+ }
}
- fn remove_partitions_from(&self, stream_id: usize, topic_id: usize, first_removed: usize) {
- self.partitions
- .lock()
- .expect("stats registry mutex poisoned")
- .retain(|(sid, tid, pid), _| {
- !(*sid == stream_id && *tid == topic_id && *pid >= first_removed)
+ /// Roll the named partitions out of their parents and drop their entries.
+ ///
+ /// Ids, never positions. `DeletePartitions` truncates the tail of a `Vec`,
+ /// and a count-based predicate (`id >= retained`) only picks the same set
+ /// while ids happen to be dense; when they are not it evicts and zeroes a
+ /// SURVIVING partition, which strips its bytes from the parents and resets
+ /// the offset its clients store against.
+ fn remove_partitions(&self, stream_id: usize, topic_id: usize, removed_ids: &[usize]) {
+ // Keyed removes, not a predicate: the ids are known, and a predicate
+ // makes the map walk every entry on the node to drop a handful of them.
+ // Bulk deletes call this per topic, so that walk squares.
+ let dropped: Vec<Arc<PartitionStats>> = {
+ let mut entries = self
+ .partitions
+ .lock()
+ .expect("stats registry mutex poisoned");
+ removed_ids
+ .iter()
+ .filter_map(|partition_id| entries.remove(&(stream_id, topic_id, *partition_id)))
+ .map(|entry| entry.stats)
+ .collect()
+ };
+ // Guard released first: the rollback cascades into parent totals, which
+ // the partition map has no part in.
+ for stats in dropped {
+ stats.zero_out_all();
+ }
+ }
+
+ /// Drop every partition entry `should_drop` selects, rolling each one's
+ /// counters out of its parent topic and stream on the way.
+ ///
+ /// Both halves are load-bearing. A partition reports by incrementing its
+ /// parents through the `Arc`, so evicting alone strands what it contributed
+ /// in the topic and stream totals. And zeroing alone leaves an entry that
+ /// outlives its ids: a topic's slab key is recycled by the next
+ /// `create_topic`, and `DeletePartitions` truncates the tail so the next
+ /// `CreatePartitions` mints the freed ids again. The survivor would hand
+ /// its successor both its counters and its `purged_generation`, and that
+ /// stale generation makes the next purge's gate skip the reset.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ fn remove_partitions_where(&self, should_drop: impl Fn(&(usize, usize, usize)) -> bool) {
+ let dropped: Vec<Arc<PartitionStats>> = {
+ let mut entries = self
+ .partitions
+ .lock()
+ .expect("stats registry mutex poisoned");
+ let mut dropped = Vec::new();
+ entries.retain(|key, entry| {
+ if should_drop(key) {
+ dropped.push(Arc::clone(&entry.stats));
+ return false;
+ }
+ true
});
+ dropped
+ };
+ // Guard released first: the rollback cascades into parent totals, which
+ // the partition map has no part in.
+ for stats in dropped {
+ stats.zero_out_all();
+ }
+ }
+
+ /// Roll one partition out of its parents and drop its entry.
+ ///
+ /// For the reconciler's owned-partition teardown. It tombstones the
+ /// namespace first, which stops any NEW frame from resolving the partition;
+ /// a frame already past that gate can still finish and touch the counters
+ /// afterwards, so this settles the bulk rather than closing the window. The
+ /// clamped rollback keeps that remainder from being taken out of a
+ /// sibling's data.
+ ///
+ /// The metadata apply has usually evicted the entry already, in which case
+ /// this is a no-op and the caller's own handle carries whatever landed
+ /// after; a stale incarnation being torn down after a slab-key reuse still
+ /// has its entry, and it must not survive into the rebuild.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ pub fn remove_partition(&self, stream_id: usize, topic_id: usize, partition_id: usize) {
Review Comment:
simplification: three-line wrapper over `remove_partitions` with one caller. call it directly with a one-element slice - `remove_partitions` has to go `pub`, so the public item count is unchanged.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4046:
URL: https://github.com/apache/iggy/pull/4046#discussion_r3955158489
##########
core/shard/src/lib.rs:
##########
@@ -2537,6 +2537,17 @@ where
// the floor with the partition value.
self.metrics
.record_partition_prepare_gap_drops(partition.take_prepare_gap_drops());
+ // Roll whatever this partition still counts out of its
+ // parent topic and stream. Here and not in the
+ // reconciler's teardown: a handler suspended mid-append
+ // holds its own `Arc` past the tombstone, and an earlier
+ // settle leaves its increment in the parents with the
+ // partition already gone. This is the drop point, so
+ // nothing can add through that handle afterwards. The
+ // rollback clamps, so the usual case -- the metadata
+ // apply already zeroed these counters at commit -- takes
+ // nothing.
+ partition.stats.zero_out_all();
Review Comment:
Leaving this one, with a reason rather than a fix.
The discarded build shares the namespace's `PartitionStats` Arc with the live incarnation, so neither arm can roll back what it folded: zeroing there is exactly the clobber that already cost a live partition its `current_offset` once.
Folding at adoption is the right shape, but it means the loader and the builder stop touching the shared Arc at all and hand `insert` a local delta instead. `load_partition` counts as it walks the chain, and `load_partition_or_fence` zeroes the shared Arc on a refusal, so that is a boot-path change rather than a local one. Both arms are unreachable today and already log plus count as damage control, so the refactor is out of proportion to this PR. Worth a separate issue.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+///
+/// Hidden from the docs: `iggy_common` re-exports this module at its root, and
+/// a server-only diagnostic has no business in the published SDK surface.
+#[doc(hidden)]
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+/// One scope/counter pair's clamp state: the labels an operator reads, plus the
+/// count the log throttle keys on.
+struct UnderflowSite {
+ scope: &'static str,
+ counter: &'static str,
+ clamped: AtomicU64,
+}
+
+impl UnderflowSite {
+ const fn new(scope: &'static str, counter: &'static str) -> Self {
+ Self {
+ scope,
+ counter,
+ clamped: AtomicU64::new(0),
+ }
+ }
+
+ /// Count the clamp globally, then log this pair's first and every power of
+ /// two after it.
+ ///
+ /// The throttle is per pair, not global: a skewed tree emits up to three
+ /// lines per counter per rollback and a bulk delete turns that into a
+ /// flood, so a shared count would hold a quiet pair's first-ever
+ /// divergence back for up to 1023 events behind a noisy one.
+ fn report(&self, shortfall: u64) {
+ ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed);
+ let clamped = self.clamped.fetch_add(1, Ordering::Relaxed) + 1;
+ if clamped.is_power_of_two() {
+ warn!(
+ scope = self.scope,
+ counter = self.counter,
+ shortfall,
+ clamped,
+ "rollup decrement exceeded the total it was subtracted from; clamped at zero"
+ );
+ }
+ }
+}
+
+static STREAM_SIZE_BYTES: UnderflowSite = UnderflowSite::new("stream", "size_bytes");
+static STREAM_MESSAGES_COUNT: UnderflowSite = UnderflowSite::new("stream", "messages_count");
+static STREAM_SEGMENTS_COUNT: UnderflowSite = UnderflowSite::new("stream", "segments_count");
Review Comment:
Keeping the nine statics.
The trade you name is the deciding one: three sites plus a `counter` argument drops the per-pair throttle, and that property is why the throttle is not global already. A bulk delete floods one pair, and a shared count would hold a quiet pair's first-ever divergence back behind it for up to 1023 events.
Nine flat statics is the cheapest shape that keeps it. An array per scope needs a counter enum plus index plumbing at every call site, which is more code than the nine lines it saves.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4046:
URL: https://github.com/apache/iggy/pull/4046#issuecomment-5580688222
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4046:
URL: https://github.com/apache/iggy/pull/4046#discussion_r3952415105
##########
core/metadata/src/stm/stream.rs:
##########
@@ -603,7 +719,82 @@ impl StatsRegistry {
self.partitions
.lock()
.expect("stats registry mutex poisoned")
- .retain(|key, _| live_partitions.contains(key));
+ .retain(|key, entry| {
Review Comment:
critical: this eviction doesn't zero the entry's stats, and the still-mounted stale incarnation holds the same `Arc` - its later `ConfirmRemove` subtracts those bytes again out of live siblings. zero the dropped entries here, like `remove_partitions` does.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+///
+/// Hidden from the docs: `iggy_common` re-exports this module at its root, and
+/// a server-only diagnostic has no business in the published SDK surface.
+#[doc(hidden)]
+#[must_use]
+pub fn rollup_underflows() -> u64 {
Review Comment:
warning: `doc(hidden)` hides this from rustdoc but not from the crate's public API, and the glob in `lib.rs` puts it at the `iggy_common` root with nothing in CI to catch it. rename to `stats_rollup_underflows()`, or reach it through a named `pub mod`.
##########
core/shard/src/lib.rs:
##########
@@ -2537,6 +2537,17 @@ where
// the floor with the partition value.
self.metrics
.record_partition_prepare_gap_drops(partition.take_prepare_gap_drops());
+ // Roll whatever this partition still counts out of its
+ // parent topic and stream. Here and not in the
+ // reconciler's teardown: a handler suspended mid-append
+ // holds its own `Arc` past the tombstone, and an earlier
+ // settle leaves its increment in the parents with the
+ // partition already gone. This is the drop point, so
+ // nothing can add through that handle afterwards. The
+ // rollback clamps, so the usual case -- the metadata
+ // apply already zeroed these counters at commit -- takes
+ // nothing.
+ partition.stats.zero_out_all();
Review Comment:
nit: this drop point rolls its counters back, but the two discard arms above at 2469 and 2493 just `drop(partition)` and leave what the build already folded into the shared stats. both are unreachable today - fold at adoption in `IggyPartitions::insert` instead.
##########
core/server/src/http/metrics.rs:
##########
@@ -203,6 +245,20 @@ mod tests {
);
}
+ #[test]
+ fn rollup_underflow_total_tracks_the_sampled_count() {
+ let metrics = HttpMetrics::init(&[]);
+ metrics.observe_rollup_underflows(3);
+ metrics.observe_rollup_underflows(5);
+ // A restart resets the source; the counter must not run backwards.
Review Comment:
nit: a restart can't produce this state - the static and `HttpMetrics` are recreated together, so `recorded` never exceeds `total`. say the call guards a non-monotonic source defensively, or drop it.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -439,22 +474,30 @@ impl StatsRegistry {
/// reads the same `Arc`, so partition-plane counters are visible
/// cross-shard without a gather.
///
+ /// Takes the committed record rather than a bare id, because a fresh entry
Review Comment:
nit: the doc says committed record, but not that it has to be the record for this `(stream_id, topic_id)` key, or that `parent` must be the committed topic's `Arc`. neither is checked, and a fresh entry inherits both.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+///
+/// Hidden from the docs: `iggy_common` re-exports this module at its root, and
+/// a server-only diagnostic has no business in the published SDK surface.
+#[doc(hidden)]
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+/// One scope/counter pair's clamp state: the labels an operator reads, plus the
+/// count the log throttle keys on.
+struct UnderflowSite {
+ scope: &'static str,
+ counter: &'static str,
+ clamped: AtomicU64,
+}
+
+impl UnderflowSite {
+ const fn new(scope: &'static str, counter: &'static str) -> Self {
+ Self {
+ scope,
+ counter,
+ clamped: AtomicU64::new(0),
+ }
+ }
+
+ /// Count the clamp globally, then log this pair's first and every power of
+ /// two after it.
+ ///
+ /// The throttle is per pair, not global: a skewed tree emits up to three
+ /// lines per counter per rollback and a bulk delete turns that into a
+ /// flood, so a shared count would hold a quiet pair's first-ever
+ /// divergence back for up to 1023 events behind a noisy one.
+ fn report(&self, shortfall: u64) {
+ ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed);
+ let clamped = self.clamped.fetch_add(1, Ordering::Relaxed) + 1;
Review Comment:
nit: `clamped` is the per-pair count but the metric it bumps is process-global and carries no scope or counter label, so the two numbers can't be reconciled. log both, or label the metric.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -282,21 +399,39 @@ impl PartitionStats {
self.parent.increment_segments_count(segments_count);
}
- pub fn decrement_size_bytes(&self, size_bytes: u64) {
- self.size_bytes.fetch_sub(size_bytes, Ordering::AcqRel);
- self.decrement_parent_size_bytes(size_bytes);
- }
-
- pub fn decrement_messages_count(&self, messages_count: u64) {
- self.messages_count
- .fetch_sub(messages_count, Ordering::AcqRel);
- self.decrement_parent_messages_count(messages_count);
- }
-
- pub fn decrement_segments_count(&self, segments_count: u32) {
- self.segments_count
- .fetch_sub(segments_count, Ordering::AcqRel);
- self.decrement_parent_segments_count(segments_count);
+ // Forward what this level actually gave up, not what was asked for. A
+ // decrement bigger than the counter holds means the bytes were never in
+ // this subtree, so the ancestors do not hold them either -- passing the
+ // full amount up would take them out of a sibling's live data instead.
+ // Under `parent == sum(children)` the two are equal and nothing changes.
+ //
+ // Each returns this level's shortfall -- how much of the amount it did not
+ // hold, zero when the decrement was covered -- so a caller that knows the
+ // ids can name where the divergence is.
+ pub fn decrement_size_bytes(&self, size_bytes: u64) -> u64 {
+ let taken = clamped_sub_u64(&self.size_bytes, size_bytes, &PARTITION_SIZE_BYTES);
+ self.decrement_parent_size_bytes(taken);
+ size_bytes - taken
+ }
+
+ pub fn decrement_messages_count(&self, messages_count: u64) -> u64 {
+ let taken = clamped_sub_u64(
+ &self.messages_count,
+ messages_count,
+ &PARTITION_MESSAGES_COUNT,
+ );
+ self.decrement_parent_messages_count(taken);
+ messages_count - taken
+ }
+
+ pub fn decrement_segments_count(&self, segments_count: u32) -> u32 {
+ let taken = clamped_sub_u32(
+ &self.segments_count,
+ segments_count,
+ &PARTITION_SEGMENTS_COUNT,
+ );
+ self.decrement_parent_segments_count(taken);
+ segments_count - taken
}
pub fn decrement_parent_size_bytes(&self, size_bytes: u64) {
Review Comment:
simplification: these twelve `increment_parent_*` and `decrement_parent_*` wrappers have no caller outside this file, and they're public on a published crate. inline them - `self.parent.increment_size_bytes(x)` is the same one line.
##########
core/integration/tests/server/scenarios/mod.rs:
##########
@@ -72,6 +74,133 @@ const USERNAME_3: &str = "user3";
const CONSUMER_KIND: ConsumerKind = ConsumerKind::Consumer;
const MESSAGES_COUNT: u32 = 1337;
+// The partition plane applies committed ops asynchronously on the owning shard
+// (sends fold into the shared stats at commit-apply; purge/delete zero them
+// when the reconciler drives the wipe), so a read racing that window can see a
+// pre-apply value. Retry until the expectation holds, then make the terminal
+// assertion for a real mismatch.
+const STATS_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
Review Comment:
simplification: same 10s and 100ms as `POLL_CONVERGENCE_TIMEOUT` and `POLL_RETRY_INTERVAL` at 65, and both wait out the same convergence. keep one neutral pair and update the doc links at 114 and 205.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -694,11 +707,12 @@ async fn reconcile_additions(
continue;
}
- // Resolve the shared stats `Arc` only for namespaces actually
- // built, not once per committed partition every pass. A topic that
- // vanished between the target snapshot and this read defers to the
- // next pass.
- let Some((partition_stats, topic_runtime)) = fetch_partition_stats(ctx, ns) else {
+ // Resolved only for namespaces actually built, not once per committed
+ // partition every pass. A stream, topic or partition that vanished
+ // between the target snapshot and this read defers to the next pass.
+ let Some((partition_stats, topic_runtime, partition_metadata)) =
+ fetch_partition_build_inputs(ctx, ns)
Review Comment:
warning: `created_revision` is in hand here but the build and row still get the snapshot `epoch`, so a mid-pass recreate lands a dead-epoch row. skip on mismatch only when the dir is absent - a bare `continue` lets the next pass adopt the old segments.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -603,7 +719,82 @@ impl StatsRegistry {
self.partitions
.lock()
.expect("stats registry mutex poisoned")
- .retain(|key, _| live_partitions.contains(key));
+ .retain(|key, entry| {
+ live_partitions
+ .get(key)
+ .is_some_and(|created_revision| *created_revision == entry.created_revision)
+ });
+ }
+
+ /// Recompute every topic and stream total as the sum of the partition
+ /// entries this node holds.
+ ///
+ /// A snapshot carries the DONOR's topic and stream totals, but partition
+ /// counters are node-local and never snapshotted: they keep moving on the
+ /// receiver while the snapshot is captured, shipped and installed. Storing
+ /// the donor's totals over the receiver's own children leaves
+ /// `topic < sum(partitions)` as the ordinary post-transfer shape, and the
+ /// next delete then subtracts a child from a parent that never counted it.
+ /// Every rollback in this file assumes `parent == sum(children)`; this is
+ /// where that gets re-established.
+ ///
+ /// Partitions this node has not materialized yet contribute nothing, the
+ /// same convergence boot relies on: the reconciler registers each one and
+ /// folds its on-disk delta in as it goes.
+ ///
+ /// The `partitions` guard fences the MAP, not the counters, which the data
+ /// plane reaches through the `Arc`. An append landing between a child read
+ /// and its parent store is folded into the parent by `fetch_add` and then
+ /// overwritten, so the invariant is restored modulo whatever arrives during
+ /// the walk. That residue is bounded by the walk and by one append, and the
+ /// saturating rollback absorbs it; quiescing the data plane for a metadata
+ /// install would cost far more than it buys.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ // One acquisition for the whole walk: the stores it makes are plain atomic
+ // writes on the topic and stream `Arc`s, with no parent cascade and no way
+ // back into the map, so there is nothing for a tighter guard to protect.
+ #[allow(clippy::significant_drop_tightening)]
+ fn rebuild_parent_totals(&self, streams: &IdSlab<Stream>) {
Review Comment:
warning: this holds the `partitions` mutex across a walk of every partition on the node, and other threads take that same lock through the get-or-create. collect under the guard, drop it, then store.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1244,47 +1261,69 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
.read(|inner| inner.revision)
}
-/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
-/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_partition_stats(
+/// Roll a torn-down partition out of its parent topic and stream by dropping
+/// its registry entry.
+///
+/// Usually a no-op: the metadata apply evicts the entry on the commit that
+/// acked the delete. What is left for this call is a stale incarnation being
+/// torn down after a slab-key reuse, whose entry the apply never named. It must
+/// not survive into the rebuild -- `StatsRegistry::partition` is a
+/// get-or-create, so the rebuild would inherit the dead incarnation's counters.
+/// (Its `purged_generation` no longer rides along either way: a fresh entry
+/// seeds that gate from the committed partition.)
+///
+/// The mounted partition's own handle is settled later, on `ConfirmRemove`.
+fn settle_partition_stats(ctx: &ReconcilerCtx, ns: IggyNamespace) {
Review Comment:
warning: on a successful teardown this eviction leaves the partition reporting the registry-miss shape, and the parents low by its whole size, until the rebuild re-folds the disk delta a pass later. document the window, or carry a materialisation signal.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -966,6 +977,12 @@ async fn tear_down_owned_partition(
partitions.tombstone(ns);
}
shards_table.remove(&ns);
+ // Registry entry only. The mounted partition's OWN counters are settled by
+ // the `ConfirmRemove` arm, which is the point where the partition value is
+ // dropped: a handler suspended mid-append resumes and increments through
+ // its cached handle, and anything settled before that drop leaves the
+ // increment in the parent totals with nothing left to roll it back.
+ settle_partition_stats(ctx, ns);
Review Comment:
warning: settle runs before the disk delete, so a failed delete returns with no `ConfirmRemove` and the still-mounted partition then reports one empty segment at offset 0 while its files remain. move it down next to the enqueue.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
Review Comment:
nit: this static is process-global, which the docs never say here. worth stating, since the throttle right below is deliberately split per pair to stop a noisy counter hiding a quiet one's first divergence.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -384,10 +389,32 @@ impl Stream {
/// read.
///
/// Shared across buffers and reader shards via `Arc` (a `StreamsInner` clone
-/// shares it). Only shard 0's writer mutates the maps, under the
-/// single-threaded Absorb; the `Mutex` is for `Sync` (uncontended), not
-/// concurrency. Ids are deterministic across replicas (same op order), so both
-/// buffers resolve the same key.
+/// shares it). Shard 0's writer owns the map mutations that ride the op-log,
+/// but it is NOT the only writer: [`Self::partition`] is a get-or-create the
+/// owning shard's reconciler calls when it materializes a namespace, and boot
+/// recovery calls it from every shard at once. The `Mutex` is load-bearing for
+/// that concurrency, not merely for `Sync`. Ids are deterministic across
+/// replicas (same op order), so both buffers resolve the same key.
+///
+/// Every eviction here therefore runs twice, once per buffer, and the second
+/// run is deferred. Two properties keep that from doing damage, and neither is
+/// guaranteed by the `Absorb` trait:
+///
+/// * `WriteCell::apply` is `append(cmd).publish()`, so a batch is one op. Batch
+/// two and a delete could be absorbed on the second buffer AFTER a create
+/// that re-minted the same id, and the delete would then evict the new
+/// partition's entry.
+/// * `left_right` drains the previous batch's `absorb_second` before the
Review Comment:
nit: the keyed evictions lean on this drain order, and `left-right = "0.11"` in the manifest floats any 0.11.x. pin `=0.11.8`, or add an `Absorb` impl that records the call sequence.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -7263,6 +7260,39 @@ pub struct SegmentRemoval {
pub budget_spent: bool,
}
+/// Roll one cleaned-up segment out of the partition counters, naming the
+/// partition when the rollback could not be covered.
+///
+/// Retention is the likeliest source of a clamped rollback: a partition on its
+/// way out keeps serving cleanup passes after the delete already settled its
+/// counters into the parents. The `stats_rollup_underflows` metric says a
+/// divergence happened and carries no ids; this says which partition and which
+/// segment, which is what an operator needs to act on it.
+fn settle_cleaned_segment(
+ stats: &PartitionStats,
+ namespace: IggyNamespace,
+ segment: &Segment,
+ messages_count: u64,
+) {
+ let size_shortfall = stats.decrement_size_bytes(segment.size.as_bytes_u64());
+ let segments_shortfall = stats.decrement_segments_count(1);
+ let messages_shortfall = stats.decrement_messages_count(messages_count);
+ if size_shortfall == 0 && segments_shortfall == 0 && messages_shortfall == 0 {
+ return;
+ }
+ warn!(
Review Comment:
nit: this warn fires per segment with no throttle, and `UnderflowSite::report` already logs the same event power-of-two throttled plus counts it. emit one line per `clean_*` call with the removed offset range.
##########
core/server/src/http/metrics.rs:
##########
@@ -45,6 +45,30 @@ pub(in crate::http) struct HttpMetrics {
pub(in crate::http) messages: Gauge,
pub(in crate::http) users: Gauge,
pub(in crate::http) clients: Gauge,
+ /// Not a legacy-parity metric: the running count of rollup decrements that
+ /// had to be clamped at zero. Non-zero means a topic or stream total was
+ /// asked to give back more than it held, so the aggregate it now reports is
+ /// low, and stays low until a rebuild or a restart.
+ ///
+ /// Not "alert on any increase". A delete, a purge, a partition teardown and
+ /// a snapshot restore each open a window where a retention pass hands back
+ /// bytes the parents have already given up, and the clamp is the intended
+ /// outcome there -- the `given_a_late_decrement_on_a_rolled_back_partition`
Review Comment:
nit: the test is `given_a_late_decrement_on_a_rolled_back_partition_should_leave_siblings_alone`, and `iggy_common::streaming_stats` isn't a reachable path since the module is `pub(crate)`. cite the full name plus the file.
##########
core/server/src/http/metrics.rs:
##########
@@ -45,6 +45,30 @@ pub(in crate::http) struct HttpMetrics {
pub(in crate::http) messages: Gauge,
pub(in crate::http) users: Gauge,
pub(in crate::http) clients: Gauge,
+ /// Not a legacy-parity metric: the running count of rollup decrements that
+ /// had to be clamped at zero. Non-zero means a topic or stream total was
Review Comment:
nit: says "a topic or stream total", but the three partition sites feed the same counter - including the example the doc cites, where the partition clamps and no parent does. say "partition, topic or stream".
##########
core/integration/tests/data_integrity/verify_after_server_restart.rs:
##########
@@ -520,3 +557,17 @@ async fn should_handle_resource_deletion_and_restart() {
);
}
}
+
+/// Payload for `should_handle_resource_deletion_and_restart`: small, fixed and
+/// non-empty, so a scope that gets deleted has bytes worth rolling back.
+fn deletion_test_messages() -> Vec<IggyMessage> {
+ (0..8u128)
+ .map(|offset| {
+ IggyMessage::builder()
+ .id(offset + 1)
+ .payload(bytes::Bytes::from_static(b"deletion-stats-payload"))
Review Comment:
nit: `Bytes` isn't in scope anywhere in this file and the rest of it imports at the top. add `use bytes::Bytes;` - this PR added the same import to `scenarios/mod.rs`.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+///
+/// Hidden from the docs: `iggy_common` re-exports this module at its root, and
+/// a server-only diagnostic has no business in the published SDK surface.
+#[doc(hidden)]
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+/// One scope/counter pair's clamp state: the labels an operator reads, plus the
+/// count the log throttle keys on.
+struct UnderflowSite {
+ scope: &'static str,
+ counter: &'static str,
+ clamped: AtomicU64,
+}
+
+impl UnderflowSite {
+ const fn new(scope: &'static str, counter: &'static str) -> Self {
+ Self {
+ scope,
+ counter,
+ clamped: AtomicU64::new(0),
+ }
+ }
+
+ /// Count the clamp globally, then log this pair's first and every power of
+ /// two after it.
+ ///
+ /// The throttle is per pair, not global: a skewed tree emits up to three
+ /// lines per counter per rollback and a bulk delete turns that into a
+ /// flood, so a shared count would hold a quiet pair's first-ever
+ /// divergence back for up to 1023 events behind a noisy one.
+ fn report(&self, shortfall: u64) {
+ ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed);
+ let clamped = self.clamped.fetch_add(1, Ordering::Relaxed) + 1;
+ if clamped.is_power_of_two() {
+ warn!(
Review Comment:
nit: no `target:` here, so this can't be filtered with the other diag streams. a scope-neutral target like `iggy.stats.diag` would do - the site logs stream and topic scopes too, so the partitions target is wrong.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -357,3 +492,115 @@ impl PartitionStats {
self.zero_out_current_offset();
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn tree() -> (Arc<StreamStats>, Arc<TopicStats>, Arc<PartitionStats>) {
+ let stream = Arc::new(StreamStats::default());
+ let topic = Arc::new(TopicStats::new(stream.clone()));
+ let partition = Arc::new(PartitionStats::new(topic.clone()));
+ (stream, topic, partition)
+ }
+
+ #[test]
+ fn given_a_parent_short_of_its_child_when_rolling_back_should_clamp_instead_of_wrapping() {
+ let (stream, topic, partition) = tree();
+ partition.increment_size_bytes(512);
+ // A snapshot restore stores absolute parent totals, so a parent can end
+ // up holding less than its children do.
+ topic.store_from_snapshot(0, 0, 0);
+ stream.store_from_snapshot(0, 0, 0);
+
+ partition.zero_out_all();
+
+ assert_eq!(topic.size_bytes_inconsistent(), 0);
+ assert_eq!(stream.size_bytes_inconsistent(), 0);
+ }
+
+ /// A deleted partition keeps its handle until the reconciler tears it down,
+ /// so a retention sweep can decrement counters the delete already rolled
+ /// back. Those bytes are not in the parents any more, and taking them again
+ /// would take a live sibling's instead.
+ #[test]
+ fn given_a_late_decrement_on_a_rolled_back_partition_should_leave_siblings_alone() {
+ let stream = Arc::new(StreamStats::default());
+ let topic = Arc::new(TopicStats::new(stream.clone()));
+ let survivor = Arc::new(PartitionStats::new(topic.clone()));
+ let deleted = Arc::new(PartitionStats::new(topic.clone()));
+ survivor.increment_size_bytes(488);
+ deleted.increment_size_bytes(512);
+
+ deleted.zero_out_all();
+ assert_eq!(topic.size_bytes_inconsistent(), 488);
+
+ // Retention retiring a segment of the partition that is on its way out.
+ deleted.decrement_size_bytes(100);
+
+ assert_eq!(
+ topic.size_bytes_inconsistent(),
+ 488,
+ "the survivor's bytes are the only ones left; they are not the deleted partition's"
+ );
+ assert_eq!(stream.size_bytes_inconsistent(), 488);
+ assert_eq!(survivor.size_bytes_inconsistent(), 488);
+ }
+
+ /// `delete_topic` settles the topic's residue on the stream, and the
+ /// reconciler later settles the same partition's own handle. The second
+ /// pass must find nothing left to take, or it takes it from another topic.
+ #[test]
+ fn given_a_topic_already_settled_when_its_partition_settles_should_leave_other_topics_alone() {
+ let stream = Arc::new(StreamStats::default());
+ let doomed_topic = Arc::new(TopicStats::new(stream.clone()));
+ let live_topic = Arc::new(TopicStats::new(stream.clone()));
+ let doomed_partition = Arc::new(PartitionStats::new(doomed_topic.clone()));
+ let live_partition = Arc::new(PartitionStats::new(live_topic.clone()));
+ live_partition.increment_size_bytes(900);
+ // Landed through the cached handle after the delete evicted its entry.
+ doomed_partition.increment_size_bytes(100);
+ assert_eq!(stream.size_bytes_inconsistent(), 1000);
+
+ doomed_topic.zero_out_all();
+ assert_eq!(stream.size_bytes_inconsistent(), 900);
+
+ doomed_partition.zero_out_all();
Review Comment:
nit: both widths expand from the same `clamped_sub!` body using `saturating_sub`, so nothing modulus-specific runs - say it covers the u32 wiring instead. same wording at `stream.rs:3480`.
##########
core/integration/tests/server/general.rs:
##########
@@ -88,6 +88,17 @@ async fn stream_size_validation(harness: &TestHarness) {
stream_size_validation_scenario::run(harness).await;
}
+#[iggy_harness(
Review Comment:
simplification: the scenario's own doc calls it transport-irrelevant and says it guards neither half, and every assertion goes through `get_topic`, `get_stream` or `get_stats`, all already covered on four transports. `[Tcp]` drops 9 server processes.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -1926,6 +1965,148 @@ mod tests {
);
}
+ /// A pass captures its targets once and then awaits disk work per
+ /// namespace, so a delete committed mid-pass leaves a stale target behind.
+ /// Resolving stats for it would get-or-CREATE the registry entry the delete
+ /// just evicted, and the apply on the second left-right buffer would then
+ /// zero and drop a live partition's counters.
+ #[compio::test]
+ async fn stats_are_refused_for_a_partition_the_committed_topic_no_longer_lists() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-a");
+ seed_topic(&mux, 2, 0, "topic-a", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ let listed = IggyNamespace::new(0, 0, 0);
+ assert!(
+ fetch_partition_build_inputs(&ctx, listed).is_some(),
+ "a partition the topic lists must still resolve"
+ );
+
+ let unlisted = IggyNamespace::new(0, 0, 7);
+ assert!(
+ fetch_partition_build_inputs(&ctx, unlisted).is_none(),
+ "a partition the committed topic does not list must not resolve"
+ );
+ assert!(
+ registry(&ctx).partition_get(0, 0, 7).is_none(),
+ "and the refusal must not have created its entry on the way"
+ );
+ }
+
+ /// The metadata apply rolls a deleted partition out of its parents at
+ /// commit, but the partition stays mounted until the reconciler tears it
+ /// down, and appends in that window land in parents with no entry left to
+ /// account for them. Teardown is where that residue gets settled.
+ #[compio::test]
+ async fn teardown_settles_the_residue_the_metadata_delete_could_not_reach() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-a");
+ seed_topic(&mux, 2, 0, "topic-a", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ reconcile_once(&ctx).await;
+ ctx.shard.apply_reconcile_ops();
+ let (stats, ..) =
+ fetch_partition_build_inputs(&ctx, ns).expect("materialised namespace has stats");
+ stats.increment_size_bytes(512);
+
+ seed_delete_topic(&ctx.shard.plane.metadata().mux_stm, 3, 0, 0);
+ // The apply evicted the entry and rolled back what it could see. This
+ // is the append that beat the teardown, through the handle the mounted
+ // partition still holds.
+ stats.increment_size_bytes(100);
+ assert_eq!(stream_size(&ctx), 100);
+
+ reconcile_once(&ctx).await;
+ ctx.shard.apply_reconcile_ops();
+
+ assert_eq!(
+ stream_size(&ctx),
+ 0,
+ "teardown must settle what landed after the commit that acked the delete"
Review Comment:
nit: the metadata apply already evicted the entry, so this passes on `zero_out_all()` at the drop point alone - delete `settle_partition_stats` and it stays green. the sibling test is the one guarding the settle, so rename this for the drop point.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -357,3 +492,115 @@ impl PartitionStats {
self.zero_out_current_offset();
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn tree() -> (Arc<StreamStats>, Arc<TopicStats>, Arc<PartitionStats>) {
+ let stream = Arc::new(StreamStats::default());
+ let topic = Arc::new(TopicStats::new(stream.clone()));
+ let partition = Arc::new(PartitionStats::new(topic.clone()));
+ (stream, topic, partition)
+ }
+
+ #[test]
+ fn given_a_parent_short_of_its_child_when_rolling_back_should_clamp_instead_of_wrapping() {
+ let (stream, topic, partition) = tree();
+ partition.increment_size_bytes(512);
+ // A snapshot restore stores absolute parent totals, so a parent can end
+ // up holding less than its children do.
+ topic.store_from_snapshot(0, 0, 0);
+ stream.store_from_snapshot(0, 0, 0);
+
+ partition.zero_out_all();
+
+ assert_eq!(topic.size_bytes_inconsistent(), 0);
+ assert_eq!(stream.size_bytes_inconsistent(), 0);
+ }
+
+ /// A deleted partition keeps its handle until the reconciler tears it down,
+ /// so a retention sweep can decrement counters the delete already rolled
+ /// back. Those bytes are not in the parents any more, and taking them again
+ /// would take a live sibling's instead.
+ #[test]
+ fn given_a_late_decrement_on_a_rolled_back_partition_should_leave_siblings_alone() {
Review Comment:
nit: no `when_` in this name, while the other four new tests here have one.
##########
core/integration/tests/server/scenarios/delete_stats_rollback_scenario.rs:
##########
@@ -0,0 +1,221 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Deleting data that is still counted must leave the parent totals correct.
+//!
+//! `stream_size_validation_scenario` covers delete too, but it purges each
+//! topic first, so every delete it performs removes an already-empty scope and
+//! cannot observe a rollback that never happened. These scenarios delete scopes
+//! that still hold messages, which is what the parent totals are wrong about.
+//!
+//! What this guards, and what it does not. Two mechanisms roll a deleted scope
+//! out of its parents: the metadata STM evicts the registry entries at commit
+//! and zeroes them into their parents, and the reconciler's partition teardown
+//! settles whatever landed after that. The STM half runs inside the apply that
+//! produces the ack, on counters shared across every shard and both left-right
+//! buffers, so it alone satisfies every assertion here. The reconciler runs as
+//! a separate task, but a delete commit wakes it (`signal_reconcile_wake`), so
+//! it too finishes before a client can complete another round trip.
+//!
+//! Measured, not reasoned: with only the STM rollback disabled this scenario
+//! PASSES (while the `metadata::stm::stream` unit tests fail), and with only
+//! the reconciler settle disabled it also passes. It fails only when BOTH are
+//! disabled. Either mechanism alone keeps a client's view correct, so this is a
+//! contract test for what a client observes and not a regression guard for
+//! either half.
+//!
+//! Each half is guarded by unit tests instead: the `metadata::stm::stream`
+//! module for the rollback and eviction, `server::partition_reconciler` for the
+//! teardown settle and the membership gate. The integration layer cannot hold
+//! the reconciler off by configuration -- the tick is capped at 30s, cannot be
+//! set to zero, and the commit wake bypasses it regardless.
+//!
+//! Verifying any of this needs `cargo build --bin iggy-server` first. The
+//! harness spawns the built binary, so a source-only mutation runs against the
+//! previous build and reports a green that means nothing.
+//!
+//! The reads after each delete are deliberately single-shot. Retrying until the
+//! numbers converge, the way the pre-delete assertions do for the async send
+//! folding, would also wait out a broken rollback and pass regardless.
+
+use crate::server::scenarios::{
+ PARTITIONS_COUNT, batch_size, create_client, create_messages, validate_stream,
Review Comment:
nit: the scenario never checks server-wide totals go back to zero after the deletes, which is the clearest statement of what this fixes. `assert_clean_system` reads no stats - add `validate_system_stats(client, 0, 0)` before it.
##########
core/server/src/http/metrics.rs:
##########
@@ -106,6 +137,17 @@ impl HttpMetrics {
self.http_requests.clone()
}
+ /// Advance the rollup-underflow counter to `total`, the process-wide count
Review Comment:
simplification: the get, `checked_sub`, `inc_by` dance exists only because `Counter` has no setter, and it guards a state the doc calls impossible. a ~12-line `Collector` with `ConstCounter` emits the same `_total` series - add a test for it, `PARITY_METRIC_NAMES` skips this one.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -7263,6 +7260,39 @@ pub struct SegmentRemoval {
pub budget_spent: bool,
}
+/// Roll one cleaned-up segment out of the partition counters, naming the
+/// partition when the rollback could not be covered.
+///
+/// Retention is the likeliest source of a clamped rollback: a partition on its
+/// way out keeps serving cleanup passes after the delete already settled its
+/// counters into the parents. The `stats_rollup_underflows` metric says a
+/// divergence happened and carries no ids; this says which partition and which
+/// segment, which is what an operator needs to act on it.
+fn settle_cleaned_segment(
Review Comment:
nit: `messages_count` is derived from the same `&segment` one line above the call. derive it inside and return it, then feed `removal.messages` from the return.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -42,18 +146,21 @@ impl StreamStats {
.fetch_add(segments_count, Ordering::AcqRel);
}
- pub fn decrement_size_bytes(&self, size_bytes: u64) {
- self.size_bytes.fetch_sub(size_bytes, Ordering::AcqRel);
+ /// Returns the shortfall -- how much of `size_bytes` this counter did not
+ /// hold -- so a caller that knows the ids can name where the divergence is.
+ /// Zero whenever the decrement was covered.
+ pub fn decrement_size_bytes(&self, size_bytes: u64) -> u64 {
+ size_bytes - clamped_sub_u64(&self.size_bytes, size_bytes, &STREAM_SIZE_BYTES)
}
- pub fn decrement_messages_count(&self, messages_count: u64) {
- self.messages_count
- .fetch_sub(messages_count, Ordering::AcqRel);
+ pub fn decrement_messages_count(&self, messages_count: u64) -> u64 {
Review Comment:
simplification: nothing reads the stream-level or topic-level shortfall anywhere - each level computes its own from `taken`. return `()` at both levels and keep the three on `PartitionStats`, which also cuts the public signature churn from nine to three.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+///
+/// Hidden from the docs: `iggy_common` re-exports this module at its root, and
+/// a server-only diagnostic has no business in the published SDK surface.
+#[doc(hidden)]
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+/// One scope/counter pair's clamp state: the labels an operator reads, plus the
+/// count the log throttle keys on.
+struct UnderflowSite {
+ scope: &'static str,
+ counter: &'static str,
+ clamped: AtomicU64,
+}
+
+impl UnderflowSite {
+ const fn new(scope: &'static str, counter: &'static str) -> Self {
+ Self {
+ scope,
+ counter,
+ clamped: AtomicU64::new(0),
+ }
+ }
+
+ /// Count the clamp globally, then log this pair's first and every power of
+ /// two after it.
+ ///
+ /// The throttle is per pair, not global: a skewed tree emits up to three
+ /// lines per counter per rollback and a bulk delete turns that into a
+ /// flood, so a shared count would hold a quiet pair's first-ever
+ /// divergence back for up to 1023 events behind a noisy one.
+ fn report(&self, shortfall: u64) {
+ ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed);
+ let clamped = self.clamped.fetch_add(1, Ordering::Relaxed) + 1;
+ if clamped.is_power_of_two() {
+ warn!(
+ scope = self.scope,
+ counter = self.counter,
+ shortfall,
+ clamped,
+ "rollup decrement exceeded the total it was subtracted from; clamped at zero"
+ );
+ }
+ }
+}
+
+static STREAM_SIZE_BYTES: UnderflowSite = UnderflowSite::new("stream", "size_bytes");
+static STREAM_MESSAGES_COUNT: UnderflowSite = UnderflowSite::new("stream", "messages_count");
+static STREAM_SEGMENTS_COUNT: UnderflowSite = UnderflowSite::new("stream", "segments_count");
Review Comment:
simplification: nine per-pair statics where three per-scope sites plus a `counter` argument would do, about 9 lines. costs you the property the doc right above defends - a noisy counter would then swallow a quiet one's first divergence.
##########
core/metadata/src/stm/stream.rs:
##########
@@ -2338,7 +2547,22 @@ impl Snapshotable for Streams {
// Boot: no live registry exists yet, so mint one. Safe because
// `new_from_empty` clones this single inner onto the other left-right
// buffer rather than building a second one.
- Ok(StreamsInner::inner_from_snapshot(snapshot, Arc::new(StatsRegistry::default())).into())
+ let inner = StreamsInner::inner_from_snapshot(snapshot, Arc::new(StatsRegistry::default()));
+ // Drop the checkpoint's aggregates here, BEFORE journal replay runs over
+ // this state machine. Boot rebuilds every counter from disk (each shard
+ // folds its `load_partition` deltas in), so the stored totals are never
+ // authoritative -- and a checkpoint reads a stream's total and its
+ // topics' as separate loads while the partition plane keeps counting, so
+ // they can disagree in either direction. Left in place, a replayed
+ // `DeleteTopic` over that torn shape rolls the topic back against a
+ // stream that never held it, clamps, and raises the rollup-underflow
+ // alarm on every boot with no real divergence behind it.
+ //
+ // The registry is empty here, so this stores zero at both levels; it is
+ // the same walk state transfer uses, which is why there is no second
+ // spelling of it.
+ inner.stats_registry.rebuild_parent_totals(&inner.items);
Review Comment:
simplification: this rebuild overwrites the two `store_from_snapshot` calls in `inner_from_snapshot` on every non-test path, and over a fresh registry it's storing 0 over 0. drop the stores and this call, about 25 lines - the tests at 4110 and 3573 go vacuous.
##########
core/common/src/types/streaming_stats.rs:
##########
@@ -19,6 +19,110 @@ use std::sync::{
Arc,
atomic::{AtomicU32, AtomicU64, Ordering},
};
+use tracing::warn;
+
+/// Number of rollup decrements that could not be covered by the counter they
+/// were subtracted from.
+///
+/// A decrement bigger than the total it targets means the tree lost
+/// `parent >= sum(children)` somewhere upstream. The counters are unsigned, so
+/// without a clamp that subtraction wraps to ~1.8e19 and every reader
+/// (`get_stream`, `get_topic`, `/stats`, `/metrics`) serves the wrapped value
+/// until the process restarts.
+///
+/// Clamping trades that for a total that is merely low, and low is where it
+/// stays: increments are `fetch_add`, so every later write stacks on the
+/// clamped base and carries the shortfall with it. Only a rebuild
+/// (`rebuild_parent_totals`) or a restart puts the total back. This counter
+/// plus the `warn!` below is what tells an operator the divergence happened,
+/// and that the aggregate needs a rebuild rather than time.
+static ROLLUP_UNDERFLOWS: AtomicU64 = AtomicU64::new(0);
+
+/// Monotonic count of clamped rollup decrements. Surfaced on `/metrics` so the
+/// clamp is alertable rather than log-grep-able.
+///
+/// Hidden from the docs: `iggy_common` re-exports this module at its root, and
+/// a server-only diagnostic has no business in the published SDK surface.
+#[doc(hidden)]
+#[must_use]
+pub fn rollup_underflows() -> u64 {
+ ROLLUP_UNDERFLOWS.load(Ordering::Relaxed)
+}
+
+/// One scope/counter pair's clamp state: the labels an operator reads, plus the
+/// count the log throttle keys on.
+struct UnderflowSite {
+ scope: &'static str,
+ counter: &'static str,
+ clamped: AtomicU64,
+}
+
+impl UnderflowSite {
+ const fn new(scope: &'static str, counter: &'static str) -> Self {
+ Self {
+ scope,
+ counter,
+ clamped: AtomicU64::new(0),
+ }
+ }
+
+ /// Count the clamp globally, then log this pair's first and every power of
+ /// two after it.
+ ///
+ /// The throttle is per pair, not global: a skewed tree emits up to three
+ /// lines per counter per rollback and a bulk delete turns that into a
+ /// flood, so a shared count would hold a quiet pair's first-ever
+ /// divergence back for up to 1023 events behind a noisy one.
+ fn report(&self, shortfall: u64) {
+ ROLLUP_UNDERFLOWS.fetch_add(1, Ordering::Relaxed);
+ let clamped = self.clamped.fetch_add(1, Ordering::Relaxed) + 1;
+ if clamped.is_power_of_two() {
+ warn!(
+ scope = self.scope,
+ counter = self.counter,
+ shortfall,
+ clamped,
+ "rollup decrement exceeded the total it was subtracted from; clamped at zero"
+ );
+ }
+ }
+}
+
+static STREAM_SIZE_BYTES: UnderflowSite = UnderflowSite::new("stream", "size_bytes");
+static STREAM_MESSAGES_COUNT: UnderflowSite = UnderflowSite::new("stream", "messages_count");
+static STREAM_SEGMENTS_COUNT: UnderflowSite = UnderflowSite::new("stream", "segments_count");
+static TOPIC_SIZE_BYTES: UnderflowSite = UnderflowSite::new("topic", "size_bytes");
+static TOPIC_MESSAGES_COUNT: UnderflowSite = UnderflowSite::new("topic", "messages_count");
+static TOPIC_SEGMENTS_COUNT: UnderflowSite = UnderflowSite::new("topic", "segments_count");
+static PARTITION_SIZE_BYTES: UnderflowSite = UnderflowSite::new("partition", "size_bytes");
+static PARTITION_MESSAGES_COUNT: UnderflowSite = UnderflowSite::new("partition", "messages_count");
+static PARTITION_SEGMENTS_COUNT: UnderflowSite = UnderflowSite::new("partition", "segments_count");
+
+/// Subtract `amount`, clamping at zero instead of wrapping. Returns what was
+/// actually taken, which is what the caller passes on to its parent.
+///
+/// `fetch_update` rather than a load followed by a subtract: the counters are
+/// written from the metadata shard and from whichever shard owns the partition,
+/// so a separate load leaves a window where the clamp reads one value and
+/// subtracts from another.
+macro_rules! clamped_sub {
+ ($name:ident, $counter:ty, $amount:ty) => {
+ fn $name(counter: &$counter, amount: $amount, site: &'static UnderflowSite) -> $amount {
+ let previous = counter
+ .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
+ Some(current.saturating_sub(amount))
Review Comment:
simplification: the closure always returns `Some`, so the `Err` arm can't happen. `(current != 0).then(|| current.saturating_sub(amount))` gives it a meaning and skips the CAS on an already-zero counter.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(metadata): roll deleted partitions and topics out of parent stats [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #4046:
URL: https://github.com/apache/iggy/pull/4046
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org