feat: event driven range load tracking

.claude/rules/metadata-state-machine.md

@@ -67,16 +67,16 @@ MetadataStateMachine (one per shard group)
 
 15. **Merges require adjacency.** Only ranges with contiguous keyspaces (`r1.keyspace_end == r2.keyspace_start` or vice versa) can merge. Otherwise the merged range's keyspace would not be a contiguous slice.
 
-16. **Seal history is deterministic.** `seal_history` on `RangeMeta` is updated only inside `apply_roll_segment()`, which runs on ALL replicas with the same input. No leader-only state leaks into `RangeSealHistory`. The split/merge *decision* is leader-only, but the data the decision reads is identical on every replica.
+16. **Range load classification is deterministic.** `load_state` on `RangeMeta` changes only when `apply_roll_segment()` successfully rolls the current active segment, which runs on ALL replicas with the same committed `SegmentRollIntent`. Data-pressure rolls advance the capped consecutive-pressure streak; idle-maintenance rolls reset it to Idle; replication-failure and recovery rolls leave it unchanged. Boundary correction, idempotent replay, and range-level split/merge seals never classify load. The split/merge *decision* is leader-only, but the data the decision reads is identical on every replica.
 
-17. **Children of a split carry a cooldown anchor.** Child ranges created by `apply_split_range()` always have `seal_history.created_by_split_at` set to the split's `created_at` timestamp. Enforces `SPLIT_COOLDOWN_MS` before children can be re-split — prevents split storms from transient load.
+17. **Pressure streaks are bounded and nonzero.** `RangeLoadState::Pressure.consecutive_rolls` is always in `1..=SPLIT_SEAL_THRESHOLD`. The pressure counter stops growing once it hits the split limit, preventing it from increasing endlessly if a split is delayed. Newly created ranges (from splits or merges) start as 'Unclassified'. We don't enforce this starting state as a strict invariant, since normal segment rolls will soon reclassify them anyway.
 
 18. **`correct_end_offset` is write-once.** Updates the sealed segment's `end_offset` only when the current value is `None` AND the incoming `end_entry_id` is `Some`. Simultaneously sets the active segment's `start_offset` to `end_entry_id + 1`. Re-application is a no-op. Restores invariant 8 after a death-triggered roll. This is now the **fallback** correction — for the follower-death race (the live leader's write-path seal arrives after a reconcile `None`-roll) and the unknown-end fallback. The leader-crash path no longer relies on it: it recovers the committed end *before* rolling (`raft-actor.md` #6), so the successor opens at `min+1` directly rather than at 0-then-corrected.
 
-19. **`RollSegment` is idempotent.** Re-applying a `RollSegment` for a segment that is no longer active returns `ApplyResult::Noop` rather than an error. Tolerates the race where both the write-path timeout and SWIM node death fire `RollSegment` for the same failure.
+19. **`RollSegment` is idempotent.** Re-applying a `RollSegment` for a segment that is no longer active succeeds without raising a metadata event rather than returning an error. Tolerates the race where both the write-path timeout and SWIM node death fire `RollSegment` for the same failure.
 
 20. **An entity's state never reverts.** Topic: Active → Sealed → Deleted. Range: Active → Sealed → Deleting. Segment: Active → Sealed → Deleting. No backward transitions. Once a segment is Sealed it never becomes Active again — `ReassignSegment` swaps its `replica_set` but keeps it `Sealed`.
 
-21. **`ReassignSegment` only re-points a sealed segment.** `apply_reassign_segment()` accepts only a `Sealed` segment, swaps `replica_set`, and changes nothing else — state stays `Sealed`; data, offsets, lineage, and timestamps stay frozen (invariant 3). An active, deleting, or unknown segment is rejected (`SegmentNotSealed` / `SegmentNotFound`), logged but not fatal (invariant 11). Re-applying with the same `replica_set` returns `ApplyResult::Noop`, tolerating duplicate death detection and no-leader re-proposals (cf. invariant 19). The swap runs through `apply`, so the umbrella `assert_invariants` re-checks every other invariant afterward — a reassignment cannot leave the machine inconsistent.
+21. **`ReassignSegment` only re-points a sealed segment.** `apply_reassign_segment()` accepts only a `Sealed` segment, swaps `replica_set`, and changes nothing else — state stays `Sealed`; data, offsets, lineage, and timestamps stay frozen (invariant 3). An active, deleting, or unknown segment is rejected (`SegmentNotSealed` / `SegmentNotFound`), logged but not fatal (invariant 11). Re-applying with the same `replica_set` succeeds without raising a metadata event, tolerating duplicate death detection and no-leader re-proposals (cf. invariant 19). The swap runs through `apply`, so the umbrella `assert_invariants` re-checks every other invariant afterward — a reassignment cannot leave the machine inconsistent.
 
 22. **A committed consumer-group generation assigns each active range exactly once.** When a group has members, its assignment keys exactly equal the topic's active ranges and every assignment names a current member. When it has no members, it has no assignments. Membership or range-topology changes advance the generation and recompute the full desired assignment through the Raft log; heartbeat refreshes that do not change membership leave the generation unchanged.

.claude/rules/raft-actor.md

@@ -130,9 +130,9 @@ RaftTransportActor              (TCP I/O)
 
 2. **On `LeaderChange → self`, the new leader reconciles against SWIM.** Actor snapshots SWIM's live member set and proposes `RemovePeer` for every current peer not in that set. Without this, SWIM facts about deaths that occurred while no leader existed (typically: the previous leader was one of the dead) never become Raft truth — the events have already been disseminated by SWIM and won't re-fire.
 
-3. **`MetadataCommitted` is dispatched only by the current leader.** During event drain, if the replica is no longer leader, its `MetadataCommitted` events are discarded (the entry still applies — only the outbound dispatch is suppressed). Otherwise every replica would emit duplicate `SegmentAssignment` / `SealResponse` to the data plane, and a former leader would dispatch entries it can no longer be authoritative about.
+3. **`MetadataCommitted` is dispatched only by the current leader.** During event drain, if the replica is no longer leader, its `MetadataCommitted` events are discarded (the entry still applies — only the outbound dispatch is suppressed). Otherwise every replica would emit duplicate `SegmentAssignment` / `SegmentRollCommitted` to the data plane, and a former leader would dispatch entries it can no longer be authoritative about.
 
-4. **`pending_seals` is bounded.** Cleared in two places: (a) on observed `LogMutation::TruncateFrom(idx)`, entries with `log_index ≥ idx` are dropped — the new leader may have truncated those proposals; (b) when `!raft.is_leader()` at drain time, all entries for that group are dropped — the `SealResponse` will not come from this replica. Without this, unbounded growth as leaders churn.
+4. **`pending_rolls` is bounded.** Cleared in two places: (a) on observed `LogMutation::TruncateFrom(idx)`, entries with `log_index ≥ idx` are dropped — the new leader may have truncated those proposals; (b) when `!raft.is_leader()` at drain time, all entries for that group are dropped — the `SegmentRollCommitted` will not come from this replica. Without this, unbounded growth as leaders churn.
 
 5. **`flush` terminates within a bounded number of rounds.** `MultiRaftActor::flush` loops `store.flush()` up to `MAX_FLUSH_ROUNDS = 8` because `route_event` may call back into the store (reconciliation proposes new `RemovePeer` entries). Bounded so a feedback bug cannot deadlock the actor. Steady-state shape is ≤ 2 rounds.
 
@@ -146,7 +146,7 @@ RaftTransportActor              (TCP I/O)
 
 7. **The recovered segment is led by the most-complete survivor.** The new `replica_set[0]` is the survivor with the highest durable extent (`argmax`, ties broken by lowest `NodeId`) — a clean "most-caught-up replica wins" election, read from the same gather. The choice is the leader's and is committed via `RollSegment.new_replica_set` ordering, so apply stays deterministic (a leader-only *decision*, identical *apply* on every replica — cf. `metadata-state-machine.md` #16).
 
-8. **Boundary recovery reads durable extents, never the notified commit cursor.** A `SealBoundaryReport` carries `next_entry_id - 1` (`SegmentTracker::durable_end_entry_id`), advanced only after `wal.flush_batch()` succeeds — distinct from and `≥` the *notified* `committed_entry_id`. A survivor's notified commit lags its durable extent (the leader acks the producer before propagating the commit notification), so trusting the cursor could drop an already-acked entry. `None` means the survivor holds nothing ⇒ nothing was committed ⇒ the segment seals empty.
+8. **Boundary recovery reads durable extents, never the notified commit cursor.** A `DurableSegmentEndReported` carries `next_entry_id - 1` (`SegmentTracker::durable_end_entry_id`), advanced only after `wal.flush_batch()` succeeds — distinct from and `≥` the *notified* `committed_entry_id`. A survivor's notified commit lags its durable extent (the leader acks the producer before propagating the commit notification), so trusting the cursor could drop an already-acked entry. `None` means the survivor holds nothing ⇒ nothing was committed ⇒ the segment seals empty.
 
 9. **Sealed-segment catch-up is re-driven until confirmed.** The `CatchUpAssignment` a committed `ReassignSegment` dispatches (via the #3 drain) is one-shot and fire-and-forget. So on applying the reassignment the leader seeds a per-segment in-repair entry (the `CatchUpRepairs` tracker, leader-only — gated on role at apply like `peer_states` in `apply_add_peer`) and re-drives `CatchUpAssignment` to every not-yet-confirmed member on each heartbeat sweep (`maybe_redrive_catch_ups`, beside the active-segment `maybe_redrive_segment_assignments`). A member clears itself with a `CatchUpAck` once it holds the segment through `sealed_end`; the entry is pruned when all members confirm, superseded when the segment is re-reassigned, and dropped on step-down (leader-volatile, cleared with `confirmed_assignment`). `CatchUpAck` is idempotent and a full-match member re-confirms on re-drive (`data_plane/state.rs`), so no single dropped message strands a repair. A re-drive also rescues an *in-flight* receive that has stalled (dead source or dropped chunk): the receiver re-requests the remainder from a freshly-picked source rather than no-op'ing on the duplicate assignment, and its chunk handler appends only the strictly-next entry so a resume overlap can't inflate the verified prefix past `sealed_end`. On takeover the new leader re-seeds the tracker for every known-end sealed segment it leads (`reseed_catch_up`, from `reconcile_on_leadership_change`), since the tracker is leader-volatile — so a repair in flight at a leadership change is re-driven, not stranded (already-complete members full-match-ack cheaply; the re-drive of all sealed segments is the cost). This is a *rule*, not an invariant (it is a liveness guarantee over time, not a snapshot predicate): no `assert_invariants` entry.