fix(tunnel): concurrent bounded liveness probes + failure hysteresis

Address two robustness defects in the TunnelManager self-heal path (#765):

- Probes were run serially inside UpdateTunnels; each dead-but-still-routed
  tunnel blackholes the SYN for the full probeDialTimeout, so N dead tunnels
  could stall tunnel creation for newly connected devices by up to
  probeDialTimeout*N. Probes now run concurrently, bounded by
  maxConcurrentProbes, so a batch of dead tunnels can't serialize into a
  multi-second stall.

- A single failed probe immediately tore down an otherwise healthy tunnel. A
  tunnel is now only rebuilt after probeFailureThreshold (2) consecutive
  failures, and the counter resets on any successful probe, so a momentary
  route hiccup or load spike no longer destroys a healthy tunnel.

Also prune unsupportedDevices and probeFailures for disconnected devices so
these maps stay bounded by the current device count instead of every udid
ever seen, and so a device that was unsupported, upgraded to a tunnel-capable
iOS, and reconnected is retried instead of skipped for the process lifetime.

Tests cover the two-failure threshold, transient-failure absorption, and
unsupported-on-disconnect pruning; all pass under -race.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J8eMENxJ1nec9CeHp4tjWk

Author: danielpaulus

ios/tunnel/tunnel_api.go

@@ -340,7 +340,12 @@ type TunnelManager struct {
 	unsupportedDevices map[string]bool
 	// lastProbe tracks when each tunnel was last liveness-probed so probes run
 	// at most once per probeInterval.
-	lastProbe            map[string]time.Time
+	lastProbe map[string]time.Time
+	// probeFailures counts consecutive failed liveness probes per udid. A tunnel
+	// is only torn down after probeFailureThreshold consecutive failures so a
+	// single transient probe error (a momentary route hiccup or load spike)
+	// doesn't destroy an otherwise healthy tunnel. A successful probe resets it.
+	probeFailures        map[string]int
 	probeInterval        time.Duration
 	startTunnelTimeout   time.Duration
 	firstUpdateCompleted bool
@@ -387,6 +392,7 @@ func newTunnelManager(pm PairRecordManager, userspaceTUN bool, udidFilter string
 		failedDevices:      map[string]failedDevice{},
 		unsupportedDevices: map[string]bool{},
 		lastProbe:          map[string]time.Time{},
+		probeFailures:      map[string]int{},
 		probeInterval:      defaultProbeInterval,
 		startTunnelTimeout: 10 * time.Second,
 		userspaceTUN:       userspaceTUN,
@@ -460,19 +466,18 @@ func (m *TunnelManager) UpdateTunnels(ctx context.Context) error {
 	// Liveness-probe existing tunnel records of still-connected devices: a
 	// tunnel can die without a usbmux disconnect (quick reboot, transport
 	// death), leaving a stale record that would otherwise be served forever. A
-	// failed probe tears the record down here so the create loop below rebuilds
-	// it in the same cycle. Records of vanished devices are handled by the
-	// disconnect teardown at the end.
+	// tunnel that fails probeFailureThreshold consecutive probes is torn down
+	// here so the create loop below rebuilds it in the same cycle. Records of
+	// vanished devices are handled by the disconnect teardown at the end.
+	//
+	// Probes run concurrently (bounded) because a dead-but-still-routed tunnel
+	// blackholes the SYN and burns the full probeDialTimeout; probing serially
+	// would let N dead tunnels delay tunnel creation for newly connected devices
+	// by up to probeDialTimeout*N.
 	now := time.Now()
-	for udid, tun := range localTunnels {
-		if !currentUDIDs[udid] || !m.shouldProbe(udid, now) {
-			continue
-		}
-		if err := m.pr.Probe(tun); err != nil {
-			golog.Warn("tunnel failed liveness probe, restarting it", "module", logModule, "udid", udid, "error", err)
-			_ = m.stopTunnel(tun)
-			delete(localTunnels, udid)
-		}
+	for udid := range m.probeTunnels(localTunnels, currentUDIDs, now) {
+		_ = m.stopTunnel(localTunnels[udid])
+		delete(localTunnels, udid)
 	}
 
 	for _, d := range devices.DeviceList {
@@ -541,6 +546,21 @@ func (m *TunnelManager) UpdateTunnels(ctx context.Context) error {
 			delete(m.failedDevices, udid)
 		}
 	}
+	// Prune per-device bookkeeping for devices that are no longer connected so
+	// these maps stay bounded by the current device count instead of growing
+	// with every udid ever seen. Pruning unsupportedDevices on disconnect also
+	// lets a device that was unsupported, then upgraded to a tunnel-capable iOS
+	// and reconnected, be retried instead of skipped for the rest of the process.
+	for udid := range m.unsupportedDevices {
+		if !currentUDIDs[udid] {
+			delete(m.unsupportedDevices, udid)
+		}
+	}
+	for udid := range m.probeFailures {
+		if !currentUDIDs[udid] {
+			delete(m.probeFailures, udid)
+		}
+	}
 	m.firstUpdateCompleted = true
 	m.mux.Unlock()
 	return nil
@@ -574,6 +594,61 @@ func (m *TunnelManager) shouldProbe(udid string, now time.Time) bool {
 	return true
 }
 
+// probeTunnels liveness-probes every due tunnel of a still-connected device and
+// returns the set of udids that should be torn down. Probes run concurrently
+// with at most maxConcurrentProbes in flight so a batch of dead-but-routed
+// tunnels (each of which blackholes the SYN for the full probeDialTimeout)
+// can't serialize into a multi-second stall of the update loop. A tunnel is only
+// reported dead after probeFailureThreshold consecutive failures; a success
+// resets its counter, so a single transient probe error is absorbed.
+func (m *TunnelManager) probeTunnels(localTunnels map[string]Tunnel, currentUDIDs map[string]bool, now time.Time) map[string]bool {
+	type probeResult struct {
+		udid string
+		err  error
+	}
+	results := make(chan probeResult)
+	sem := make(chan struct{}, maxConcurrentProbes)
+	var wg sync.WaitGroup
+	for udid, tun := range localTunnels {
+		if !currentUDIDs[udid] || !m.shouldProbe(udid, now) {
+			continue
+		}
+		wg.Add(1)
+		go func(udid string, tun Tunnel) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			results <- probeResult{udid: udid, err: m.pr.Probe(tun)}
+		}(udid, tun)
+	}
+	go func() {
+		wg.Wait()
+		close(results)
+	}()
+
+	dead := map[string]bool{}
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	for r := range results {
+		if r.err == nil {
+			delete(m.probeFailures, r.udid)
+			continue
+		}
+		m.probeFailures[r.udid]++
+		fails := m.probeFailures[r.udid]
+		if fails < probeFailureThreshold {
+			golog.Warn("tunnel failed liveness probe, will retry before restarting it",
+				"module", logModule, "udid", r.udid, "consecutiveFailures", fails, "threshold", probeFailureThreshold, "error", r.err)
+			continue
+		}
+		golog.Warn("tunnel failed liveness probe repeatedly, restarting it",
+			"module", logModule, "udid", r.udid, "consecutiveFailures", fails, "error", r.err)
+		delete(m.probeFailures, r.udid)
+		dead[r.udid] = true
+	}
+	return dead
+}
+
 // failedDeviceBackoff returns how long to wait before retrying a device after
 // failCount consecutive failures: 30s, 60s, 120s, 240s, capped at 5 minutes.
 func failedDeviceBackoff(failCount int) time.Duration {
@@ -598,6 +673,7 @@ func (m *TunnelManager) RemoveTunnel(ctx context.Context, serialNumber string) e
 		delete(m.tunnels, serialNumber)
 	}
 	delete(m.lastProbe, serialNumber)
+	delete(m.probeFailures, serialNumber)
 	m.mux.Unlock()
 
 	if !exists {
@@ -612,6 +688,7 @@ func (m *TunnelManager) stopTunnel(t Tunnel) error {
 	golog.Info("stopping tunnel", "module", logModule, "udid", t.Udid)
 	delete(m.tunnels, t.Udid)
 	delete(m.lastProbe, t.Udid)
+	delete(m.probeFailures, t.Udid)
 	m.mux.Unlock()
 
 	return t.Close()
@@ -673,15 +750,29 @@ type tunnelProber interface {
 }
 
 // defaultProbeInterval is how often the TunnelManager liveness-probes each
-// tunnel record. Combined with the probe's short dial timeout, a dead tunnel
-// is detected and rebuilt well within a minute instead of never.
+// tunnel record. Combined with the probe's short dial timeout and the
+// probeFailureThreshold consecutive-failure requirement, a persistently dead
+// tunnel is detected and rebuilt within roughly a minute instead of never,
+// while a single transient probe error never tears down a healthy tunnel.
 const defaultProbeInterval = 30 * time.Second
 
 // probeDialTimeout bounds the liveness probe's TCP connect. A healthy tunnel
 // answers in milliseconds; a dead-but-still-routed one swallows the SYN, so a
 // short timeout is enough to tell them apart without stalling UpdateTunnels.
 const probeDialTimeout = 5 * time.Second
 
+// maxConcurrentProbes caps how many liveness probes run at once. Probes are
+// concurrent so a batch of dead-but-routed tunnels can't serialize into a
+// probeDialTimeout*N stall, but bounded so a large fleet can't open an
+// unbounded number of dial sockets in one cycle.
+const maxConcurrentProbes = 8
+
+// probeFailureThreshold is the number of consecutive failed liveness probes
+// required before a tunnel is torn down and rebuilt. Requiring two failures
+// keeps a single transient probe error (a momentary route hiccup) from
+// destroying a healthy tunnel.
+const probeFailureThreshold = 2
+
 // rsdProber is the default tunnelProber: a short-timeout TCP dial of the
 // tunnel's RSD endpoint over the kernel TUN route.
 type rsdProber struct {

ios/tunnel/tunnel_manager_robustness_test.go

@@ -51,15 +51,24 @@ func (f *fakeStarter) callsFor(udid string) int {
 }
 
 type fakeProber struct {
+	mu     sync.Mutex
 	err    error
 	probed []string
 }
 
 func (f *fakeProber) Probe(t Tunnel) error {
+	f.mu.Lock()
 	f.probed = append(f.probed, t.Udid)
+	f.mu.Unlock()
 	return f.err
 }
 
+func (f *fakeProber) probedUDIDs() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.probed...)
+}
+
 // robustnessManager builds a fully faked TunnelManager that probes on every
 // UpdateTunnels cycle and resolves every device to iOS 18.0.0.
 func robustnessManager(ts tunnelStarter, pr tunnelProber, entries ...ios.DeviceEntry) *TunnelManager {
@@ -71,6 +80,7 @@ func robustnessManager(ts tunnelStarter, pr tunnelProber, entries ...ios.DeviceE
 		failedDevices:      map[string]failedDevice{},
 		unsupportedDevices: map[string]bool{},
 		lastProbe:          map[string]time.Time{},
+		probeFailures:      map[string]int{},
 		probeInterval:      time.Nanosecond,
 		startTunnelTimeout: time.Second,
 		productVersion: func(ios.DeviceEntry) (*semver.Version, error) {
@@ -79,20 +89,32 @@ func robustnessManager(ts tunnelStarter, pr tunnelProber, entries ...ios.DeviceE
 	}
 }
 
-// A tunnel record whose device is still connected but whose probe fails must be
-// torn down and rebuilt in the same update cycle — the agent-side self-heal
-// from issue #765.
+// A tunnel record whose device is still connected but whose probe keeps failing
+// must be torn down and rebuilt — the agent-side self-heal from issue #765.
+// Teardown requires probeFailureThreshold consecutive failures, so it happens on
+// the second failing cycle, not the first.
 func TestUpdateTunnelsRebuildsDeadTunnel(t *testing.T) {
 	starter := &fakeStarter{rsdPort: 4321}
 	prober := &fakeProber{err: errors.New("connection timed out")}
 	tm := robustnessManager(starter, prober, devEntry("dead-1", "USB"))
 	var closed int
 	tm.tunnels["dead-1"] = Tunnel{Udid: "dead-1", Address: "fd00::1", RsdPort: 1234, closer: func() error { closed++; return nil }}
 
+	// First failing probe: below the threshold, so the tunnel survives.
 	if err := tm.UpdateTunnels(context.Background()); err != nil {
-		t.Fatalf("UpdateTunnels: %v", err)
+		t.Fatalf("UpdateTunnels cycle 1: %v", err)
+	}
+	if closed != 0 {
+		t.Fatalf("after one failed probe closer called %d times, want 0 (needs %d consecutive)", closed, probeFailureThreshold)
+	}
+	if got := starter.callsFor("dead-1"); got != 0 {
+		t.Fatalf("after one failed probe tunnel restarted %d times, want 0", got)
 	}
 
+	// Second consecutive failing probe: threshold reached, tear down and rebuild.
+	if err := tm.UpdateTunnels(context.Background()); err != nil {
+		t.Fatalf("UpdateTunnels cycle 2: %v", err)
+	}
 	if closed != 1 {
 		t.Fatalf("dead tunnel closer called %d times, want 1", closed)
 	}
@@ -105,6 +127,38 @@ func TestUpdateTunnelsRebuildsDeadTunnel(t *testing.T) {
 	}
 }
 
+// A single transient probe failure followed by a success must NOT tear down the
+// tunnel: the consecutive-failure counter resets on success (issue #765 —
+// avoiding false-positive teardown of healthy tunnels under momentary load).
+func TestUpdateTunnelsAbsorbsTransientProbeFailure(t *testing.T) {
+	starter := &fakeStarter{rsdPort: 4321}
+	prober := &fakeProber{err: errors.New("temporary hiccup")}
+	tm := robustnessManager(starter, prober, devEntry("flap-1", "USB"))
+	var closed int
+	tm.tunnels["flap-1"] = Tunnel{Udid: "flap-1", Address: "fd00::1", RsdPort: 1234, closer: func() error { closed++; return nil }}
+
+	// One failing cycle (count=1), then the probe recovers.
+	if err := tm.UpdateTunnels(context.Background()); err != nil {
+		t.Fatalf("UpdateTunnels cycle 1: %v", err)
+	}
+	prober.mu.Lock()
+	prober.err = nil
+	prober.mu.Unlock()
+	if err := tm.UpdateTunnels(context.Background()); err != nil {
+		t.Fatalf("UpdateTunnels cycle 2: %v", err)
+	}
+
+	if closed != 0 {
+		t.Fatalf("tunnel torn down after a single transient failure, closer called %d times", closed)
+	}
+	if got := starter.callsFor("flap-1"); got != 0 {
+		t.Fatalf("tunnel restarted %d times after transient failure, want 0", got)
+	}
+	if n := tm.probeFailures["flap-1"]; n != 0 {
+		t.Fatalf("probeFailures[flap-1] = %d after a success, want 0 (counter must reset)", n)
+	}
+}
+
 // A healthy tunnel record must survive the probe untouched: no teardown, no
 // restart.
 func TestUpdateTunnelsKeepsHealthyTunnel(t *testing.T) {
@@ -118,8 +172,8 @@ func TestUpdateTunnelsKeepsHealthyTunnel(t *testing.T) {
 		t.Fatalf("UpdateTunnels: %v", err)
 	}
 
-	if len(prober.probed) != 1 || prober.probed[0] != "ok-1" {
-		t.Fatalf("probed = %v, want [ok-1]", prober.probed)
+	if probed := prober.probedUDIDs(); len(probed) != 1 || probed[0] != "ok-1" {
+		t.Fatalf("probed = %v, want [ok-1]", probed)
 	}
 	if closed != 0 {
 		t.Fatalf("healthy tunnel closer called %d times, want 0", closed)
@@ -144,8 +198,8 @@ func TestUpdateTunnelsDoesNotProbeDisconnectedDevice(t *testing.T) {
 		t.Fatalf("UpdateTunnels: %v", err)
 	}
 
-	if len(prober.probed) != 0 {
-		t.Fatalf("probed = %v, want none for a disconnected device", prober.probed)
+	if probed := prober.probedUDIDs(); len(probed) != 0 {
+		t.Fatalf("probed = %v, want none for a disconnected device", probed)
 	}
 	if _, ok := tm.tunnels["gone-1"]; ok {
 		t.Fatal("disconnected device's tunnel should have been torn down")
@@ -274,3 +328,64 @@ func TestUpdateTunnelsUdidFilter(t *testing.T) {
 		t.Fatalf("expected tunnels for both devices, got %v", tmAll.tunnels)
 	}
 }
+
+// mutableLister lets a test change the connected-device list between cycles.
+type mutableLister struct {
+	mu      sync.Mutex
+	entries []ios.DeviceEntry
+}
+
+func (l *mutableLister) ListDevices() (ios.DeviceList, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return ios.DeviceList{DeviceList: append([]ios.DeviceEntry(nil), l.entries...)}, nil
+}
+
+func (l *mutableLister) set(entries ...ios.DeviceEntry) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.entries = entries
+}
+
+// An unsupported classification must not survive a disconnect: once the device
+// leaves usbmux the entry is pruned, so reconnecting (e.g. after an iOS upgrade
+// to a tunnel-capable version) is retried instead of skipped forever. This also
+// bounds the unsupportedDevices map by the current device count rather than by
+// every udid ever seen (issue #523 growth follow-up).
+func TestUpdateTunnelsPrunesUnsupportedOnDisconnect(t *testing.T) {
+	lister := &mutableLister{}
+	lister.set(devEntry("up-1", "USB"))
+	starter := &fakeStarter{err: fmt.Errorf("manualPairingTunnelStart: %w 16.6.0", ErrUnsupportedVersion)}
+	tm := robustnessManager(starter, &fakeProber{})
+	tm.dl = lister
+
+	// Cycle 1: device is unsupported, gets marked and skipped.
+	if err := tm.UpdateTunnels(context.Background()); err != nil {
+		t.Fatalf("UpdateTunnels cycle 1: %v", err)
+	}
+	if !tm.unsupportedDevices["up-1"] {
+		t.Fatal("device should be marked unsupported after cycle 1")
+	}
+
+	// Cycle 2: device disconnects; the unsupported entry must be pruned.
+	lister.set()
+	if err := tm.UpdateTunnels(context.Background()); err != nil {
+		t.Fatalf("UpdateTunnels cycle 2: %v", err)
+	}
+	if tm.unsupportedDevices["up-1"] {
+		t.Fatal("unsupported entry must be pruned once the device disconnects")
+	}
+
+	// Cycle 3: device reconnects on a supported version; it must be retried.
+	lister.set(devEntry("up-1", "USB"))
+	starter.mu.Lock()
+	starter.err = nil
+	starter.rsdPort = 9999
+	starter.mu.Unlock()
+	if err := tm.UpdateTunnels(context.Background()); err != nil {
+		t.Fatalf("UpdateTunnels cycle 3: %v", err)
+	}
+	if _, ok := tm.tunnels["up-1"]; !ok {
+		t.Fatal("reconnected (now-supported) device should get a tunnel, not stay skipped")
+	}
+}