[dbsp] Probe multihost exchange connections when acks come slowly - #6793

This is claude's feedback, but I read the comments and the code and tend to agree with the first two:

1. The pong wait accepts any activity, not evidence of loss. wait_or_probe captures the oldest unacked message before the ping, then treats any activity notification as a pong. An ordinary ack also notifies. So an ack landing inside the pong window produces "retransmitting ... after a confirmed pong (connection is not half-open)" plus min_sequence = 0, even though the receiver just made progress. In production the trigger band is an ack taking between ping_interval and ping_interval + pong_timeout, so 1-2 s, which is ordinary for a receiver whose workers are consuming a large batch. That is a false alarm on the very diagnostic the PR exists to collect, and it resends the backlog into a busy receiver, the exact cost the ping was designed to avoid. Fix: after the pong, re-read channel.get(0) and warn plus retransmit only if the same sequence is still the oldest unacked.

2. New messages wait on the pong. During timeout(pong_timeout(), pong) nothing watches the channel, so a message queued in that window is delayed by up to 1 s in production. Fix: put the channel's nonempty waiter in the same select! as the pong.

Not sure if this is a real concern:

3. Unanswered pings never escalate. If the peer accepts bytes but nobody reads them (its socket alive, its reader gone), or a middlebox blackholes the flow, the write never errors and the sender pings once a second forever. My PR has the same hole with retransmits. Fix: after K consecutive unanswered pings, drop the connection and reconnect. take_over_connection exists precisely to make a self-initiated reconnect safe.

Also not sure if this is important:

4. Minor: pings bypass inject_fault. write_ping is a plain write_all, so the fault-injecting tests never exercise a failing ping write. One line to add at exchange.rs:522-style.

APPROVE — strong preference for this over #6784.

The reasoning is unusually careful and correct: a blind retransmit of a large backlog to a merely-busy receiver is a real cost; probing with a ~few-dozen-byte ping is nearly free; and the ping doubles as the same half-open-surfacing write attempt a blind retransmit would make. The PING_SEQUENCE = u64::MAX reservation is defensible (real sequences start at 0 and step by one) and cleanly separates the probe from delivery on the receiver side (peek_next_sequence avoids side effects on the counter).

The activity-notify-before-write ordering in wait_or_probe is the right race avoidance (let pong = activity.notified(); write_ping(...).await?;) — noted explicitly, good. The confirmed-pong-plus-still-unacknowledged retransmit path is honest about being defense-in-depth without a confirmed trigger, and the warn! log means we'll actually learn if it ever fires in production rather than guessing forever. That's exactly the right posture for hedging.

The deterministic wire-level test drives wait_or_probe's decision directly without needing to fake a half-open TCP socket, and the caveat in the test comment about what it does and doesn't cover is a model piece of test documentation. Nice touch that ping_interval()/pong_timeout() are #[cfg(test)]-shortened so the test doesn't spend seconds waiting.

Minor observations, not blockers:

• PING_SEQUENCE = u64::MAX colliding with a legitimate sequence would require ~10^19 messages on one channel — practically unreachable, but a debug_assert!(sequence < PING_SEQUENCE) when incrementing the sender's counter would make the invariant load-bearing rather than implicit. Purely defensive.
• Under a sustained genuinely-half-open condition, wait_or_probe will re-enter every ping_interval and issue another ping while the write to the dead socket eventually errors via run. That's the intended behavior, but worth confirming that ping-write errors in write_ping propagate through run_connection_tx's ? and reach run's reconnect path — from reading, they do (std::io::Result bubbles up through wait_or_probe -> run_connection_tx -> select -> run), just calling it out.
• warn! on the confirmed-pong retransmit is right, but if it turns out to fire often in the wild, the message could use the exchange_id displayed in a slightly more grep-friendly form (exchange={} sequence={} with structured fields, e.g. warn!(exchange_id = %message.exchange_id, sequence, "…")). Cosmetic.

Prefer this landing over #6784.

APPROVE — the simplification is the right response to ryzhyk's #1 and #2.

Dropping the pong-tracking machinery entirely is exactly the fix. With no pong, there is no "any activity ≠ evidence of loss" bug; and because the outer loop now uses timeout(ping_interval(), txq_grew), a newly-queued message is what races the timer, so nothing sits behind a pong wait. Both false-alarm cases from the previous review are gone.

The remaining design is small enough to hold in your head: if we have unacked data and no new message shows up within ping_interval(), write a ping frame; either the write errors and run reconnects, or it doesn't and we loop. The PING_SEQUENCE = u64::MAX reservation with a continue on the receiver keeps probes off the delivery path and the sequence counter, which is the important structural property.

Two small observations, non-blocking:

• new_ping uses channel.get(0) after has_unacknowledged() returned true, purely for metadata (exchange_id / sender). There is a tiny race where an ack lands in between and empties the queue; the if let Ok(...) handles it cleanly by skipping the ping. Worth a one-line comment above the if let so a future reader doesn't wonder whether the "impossible" Err branch matters.
• The two new tests are nicely deterministic and drive the wire protocol directly. new_message_sent_without_waiting_after_a_ping in particular is a good regression pin for #2 — a version that queued the pong-wait between txq notifications would fail it.

ryzhyk's #3 (unanswered pings never escalate) is theoretical and the retransmit-based alternative had the same hole, so punting is fine. #4 (pings bypass inject_fault) is cosmetic.

Prefer this landing over #6784.