broker: make the committed high-water mark a value, not a droppable event (#264)

[allamiro]

Closes #264.

propagate_committed_hwm rode the bounded per-follower command queue on a try_send whose result was discarded:

let _ = follower.cmd_tx.try_send(FollowerCmd::PropagateHwm { .. });

A follower under load — precisely the case where its command channel backs up — silently stopped learning what had been acknowledged, and stayed stopped, because a dropped update was never retried.

Why this is a safety fix, not a reporting one

A follower's cluster_committed is a safety input. It is the bound that stops truncation from discarding acknowledged records:

if offset < high_watermark {
    return Err(BrokerError::TruncationBelowAcknowledged { .. });
}

A follower stuck on a stale mark permits a truncation it should refuse. That is the guard protecting this path: a leader acknowledges a record on a quorum of two and dies; the surviving pair yields a promotion boundary below that record; the replica still holding it reconciles it away (#263). The bound is what makes that replica refuse and drop out of the quorum instead — so promotion fails rather than acknowledged data being lost.

I originally attributed that protection to a new-epoch marker in #265. That was wrong — the marker sits at the boundary and both replicas hold through it, so it acks and changes nothing. This is the fix.

The change

The mark is a monotonic scalar and a follower only ever needs the newest one, so the queue was modelling a value as a stream of events. It is now a watch cell: send_replace cannot fail on a full channel because there is no queue to fill, and a burst collapses to the newest value instead of overflowing. The failure mode is removed by construction rather than made observable.

Also re-sends the current mark on every fresh session. changed() only fires on new values, so a follower that was down when the last update was published would otherwise sit on a stale mark until the next one happened to arrive — and on a quiet range that may be never.

No new test, deliberately

I wrote one, then ran it against the old code: it passed there too. The stall it used blocks the follower's server-side apply, which does not back up the leader's command queue at all — the condition the bug actually needs. A green test that cannot fail on the broken code is worse than no test, because it reads as coverage.

I could not construct an end-to-end test that discriminates without reaching into private driver internals. Reporting that rather than shipping the misleading one.

The existing replication suite passes unchanged, which is what this change should do: it alters delivery, not behaviour, everywhere the queue was not already full.

Workspace tests, clippy -D warnings, fmt clean.

---

Summary by cubic

Switch committed high-water mark propagation to a latest-wins value using tokio::sync::watch, eliminating dropped updates under load. Fixes a safety bug where a stale follower mark could allow truncation below acknowledged records.

• Bug Fixes
  • Replace queued PropagateHwm over mpsc with a watch::channel and send_replace, so bursts collapse to the newest value and updates can’t be dropped.
  • Re-send the current mark on new sessions so followers that were down don’t stay stale if no new updates arrive (changed() only fires on new values).

^{Written for commit d953af1. Summary will update on new commits.}

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.
To continue using code reviews, you can upgrade your account or add credits to your account and enable them for code reviews in your settings.

[cursor]

Bugbot is not enabled for your account, so this pull request was not reviewed.

Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs.

[cubic-dev-ai]

2 issues found across 1 file

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="crates/vtop-broker/src/replication/network.rs">

<violation number="1" location="crates/vtop-broker/src/replication/network.rs:870">
P1: A reconnecting follower can remain permanently below the committed HWM: this block sends the mark before replaying pending appends, and follower-side `observe_hwm` clamps it to current local durability while no HWM is sent after replay completes. Reapply the current mark after catch-up (or otherwise trigger HWM observation after the replay) so a quiet range cannot retain a stale truncation guard.</violation>

<violation number="2" location="crates/vtop-broker/src/replication/network.rs:874">
P3: The new high-water-mark send path is built twice with essentially identical logic: the fresh-session re-send block and the `changed()` arm both clone the current value via a scoped `borrow_and_update()` and then construct/write the same `CommittedHwmUpdate` frame (request_id 0, stream_id 0), differing only in error handling. Extracting a small helper (e.g. `write_hwm(&mut stream, &CommittedHwmUpdate) -> Result<(), SessionOutcome>`) and a `current_hwm()` accessor would remove the duplication and keep the two paths from drifting if the frame layout or error handling changes later. This is a low-severity maintainability nit; the safety logic itself is correct.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[cubic-dev-ai]

P1: A reconnecting follower can remain permanently below the committed HWM: this block sends the mark before replaying pending appends, and follower-side observe_hwm clamps it to current local durability while no HWM is sent after replay completes. Reapply the current mark after catch-up (or otherwise trigger HWM observation after the replay) so a quiet range cannot retain a stale truncation guard.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At crates/vtop-broker/src/replication/network.rs, line 870:

<comment>A reconnecting follower can remain permanently below the committed HWM: this block sends the mark before replaying pending appends, and follower-side `observe_hwm` clamps it to current local durability while no HWM is sent after replay completes. Reapply the current mark after catch-up (or otherwise trigger HWM observation after the replay) so a quiet range cannot retain a stale truncation guard.</comment>

<file context>
@@ -841,6 +855,32 @@ impl FollowerDriver {
+            let guard = self.hwm_rx.borrow_and_update();
+            guard.clone()
+        };
+        if let Some(update) = current_hwm {
+            let frame = WireFrame {
+                request_id: 0,
</file context>

[cubic-dev-ai]

P3: The new high-water-mark send path is built twice with essentially identical logic: the fresh-session re-send block and the changed() arm both clone the current value via a scoped borrow_and_update() and then construct/write the same CommittedHwmUpdate frame (request_id 0, stream_id 0), differing only in error handling. Extracting a small helper (e.g. write_hwm(&mut stream, &CommittedHwmUpdate) -> Result<(), SessionOutcome>) and a current_hwm() accessor would remove the duplication and keep the two paths from drifting if the frame layout or error handling changes later. This is a low-severity maintainability nit; the safety logic itself is correct.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At crates/vtop-broker/src/replication/network.rs, line 874:

<comment>The new high-water-mark send path is built twice with essentially identical logic: the fresh-session re-send block and the `changed()` arm both clone the current value via a scoped `borrow_and_update()` and then construct/write the same `CommittedHwmUpdate` frame (request_id 0, stream_id 0), differing only in error handling. Extracting a small helper (e.g. `write_hwm(&mut stream, &CommittedHwmUpdate) -> Result<(), SessionOutcome>`) and a `current_hwm()` accessor would remove the duplication and keep the two paths from drifting if the frame layout or error handling changes later. This is a low-severity maintainability nit; the safety logic itself is correct.</comment>

<file context>
@@ -841,6 +855,32 @@ impl FollowerDriver {
+            let frame = WireFrame {
+                request_id: 0,
+                stream_id: 0,
+                message: Message::CommittedHwmUpdate(update),
+            };
+            if write_frame(&mut stream, &frame, REPLICA_LIMITS)
</file context>