[Rust] Recovery retry budget resets on every reconnect, allowing unbounded reconnection

[zlata-stefanovic-db]

Summary

The supervisor rebuilds a fresh recovery retry budget on every loop iteration, so recovery_retries only bounds consecutive create-stream failures within a single recovery episode. A server that repeatedly accepts a stream and then immediately drops it (RST/EOF with no acks) causes the SDK to reconnect indefinitely — the retry budget is never consumed, because each create call succeeds.

Location

rust/sdk/src/stream/grpc/supervisor.rs:57 (loop start), :78-88 (per-iteration budget), :149 (RetryIf::spawn), :307 (loop back-edge)

loop {
    // ...
    let strategy = FixedInterval::from_millis(options.recovery_backoff_ms)
        .take(options.recovery_retries as usize);   // rebuilt every iteration
    let attempt = AtomicUsize::new(0);              // reset every iteration
    // ...
    let creation = RetryIf::spawn(strategy, create_attempt, should_retry).await;
    // stream runs; on mid-stream failure the loop returns to the top with a brand-new budget
}

RetryIf only bounds the create-stream retries inside one episode. Once a stream is established and later fails, the loop re-enters and gets a full fresh budget.

Why it matters

Against a degenerate server (accept then immediate drop), the SDK spins reconnecting forever and never surfaces a terminal error to the caller, so unacked records are never returned and the failure is never reported. There is also no notion of a connection-uptime-based budget reset.

Suggested fix

Track a lifetime consecutive-failure counter that survives across episodes, break after recovery_retries cycles without progress, and reset it only on durable progress or on a connection that stays healthy for some minimum uptime.

Reference: how the pure-Go core handles this

The pure-Go reimplementation uses a single lifetime counter and an uptime/progress-gated reset:

purego/internal/stream/supervisor.go:36,46,130 — failedAttempts persists across reconnects; the loop breaks when failedAttempts > cs.cfg.RecoveryRetries.

purego/internal/stream/core.go:829-830 — the budget resets only on durable progress or RecoveryResetAfter (default 15s) of stable uptime:

resetRecoveryBudget = cs.wm.current() > startAck ||
    time.Since(openedAt) >= cs.cfg.RecoveryResetAfter