daemon: address watcher follow-ups — own retry backoff, Stop not held by a redial, no duplicate heartbeat recovery

CHANGELOG.md

@@ -386,6 +386,13 @@ Detailed per-release notes are on the
   the oldest of them. After a parked segment was resent the timeout went from
   437 ms to 4.5 s. Following Karn's algorithm as Linux applies it, an ACK that
   newly acknowledges any resent segment gives no round-trip sample.
+- **Address-change recovery follow-ups:** registry retries no longer delay
+  the next address move (it waited up to 80 s, now 10 s); stopping the
+  daemon no longer waits up to 47 s for a recovery redialling a registry
+  that refuses connections; and the heartbeat no longer repeats the registry
+  reconnect and full re-registration a recovery has just done. A recovery
+  whose re-registration failed, or that restored the node's visibility,
+  hostname or trust pairs only in part, does not count as done.
 - **A bulk transfer no longer hangs when the receiver's application falls
   behind.** When a receiver's application stops reading for a second, the
   receiver parks the next in-order segment in its reorder buffer and keeps

pkg/daemon/addrwatch.go

@@ -72,10 +72,12 @@ import (
 // addrObservedCooldownMax for a beacon-observed one, and drops back after a
 // quiet period. A flapping interface costs a handful of re-registrations and
 // then one every two minutes; a NAT whose public IP keeps flapping costs one
-// an hour. A change seen during the cooldown is not lost: it runs when the
-// cooldown ends, unless the address has gone back to the one already
-// announced. The gaps are measured on the wall clock (addrWatchNow), so time
-// the host spends asleep counts.
+// an hour. The registry-only retries (addrRegistryRetryMax) have a backoff of
+// their own, on the local-address schedule and started over by every move,
+// so they never hold the next move back. A change seen during the cooldown
+// is not lost: it runs when the cooldown ends, unless the address has gone
+// back to the one already announced. The gaps are measured on the wall clock
+// (addrWatchNow), so time the host spends asleep counts.
 //
 // -no-addr-watch (Config.DisableAddrWatch) turns the watcher off.
 const (
@@ -150,8 +152,9 @@ type addrWatchState struct {
 	registryRetries int // registry-only retries still owed
 
 	lastRecover time.Time   // the latest recovery of any kind
-	localGap    addrBackoff // local-address and registry-retry recoveries
+	localGap    addrBackoff // local-address recoveries
 	observedGap addrBackoff // observed-endpoint recoveries
+	registryGap addrBackoff // registry-only retries, started over by every move
 }
 
 // addrBackoff is one input's run of recoveries.
@@ -243,6 +246,7 @@ func (st *addrWatchState) pending() string {
 
 // addrGapParams is the backoff schedule of the input behind reason: the
 // first gap, its cap, and how long without a recovery before it drops back.
+// Registry retries follow the local-address schedule.
 func addrGapParams(reason string) (base, max, quiet time.Duration) {
 	if reason == addrReasonObserved {
 		return addrObservedCooldown, addrObservedCooldownMax, addrObservedQuiet
@@ -252,8 +256,11 @@ func addrGapParams(reason string) (base, max, quiet time.Duration) {
 
 // gapFor returns the backoff of the input behind reason.
 func (st *addrWatchState) gapFor(reason string) *addrBackoff {
-	if reason == addrReasonObserved {
+	switch reason {
+	case addrReasonObserved:
 		return &st.observedGap
+	case addrReasonRegistry:
+		return &st.registryGap
 	}
 	return &st.localGap
 }
@@ -327,6 +334,10 @@ func (st *addrWatchState) began(now time.Time, reason string) (previous, current
 		st.registryRetries--
 	} else {
 		st.registryRetries = 0
+		// The retries this move may leave owed are spaced from it, on their
+		// own backoff. Counted in the move's, each one would hold the next
+		// move back as if the address had changed again.
+		st.registryGap = addrBackoff{last: now, streak: 1}
 	}
 	return previous, current
 }
@@ -550,25 +561,15 @@ func (d *Daemon) recoverFromAddrChange(reason, previous, current string, announc
 // otherwise relay to the old address until the next keepalive, up to a
 // minute later.
 func (d *Daemon) addrAnnounceFollowUp(started time.Time) {
-	sleep := func(wait time.Duration) bool {
-		t := time.NewTimer(wait)
-		defer t.Stop()
-		select {
-		case <-d.stopCh:
-			return false
-		case <-t.C:
-			return true
-		}
-	}
 	for _, wait := range addrAnnounceRetryDelays {
-		if !sleep(wait) {
+		if !d.sleepOrStop(wait) {
 			return
 		}
 		if d.announceToPeers(started) == 0 {
 			break
 		}
 	}
-	if !sleep(time.Until(started.Add(addrBeaconReregisterDelay))) {
+	if !d.sleepOrStop(time.Until(started.Add(addrBeaconReregisterDelay))) {
 		return
 	}
 	d.tunnels.RegisterWithBeacon()

pkg/daemon/daemon.go

@@ -452,9 +452,9 @@ type Daemon struct {
 	// address watcher) and the heartbeat's own in trustRepublishLoop. They
 	// can all want one at once, and one caller's forceReconnectRegistry
 	// closes the connection another is mid-request on. reestablishOKWall
-	// (wall-clock unix nanos, under the mutex) is when the last
-	// reestablishTransport run the registry accepted finished, and
-	// reestablishOKFull whether it was a full re-registration.
+	// (wall-clock unix nanos, under the mutex) is when the last of their
+	// re-registrations the registry accepted finished, and
+	// reestablishOKFull whether it was a full one (noteReestablished).
 	reestablishMu     sync.Mutex
 	reestablishOKWall int64
 	reestablishOKFull bool
@@ -1652,6 +1652,21 @@ func (d *Daemon) stopping() bool {
 	}
 }
 
+// sleepOrStop waits for wait, or until Stop is called, whichever comes
+// first. Reports whether the whole wait passed. Background work that backs
+// off uses it instead of time.Sleep: Stop gives that work only 5 s to
+// return before it carries on without it.
+func (d *Daemon) sleepOrStop(wait time.Duration) bool {
+	t := time.NewTimer(wait)
+	defer t.Stop()
+	select {
+	case <-d.stopCh:
+		return false
+	case <-t.C:
+		return true
+	}
+}
+
 func (d *Daemon) Stop() error {
 	// Idempotent: only the first caller runs shutdown; others wait and return nil.
 	d.stopOnce.Do(func() {
@@ -2414,6 +2429,11 @@ func (d *Daemon) dialRegistryClient() (*registry.Client, error) {
 	registryDialBackoff := 500 * time.Millisecond
 	dialOpts := d.registryDialOptions()
 	for attempt := 1; attempt <= maxRegistryDialAttempts; attempt++ {
+		// A recovery redialling a registry that refuses us would otherwise
+		// keep at it for the whole backoff, about 47 s, long after Stop.
+		if d.stopping() {
+			return nil, errDaemonStopping
+		}
 		if d.config.RegistryTLS {
 			trust := d.config.RegistryTrust
 			if trust == "" {
@@ -2448,7 +2468,9 @@ func (d *Daemon) dialRegistryClient() (*registry.Client, error) {
 		slog.Warn("registry dial failed, retrying",
 			"attempt", attempt, "max", maxRegistryDialAttempts,
 			"backoff", registryDialBackoff, "error", err)
-		time.Sleep(registryDialBackoff)
+		if !d.sleepOrStop(registryDialBackoff) {
+			return nil, errDaemonStopping
+		}
 		if registryDialBackoff < 5*time.Second {
 			registryDialBackoff *= 2
 		}
@@ -5753,6 +5775,7 @@ func (d *Daemon) trustRepublishLoop() {
 			if rc == nil {
 				continue
 			}
+			callStart := time.Now()
 			_, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
 				return rc.Heartbeat(d.NodeID())
 			})
@@ -5761,10 +5784,16 @@ func (d *Daemon) trustRepublishLoop() {
 				if errors.Is(err, errRegistryCallTimedOut) {
 					slog.Warn("heartbeat timed out — registry connection likely half-open, forcing reconnect",
 						"consecutive_failures", consecutiveFailures, "deadline", registryCallDeadline)
-					if rcErr := d.reconnectRegistrySerialised(); rcErr != nil {
+					rereg, rcErr := d.reconnectRegistrySerialised(rc, callStart)
+					switch {
+					case rcErr != nil:
 						slog.Warn("registry force-reconnect failed", "error", rcErr)
-					} else {
+					case rereg:
 						consecutiveFailures = HeartbeatReregThresh
+					default:
+						// A recovery has replaced the connection the call
+						// waited on, and re-registered over the new one.
+						slog.Info("registry connection already replaced by a recovery — not reconnecting again")
 					}
 				}
 				// If the registry rejects our identity (node not found, or a
@@ -5779,7 +5808,9 @@ func (d *Daemon) trustRepublishLoop() {
 				if consecutiveFailures >= HeartbeatReregThresh {
 					// Backoff with jitter before attempting re-registration.
 					jitter := time.Duration(rand.Int63n(int64(reregBackoff) / 2))
-					time.Sleep(reregBackoff + jitter)
+					if !d.sleepOrStop(reregBackoff + jitter) {
+						return
+					}
 
 					slog.Info("attempting re-registration", "backoff", reregBackoff)
 					_ = d.reRegisterSerialised() // logs its own failure; the next heartbeat tells
@@ -5807,16 +5838,48 @@ func (d *Daemon) trustRepublishLoop() {
 // own registry reconnect and re-registration, under reestablishMu like the
 // recoveries' (reestablishTransport): run at the same time as one, either
 // would replace or use the registry connection under the other's requests.
-func (d *Daemon) reconnectRegistrySerialised() error {
+//
+// Neither repeats what a recovery has just done. The heartbeat's call can
+// time out on a connection a recovery replaced while the call waited (a
+// resume whose own dial was slow); reconnecting would then replace the fresh
+// connection again, and the re-registration after it would repeat the full
+// restore, one ReportTrust per trusted peer included. So the reconnect
+// replaces stale, the connection the call timed out on, only while it is
+// still the current one.
+//
+// It reports whether the heartbeat should re-register now: after its own
+// reconnect, and also when a recovery replaced the connection but nothing
+// has registered since callStart (the recovery's re-registration failed).
+func (d *Daemon) reconnectRegistrySerialised(stale *registry.Client, callStart time.Time) (bool, error) {
 	d.reestablishMu.Lock()
 	defer d.reestablishMu.Unlock()
-	return d.forceReconnectRegistry()
+	if d.reg() != stale {
+		return d.reestablishOKWall < callStart.UnixNano(), nil
+	}
+	if err := d.forceReconnectRegistry(); err != nil {
+		return false, err
+	}
+	return true, nil
 }
 
+// reRegisterSerialised re-registers in full, unless a full re-registration
+// the registry accepted finished less than reestablishCoalesce ago. Its own
+// counts the same way for a recovery right after it, if it restored all of
+// this node's state.
 func (d *Daemon) reRegisterSerialised() error {
 	d.reestablishMu.Lock()
 	defer d.reestablishMu.Unlock()
-	return d.reRegister()
+	if age, ok := d.reestablishedRecently(true); ok {
+		slog.Info("registry re-registered moments ago — heartbeat not repeating it",
+			"age", age.Truncate(time.Millisecond).String())
+		return nil
+	}
+	full, err := d.reRegisterFull()
+	if err != nil {
+		return err
+	}
+	d.noteReestablished(full)
+	return nil
 }
 
 // tunnelKeepaliveLoop (L4) refreshes the daemon's beacon registration so the
@@ -5964,12 +6027,20 @@ func (d *Daemon) publishHeartbeatEvent() {
 // registry did not accept the registration; the restore steps after it are
 // best-effort and only logged.
 func (d *Daemon) reRegister() error {
+	_, err := d.reRegisterFull()
+	return err
+}
+
+// reRegisterFull is reRegister, and also reports whether the registry took
+// back all of this node's state. A write that failed is only logged; the
+// recoveries count such a run as endpoint-only (noteReestablished), so the
+// next full re-registration is not skipped as recently done.
+func (d *Daemon) reRegisterFull() (bool, error) {
 	nodeID, _, err := d.registerEndpoint()
 	if err != nil {
-		return err
+		return false, err
 	}
-	d.restoreRegistryState(nodeID, true)
-	return nil
+	return d.restoreRegistryState(nodeID, true), nil
 }
 
 // endpointOnlyRegistryFresh bounds when reRegisterEndpoint trusts the
@@ -6097,41 +6168,53 @@ func (d *Daemon) registerEndpoint() (uint32, map[string]interface{}, error) {
 
 // restoreRegistryState re-applies what a registry that lost this node also
 // lost: visibility and hostname, and with trust set the local trust pairs
-// and the beacon registration. Best-effort; failures are logged.
-func (d *Daemon) restoreRegistryState(nodeID uint32, trust bool) {
+// and the beacon registration. Best-effort; failures are logged. Reports
+// whether every write succeeded.
+func (d *Daemon) restoreRegistryState(nodeID uint32, trust bool) bool {
 	if d.stopping() {
-		return
+		return false
 	}
+	complete := true
 
 	// Restore visibility and hostname after re-registration
 	if d.config.Public {
 		if _, err := d.reg().SetVisibility(nodeID, true); err != nil {
 			slog.Warn("re-registration: failed to restore visibility", "error", err)
+			complete = false
 		}
 	}
 	if d.config.Hostname != "" {
 		if _, err := d.reg().SetHostname(nodeID, d.config.Hostname); err != nil {
 			slog.Warn("re-registration: failed to restore hostname", "error", err)
+			complete = false
 		}
 	}
 
-	if d.stopping() || !trust {
-		return
+	if d.stopping() {
+		return false
+	}
+	if !trust {
+		return complete
 	}
 
 	// Re-sync local trust pairs to registry (trust survives disconnection locally
 	// but the registry may have lost and re-loaded state)
 	if d.handshakes != nil {
 		peers := d.handshakes.TrustedPeers()
+		failed := 0
 		for _, rec := range peers {
 			if d.stopping() {
-				return
+				return false
 			}
 			if _, err := d.reg().ReportTrust(nodeID, rec.NodeID); err != nil {
 				slog.Debug("re-registration: failed to re-sync trust pair", "peer", rec.NodeID, "error", err)
+				failed++
 			}
 		}
-		if len(peers) > 0 {
+		if failed > 0 {
+			slog.Warn("re-registration: some trust pairs not re-synced", "synced", len(peers)-failed, "failed", failed)
+			complete = false
+		} else if len(peers) > 0 {
 			slog.Info("re-synced trust pairs", "count", len(peers))
 		}
 	}
@@ -6140,6 +6223,7 @@ func (d *Daemon) restoreRegistryState(nodeID uint32, trust bool) {
 	if d.config.BeaconAddr != "" {
 		d.tunnels.RegisterWithBeacon()
 	}
+	return complete
 }
 
 // reapStalePeers removes tunnel peers that have no active connections