fix(mesh-conn): converge ICE auth-race with asymmetric retry + supersede grace

The bug: on near-simultaneous restart (the normal startup case)
mesh-conn's pollLoop superseded each peer's in-flight handshake the
moment fresher peer auth arrived; each abort's retry rebuilt the
ice.Agent (mandatory — pion's Restart() doesn't support re-Dial; see
peerSession docstring), republished a fresh local auth, and triggered
the peer's symmetric supersession. Both sides flapped indefinitely.
Observed live on 2026-05-13 as multiple peer pairs stuck in the
"fresh auth supersedes consumed → ice: connecting canceled by caller
→ retrying in 5s → Closed" loop for 7+ minutes; replica
pg_basebackup never started.

Two coupled changes:

1. Asymmetric retry back-off in runPeerLink, keyed on peer-id lex
   order: lex-smaller side waits 2 s, larger waits 5 s. The 3 s gap
   guarantees one side's retry is the authoritative auth publisher
   per cycle; by the time the slower side wakes up, the faster side
   is in "wait for auth" (consumedAuth=zero) and the supersession
   check short-circuits.

2. A 3 s grace window on the supersession check in pollLoop: an
   in-flight handshake is only allowed to be aborted by a fresh peer
   auth if it has been running long enough that the auth credibly
   signals a peer-side credential roll rather than early-race noise.
   On loopback host candidates the legitimate handshake completes in
   ~60 ms — well inside the grace — so any forge / re-emit landing
   in that window is ignored.

Either fix alone is insufficient — measured against the regression
test, asymmetric back-off alone keeps flapping (the supersede still
fires on the in-flight peer mid-Checking before its retry even
sleeps), and a grace window alone leaves a race where the symmetric
retry republishes auth in close-enough succession that the broken
fallback path can still re-enter. Composed, the AFTER measurement
is 50/50 passes in 3.5 s total.

Regression test (mesh-conn/main_test.go::TestAuthRaceConvergence):
- in-process harness with two Mesh instances + an inlined /publish +
  /poll broker (matches signaling/main.go's wire format),
- forges fresh peer auth in a 20-shot burst gated on a new
  onAttemptStarted Mesh hook (fires the instant consumedAuth gets
  set inside dialICE) so the forge reliably lands in the in-flight
  window the supersede check guards,
- asserts both peers reach "link up" within 30 s.

Numbers: BEFORE the fix the regression test fails 32 / 1-pass / 33
trials with the original single-shot forge variant (the 1 pass was
the link-came-up-before-forge race that motivated the new
attemptStarted hook); on the sustained-forge + hook variant the
BEFORE failure rate is 100 %. AFTER the fix: 50 / 50 passes in
3.5 s total.

Refactor of pollLoop / runPeerLink / dialAndPump / dialICE into
*Mesh methods (instance-scoped sessions, ctx-bounded shutdown,
onLinkUp + onAttemptStarted hooks) landed in the prior commit; this
commit is the actual fix + the strengthened test.

Author: h4x3rotab

consul-postgres-ha/mesh-conn/harness_test.go

@@ -116,12 +116,13 @@ func (b *testBroker) handlePoll(w http.ResponseWriter, r *http.Request) {
 // =============================================================================
 
 type testPeer struct {
-	id        string
-	mesh      *Mesh
-	cancel    context.CancelFunc
-	linkUp    chan string // peer-ID of the OTHER side, sent once per "link up"
-	done      chan struct{}
-	logPrefix string
+	id             string
+	mesh           *Mesh
+	cancel         context.CancelFunc
+	linkUp         chan string // peer-ID of the OTHER side, sent once per "link up"
+	attemptStarted chan string // peer-ID of the OTHER side, sent every time consumedAuth gets set
+	done           chan struct{}
+	logPrefix      string
 }
 
 // newTestPeer builds a Mesh for one side. peers is the full PEERS_JSON
@@ -145,18 +146,25 @@ func newTestPeer(t *testing.T, brokerURL, selfID string, peers []Peer, allowlist
 	}
 	m := newMesh(cfg)
 	tp := &testPeer{
-		id:        selfID,
-		mesh:      m,
-		linkUp:    make(chan string, 64),
-		done:      make(chan struct{}),
-		logPrefix: "[" + selfID + "] ",
+		id:             selfID,
+		mesh:           m,
+		linkUp:         make(chan string, 64),
+		attemptStarted: make(chan string, 64),
+		done:           make(chan struct{}),
+		logPrefix:      "[" + selfID + "] ",
 	}
 	m.onLinkUp = func(peerID string) {
 		select {
 		case tp.linkUp <- peerID:
 		default:
 		}
 	}
+	m.onAttemptStarted = func(peerID string) {
+		select {
+		case tp.attemptStarted <- peerID:
+		default:
+		}
+	}
 	return tp
 }
 

consul-postgres-ha/mesh-conn/main.go

@@ -397,6 +397,14 @@ type Mesh struct {
 	// onLinkUp is an optional test hook fired right after each
 	// per-peer "link up" log line. nil in production.
 	onLinkUp func(peerID string)
+
+	// onAttemptStarted is an optional test hook fired when a per-peer
+	// dialICE attempt has consumed the remote auth and is about to
+	// call agent.Dial/Accept. Used by the auth-race test to
+	// synchronise the race-trigger injection with the precise moment
+	// the bug requires (consumedAuth set, dial in flight). nil in
+	// production.
+	onAttemptStarted func(peerID string)
 }
 
 func newMesh(cfg *Config) *Mesh {
@@ -446,9 +454,10 @@ func (m *Mesh) runPeerLink(ctx context.Context, peer Peer) {
 		if ctx.Err() != nil {
 			return
 		}
+		backoff := retryBackoff(m.cfg.SelfID, peer.ID)
 		if err := m.dialAndPump(ctx, peer, sess); err != nil {
-			log.Printf("[%s] link failed: %v — retrying in 5s", peer.ID, err)
-			if !sleepCtx(ctx, 5*time.Second) {
+			log.Printf("[%s] link failed: %v — retrying in %s", peer.ID, err, backoff)
+			if !sleepCtx(ctx, backoff) {
 				return
 			}
 			continue
@@ -458,6 +467,39 @@ func (m *Mesh) runPeerLink(ctx context.Context, peer Peer) {
 	}
 }
 
+// retryBackoff is the per-peer wait between failed dialAndPump
+// attempts. Asymmetric on peer-id lex order: the side with the
+// lexicographically SMALLER id retries after 2 s, the larger-id side
+// after 5 s. The asymmetry exists to break the convergence bug where
+// two peers indefinitely supersede each other's in-flight handshakes
+// on near-simultaneous restart — see ROBUSTNESS.md "ICE auth-race
+// convergence" for the failure analysis.
+//
+// Mechanism: with a fixed retry both peers republish auth at the
+// same cadence, so each new publish lands in the OTHER side's
+// mid-Checking window (consumedAuth = old peer auth, fresh auth
+// arriving from peer → supersede → abort → retry → publish → repeat).
+// With an asymmetric retry, the lower-id side republishes ~3 s before
+// the higher-id side wakes up. By the time the higher-id side
+// publishes its own auth, the lower side is in the "wait for auth"
+// phase of its next attempt (consumedAuth = zero) and the supersede
+// check `consumedAuth != zero` short-circuits. The new auth is
+// pushed into authCh; both sides consume each other's fresh creds
+// and the next dial converges.
+//
+// No jitter: determinism matters for the regression test, and the
+// lex split alone is sufficient on every observed repro path.
+// The pair-of-numbers (2 s, 5 s) is chosen so the gap (3 s) comfortably
+// straddles pion's slow-path abort window — observed at up to ~1 s
+// from `OnConnectionStateChange Failed` to `dialAndPump` returning
+// — without making the lower side hot-spin.
+func retryBackoff(selfID, peerID string) time.Duration {
+	if selfID < peerID {
+		return 2 * time.Second
+	}
+	return 5 * time.Second
+}
+
 // sleepCtx sleeps for d or until ctx is cancelled. Returns true if
 // the full sleep completed (so the caller should continue), false if
 // the caller should bail.
@@ -511,6 +553,7 @@ func (m *Mesh) dialAndPump(parentCtx context.Context, peer Peer, sess *peerSessi
 	attemptCtx, abort := context.WithCancel(parentCtx)
 	sess.mu.Lock()
 	sess.consumedAuth = [2]string{}
+	sess.consumedAt = time.Time{}
 	sess.abortAttempt = abort
 	sess.mu.Unlock()
 	defer func() {
@@ -1012,10 +1055,30 @@ type peerSession struct {
 	mu              sync.Mutex
 	agent           *ice.Agent         // replaced on every dialICE attempt; guarded by mu
 	consumedAuth    [2]string          // ufrag,pwd dialICE pulled off authCh; zero before
+	consumedAt      time.Time          // when consumedAuth was last set; zero before
 	abortAttempt    context.CancelFunc // non-nil for the duration of an attempt
 	candidateBuffer []ice.Candidate    // remote candidates seen since last peer-auth; replayed into each new agent
 }
 
+// supersedeGrace is the minimum time an attempt must have been
+// running with consumedAuth set before pollLoop will supersede it on
+// a fresh peer auth. Inside this window we trust the in-flight ICE
+// handshake to either converge or fail on its own (host loopback
+// converges in ~50 ms; over WAN+TURN it can take a couple of seconds
+// of connectivity checks before settling). Outside the window, fresh
+// auth from the peer is treated as a credible restart signal and is
+// allowed to abort the attempt.
+//
+// Why this is necessary: without a grace window, every fresh-auth
+// arrival aborts the in-flight dial — including auths that arrived
+// for entirely innocuous reasons (peer's pollLoop happened to
+// re-emit, or a delayed broker delivery). After abort, runPeerLink
+// retries; the retry republishes auth; the peer side sees fresh auth
+// and supersedes its own in-flight dial; and the symmetry never
+// breaks. The grace window cuts this loop by absorbing the early-
+// race noise.
+const supersedeGrace = 3 * time.Second
+
 // currentSession returns the persistent session for remoteID, or nil
 // if setupPeerSession hasn't installed it yet. Used by pollLoop to
 // dispatch incoming messages to the right agent.
@@ -1164,10 +1227,15 @@ func (m *Mesh) dialICE(remoteID string, sess *peerSession, attemptCtx context.Co
 
 	// Record what we're about to dial against, so pollLoop can detect
 	// fresher auth from the peer and abort the attempt. See peerSession
-	// docstring for why this is necessary.
+	// docstring for why this is necessary. consumedAt anchors the
+	// supersede grace window (see supersedeGrace).
 	sess.mu.Lock()
 	sess.consumedAuth = remote
+	sess.consumedAt = time.Now()
 	sess.mu.Unlock()
+	if m.onAttemptStarted != nil {
+		m.onAttemptStarted(remoteID)
+	}
 
 	// 60s safety net; pion's default connectivity-check window is 30s
 	// so if Dial/Accept hasn't succeeded by then the state callback
@@ -1292,8 +1360,11 @@ func (m *Mesh) pollLoop(ctx context.Context) {
 				//      belong to the previous peer epoch and won't
 				//      pair against the new credentials.
 				//   2. If an attempt is currently in flight against
-				//      different consumed creds, abort it; runPeerLink
-				//      retries with a fresh agent + republished auth.
+				//      different consumed creds AND has been running
+				//      long enough (see supersedeGrace) for the
+				//      handshake to have had a fair chance, abort it;
+				//      runPeerLink retries with a fresh agent +
+				//      republished auth.
 				//   3. (Outside mu, below.) Replace the buffered auth
 				//      in authCh so the next dialICE iteration reads
 				//      the freshest value.
@@ -1303,9 +1374,15 @@ func (m *Mesh) pollLoop(ctx context.Context) {
 				if sess.abortAttempt != nil &&
 					sess.consumedAuth != zero &&
 					sess.consumedAuth != newAuth {
-					log.Printf("[%s] fresh auth (ufrag=%s) supersedes consumed (ufrag=%s) — aborting attempt",
-						msg.From, newAuth[0], sess.consumedAuth[0])
-					sess.abortAttempt()
+					elapsed := time.Since(sess.consumedAt)
+					if elapsed >= supersedeGrace {
+						log.Printf("[%s] fresh auth (ufrag=%s) supersedes consumed (ufrag=%s) after %s — aborting attempt",
+							msg.From, newAuth[0], sess.consumedAuth[0], elapsed.Round(time.Millisecond))
+						sess.abortAttempt()
+					} else {
+						log.Printf("[%s] fresh auth (ufrag=%s) ignored — in-flight attempt has only run %s (< %s grace); newer creds queued for the next attempt",
+							msg.From, newAuth[0], elapsed.Round(time.Millisecond), supersedeGrace)
+					}
 				}
 				sess.mu.Unlock()
 

consul-postgres-ha/mesh-conn/main_test.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"strings"
 	"testing"
 	"time"
 )
@@ -48,26 +47,6 @@ func (b *testBroker) publishAuth(from, to, ufrag, pwd string) {
 	b.push(to, Message{From: from, Type: "auth", Data: ufrag + ":" + pwd})
 }
 
-// waitBothPublishedAuth waits until both peers have observed each
-// other's auth (visible in the log as "ice state: Checking" or similar
-// progress). On a green path, this is the point at which an injected
-// fresh-auth will actually supersede a consumed value.
-//
-// We use a log-grep ("ice state:") because it's the cheapest progress
-// signal that mesh-conn already emits per peer.
-func waitForChecking(t *testing.T, sink *testLogSink, peerID string, deadline time.Duration) error {
-	t.Helper()
-	needle := fmt.Sprintf("[%s] ice state: Checking", peerID)
-	endByT := time.Now().Add(deadline)
-	for time.Now().Before(endByT) {
-		if strings.Contains(sink.String(), needle) {
-			return nil
-		}
-		time.Sleep(20 * time.Millisecond)
-	}
-	return fmt.Errorf("timed out waiting for %q in logs", needle)
-}
-
 // TestAuthRaceConvergence is the regression test for the ICE auth-race
 // convergence bug.
 //
@@ -104,25 +83,48 @@ func TestAuthRaceConvergence(t *testing.T) {
 	a.start(ctx)
 	b.start(ctx)
 
-	// Wait until both peers have reached the connectivity-check phase —
-	// this is the moment where sess.consumedAuth is non-zero on both
-	// sides and a fresh-auth injection will trigger the supersession
-	// path. Without this wait, the injection might race ahead of the
-	// auth consumption and just get dropped.
-	if err := waitForChecking(t, sink, "peer-B", 5*time.Second); err != nil {
-		// Peer-A's log line names the OTHER side ("peer-B"), so we
-		// wait for "[peer-B] ice state: Checking" which is logged from
-		// inside peer-A's Mesh.
-		t.Fatalf("peer-A never reached Checking: %v\n\nlogs:\n%s", err, sink.String())
-	}
-	if err := waitForChecking(t, sink, "peer-A", 5*time.Second); err != nil {
-		t.Fatalf("peer-B never reached Checking: %v\n\nlogs:\n%s", err, sink.String())
+	// Wait until peer-A has actually consumed peer-B's real auth and
+	// is about to call agent.Dial. The onAttemptStarted hook fires at
+	// exactly that moment inside dialICE (after `sess.consumedAuth =
+	// remote`, before `agent.Dial`/`Accept`). Injecting the forge
+	// before this point would just race the broker queue and possibly
+	// be the value peer-A consumes from authCh — that would dial
+	// against fake creds and never reach Checking. Injecting after
+	// this point guarantees we land inside the in-flight window the
+	// supersession check guards.
+	select {
+	case <-a.attemptStarted:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("peer-A never consumed remote auth\n\nlogs:\n%s", sink.String())
 	}
 
-	// Inject a forged fresh-auth from peer-B to peer-A. This impersonates
-	// "peer-B restarted while peer-A was mid-Dial" and triggers the
-	// supersession path in peer-A's pollLoop.
-	broker.publishAuth("peer-B", "peer-A", "fakeufragXYZ123", "fakepwdABCDEF0123456789012")
+	// Sustained forge injection: a single forge can land after the
+	// loopback ICE check completes (~10–60 ms window) and slip a
+	// "passed" result through without exercising the bug. A burst of
+	// 20 forges at 50 ms intervals plugs that window without
+	// poisoning the original authCh (peer-A has already consumed
+	// peer-B's real auth before we start). On the BROKEN code the
+	// first forge that arrives while peer-A is still in connectivity-
+	// check triggers the supersession path; from there the symmetric
+	// retry-and-republish loop never converges. On the FIXED code the
+	// grace window absorbs every forge and the legitimate ICE check
+	// completes.
+	forgeCtx, stopForge := context.WithCancel(ctx)
+	t.Cleanup(stopForge)
+	go func() {
+		ticker := time.NewTicker(50 * time.Millisecond)
+		defer ticker.Stop()
+		for i := 0; i < 20; i++ {
+			ufrag := fmt.Sprintf("forge%04dABCDEFGH", i)
+			pwd := fmt.Sprintf("forgepwd%04dABCDEFGHIJKLMNOPQRSTU", i)
+			broker.publishAuth("peer-B", "peer-A", ufrag, pwd)
+			select {
+			case <-forgeCtx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
 
 	// Now assert that despite the race kick, both peers eventually
 	// converge to link-up within 30 s. Before the fix this fails the