[BUG]: etcd discovery watch stays silently stale when an etcd member hangs (no HTTP/2 keepalive on the etcd channel)

[devabhixda]

Describe the Bug

When the etcd member serving a discovery watch hangs without closing its TCP connection, the watch
stays open but receives nothing, forever. monitor_watch_stream only reconnects on a stream error or
EOF, and neither arrives, so the resync-after-reconnect path from #11146 never runs. Every instance
removed after the hang stays in the consumer's view.

For a frontend, this means it keeps routing to workers and routers that no longer exist. Requests to
them fail with connect timeouts, "connection refused" or "no route to host", and the frontend never
learns about their replacements. It logs no watch error, reconnect or resync, so nothing points at
etcd. It stays in this state until the process is restarted.

We hit this in production when the node hosting one member of a 3-member etcd cluster ran out of
memory and hung. Quorum survived and the other members kept serving. But the frontends whose
connection was pinned to the hung member went stale for hours, while frontends on the other members
were fine. Across that window, the stale frontends logged zero Error watching stream,
Watch stream unexpectedly closed or Reconnecting to ETCD lines.

Root cause: ClientOptions::default() builds ConnectOptions without with_keep_alive(...), so the
tonic channel sends no HTTP/2 PINGs and cannot detect a peer that stops responding. Clients in
Kubernetes typically dial a headless service (pod IPs, no VIP), so a hung member is exactly a live
TCP connection that never answers.

A second, smaller issue in the same loop: if new_watch_stream fails after its 10 s reconnect
deadline, the watch task hits Err(_) => return and exits permanently. Once keepalive makes the
failure detectable, a failure that lasts longer than 10 s would turn a stale watch into a dead one.

Steps to Reproduce

Standalone harness: a 3-member etcd cluster in Docker plus a ~50-line watcher on dynamo-runtime's
own etcd client (Client::new(ClientOptions::default()) + kv_get_and_watch_prefix), attached below.

1. docker compose up -d (etcd v3.5.21 x3).
2. Seed keys inst/w1..w5 and start the watcher. It prints its view every 2 s.
3. Find the member serving the watch (etcd_debugging_mvcc_watch_stream_total) and
   docker pause it. The process is frozen, and TCP to it stays established.
4. Through a healthy member, delete inst/w1 and inst/w2 and put inst/w6.
5. Wait 45 s, delete inst/w3, then unpause.

Expected Behavior

The watcher detects the unresponsive member within seconds, reconnects to a healthy member, resyncs,
and its view becomes w3 w4 w5 w6 (then w4 w5 w6) while the member is still frozen.

Actual Behavior

On main @ a76e12e65 the view stays w1 w2 w3 w4 w5 for the entire freeze. It keeps the deleted
keys and never sees the new one. Nothing is logged. It catches up only when the member is unpaused;
a member that never comes back (for example a dead node) leaves it stale indefinitely.

[   0.0s] view=5 {"inst/w1", "inst/w2", "inst/w3", "inst/w4", "inst/w5"}
=== 12:00:28 PAUSE etcd2
=== 12:00:32 truth now: w3 w4 w5 w6
      ... view unchanged, no log lines ...
=== 12:01:17 truth now: w4 w5 w6
=== 12:01:27 UNPAUSE etcd2
[  66.0s] view=3 {"inst/w4", "inst/w5", "inst/w6"}

With the proposed fix applied, the same run logs Error watching stream about 5-9 s after the freeze,
resyncs (4 keys), shows the correct view, and sees the later delete while the member is still frozen.
This also holds when etcd is compacted during the freeze, so recovery goes through the resync, not
revision replay.

Proposed fix

1. Enable HTTP/2 keepalive on the etcd channel in ClientOptions::default(): 10 s interval, 5 s
   timeout, env-overridable, with 0 disabling. The interval must stay at or above etcd's
   --grpc-keepalive-min-time (default 5 s) to avoid GOAWAY too_many_pings. A hung peer then fails
   the channel, the watch errors, and the existing reconnect + resync runs.
2. Replace Err(_) => return after new_watch_stream with a retry using the existing watch backoff,
   stopping only when the receiver is dropped or the runtime is cancelled.

A PR with this fix follows. Related: #11144 / #11146 (resync after reconnect, which this makes reachable) and
#7932 (lease keep-alive resilience, which is complementary and does not add channel keepalive).

Environment

• Dynamo main @ a76e12e6584f6029aeaa551c7cc8967671f67345 (lib/runtime/src/transports/etcd.rs)
• etcd v3.5.21 (quay.io/coreos/etcd), 3 members
• etcd-client 0.17 (ConnectOptions::with_keep_alive / keep_alive_while_idle are available)
• Observed in production on Linux / Kubernetes (etcd behind a headless service); the local repro above is on macOS + Docker Desktop

Additional Context

Harness below. Point dynamo-runtime in Cargo.toml at a Dynamo checkout, copy that checkout's Cargo.lock and .cargo/config.toml (for --cfg tokio_unstable), then docker compose up -d && cargo build && BIN=target/debug/etcd-watch-repro ./run.sh baseline 10. COMPACT=1 also compacts etcd during the freeze.

compose.yaml

x-etcd: &etcd
  image: quay.io/coreos/etcd:v3.5.21
services:
  etcd1:
    <<: *etcd
    command: etcd --name etcd1 --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://127.0.0.1:23791 --listen-peer-urls http://0.0.0.0:2380 --initial-advertise-peer-urls http://etcd1:2380 --initial-cluster etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 --initial-cluster-state new
    ports: ["23791:2379"]
  etcd2:
    <<: *etcd
    command: etcd --name etcd2 --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://127.0.0.1:23792 --listen-peer-urls http://0.0.0.0:2380 --initial-advertise-peer-urls http://etcd2:2380 --initial-cluster etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 --initial-cluster-state new
    ports: ["23792:2379"]
  etcd3:
    <<: *etcd
    command: etcd --name etcd3 --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://127.0.0.1:23793 --listen-peer-urls http://0.0.0.0:2380 --initial-advertise-peer-urls http://etcd3:2380 --initial-cluster etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 --initial-cluster-state new
    ports: ["23793:2379"]

src/main.rs

use std::collections::BTreeSet;
use std::time::Instant;

use dynamo_runtime::Runtime;
use dynamo_runtime::transports::etcd::{Client, ClientOptions, WatchEvent};

fn main() -> anyhow::Result<()> {
    dynamo_runtime::logging::init();
    let rt = Runtime::from_settings()?;
    let rt2 = rt.clone();
    rt.primary().block_on(async move {
        let opts = ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
        let client = Client::new(opts, rt2).await?;
        let (_, mut rx) = client.kv_get_and_watch_prefix("inst/").await?.dissolve();
        let start = Instant::now();
        let mut seen: BTreeSet<String> = BTreeSet::new();
        let mut tick = tokio::time::interval(std::time::Duration::from_secs(2));
        loop {
            tokio::select! {
                ev = rx.recv() => match ev {
                    Some(WatchEvent::Put(kv)) => { seen.insert(String::from_utf8_lossy(kv.key()).into()); }
                    Some(WatchEvent::Delete(kv)) => { seen.remove(&String::from_utf8_lossy(kv.key()).to_string()); }
                    Some(WatchEvent::Resync(kvs)) => {
                        seen = kvs.iter().map(|kv| String::from_utf8_lossy(kv.key()).into()).collect();
                        println!("[{:>6.1}s] RESYNC -> {} keys", start.elapsed().as_secs_f64(), seen.len());
                    }
                    None => { println!("[{:>6.1}s] WATCH CHANNEL CLOSED", start.elapsed().as_secs_f64()); break; }
                },
                _ = tick.tick() => println!("[{:>6.1}s] view={} {:?}", start.elapsed().as_secs_f64(), seen.len(), seen),
            }
        }
        anyhow::Ok(())
    })
}

run.sh

#!/usr/bin/env bash
# usage: run.sh <label> <keepalive_interval_secs>
set -u
cd "$(dirname "$0")"
LABEL=$1; KA=$2
BIN=${BIN:?set BIN to the built etcd-watch-repro binary}
ctl() { docker compose exec -T "$1" etcdctl --endpoints=http://127.0.0.1:2379 "${@:2}" >/dev/null; }
ts() { date +%H:%M:%S; }
docker compose unpause etcd1 etcd2 etcd3 >/dev/null 2>&1
ctl etcd2 del --prefix inst/
: > "out-$LABEL.log"
for i in 1 2 3 4 5; do ctl etcd2 put inst/w$i x; done
ETCD_ENDPOINTS=http://127.0.0.1:23791,http://127.0.0.1:23792,http://127.0.0.1:23793 \
ETCD_KEEPALIVE_INTERVAL_SECONDS=$KA DYN_LOG=info \
  "$BIN" >> "out-$LABEL.log" 2>&1 &
PID=$!
sleep 6
SERVING=""
for n in 1 2 3; do
  c=$(curl -s http://127.0.0.1:2379$n/metrics | awk '/^etcd_debugging_mvcc_watch_stream_total/{print $2}')
  echo "etcd$n watch_streams=$c"
  [ "${c%.*}" -ge 1 ] 2>/dev/null && SERVING=etcd$n
done
HEALTHY=$( [ "$SERVING" = etcd2 ] && echo etcd3 || echo etcd2 )
echo "$(ts) pausing $SERVING (writes via $HEALTHY)"; echo "=== $(ts) PAUSE $SERVING" >> "out-$LABEL.log"
docker compose pause "$SERVING" >/dev/null
sleep 3
ctl $HEALTHY del inst/w1; ctl $HEALTHY del inst/w2; ctl $HEALTHY put inst/w6 x
echo "=== $(ts) truth now: w3 w4 w5 w6" >> "out-$LABEL.log"
if [ "${COMPACT:-0}" = 1 ]; then
  REV=$(docker compose exec -T $HEALTHY etcdctl --endpoints=http://127.0.0.1:2379 get inst/ --prefix -w json | python3 -c "import json,sys;print(json.load(sys.stdin)['header']['revision'])")
  ctl $HEALTHY compact $REV; echo "=== $(ts) COMPACTED to $REV" >> "out-$LABEL.log"
fi
sleep 45
ctl $HEALTHY del inst/w3
echo "=== $(ts) truth now: w4 w5 w6" >> "out-$LABEL.log"
sleep 10
echo "=== $(ts) UNPAUSE $SERVING" >> "out-$LABEL.log"
docker compose unpause "$SERVING" >/dev/null
sleep 10
kill $PID

Cargo.toml

[package]
name = "etcd-watch-repro"
version = "0.1.0"
edition = "2024"

[dependencies]
dynamo-runtime = { path = "<path-to-dynamo>/lib/runtime" }
tokio = { version = "1", features = ["full"] }
anyhow = "1"

[workspace]