From 9a4566d1d89f25e17f8a55f0c30d1c152cc193e1 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 11:50:37 +0300 Subject: [PATCH 1/6] daemon: registry retries no longer push back the next address move The address watcher's registry-only retries shared the local-address backoff, and each one bumped its streak. After a move the registry did not take, the three retries (+10 s, +30 s, +70 s) left the streak at 4, so the next real move waited 80 s after the last retry instead of 10 s. The retries now have a backoff of their own, on the local-address schedule and started over by every move, so they keep their +10/+30/+70 spacing and leave the move's cooldown alone. Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/addrwatch.go | 23 ++++++++++++----- pkg/daemon/zz_addrwatch_test.go | 45 +++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/pkg/daemon/addrwatch.go b/pkg/daemon/addrwatch.go index f380048a..59504a37 100644 --- a/pkg/daemon/addrwatch.go +++ b/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 } diff --git a/pkg/daemon/zz_addrwatch_test.go b/pkg/daemon/zz_addrwatch_test.go index e0e3e7d2..caa85fab 100644 --- a/pkg/daemon/zz_addrwatch_test.go +++ b/pkg/daemon/zz_addrwatch_test.go @@ -619,6 +619,51 @@ func TestAddrWatchRegistryRetryIsBounded(t *testing.T) { } } +// Registry retries do not count as moves. When they shared the local-address +// backoff, the three retries after a move (at +10 s, +30 s and +70 s) left its +// streak at 4, and the next real move waited 80 s after the last retry +// instead of the 10 s every recovery keeps from the one before. +func TestAddrWatchRegistryRetriesDoNotHoldBackTheNextMove(t *testing.T) { + ok := false // the registry is unreachable throughout + calls := swapAddrRecoverForTest(t, &ok) + d := newAddrWatchTestDaemon() + src := &fakeAddrSource{ip: "10.0.0.4", ok: true} + st := &addrWatchState{} + now := time.Now() + d.addrWatchTick(st, src.fn, now) + + src.ip = "10.0.0.77" + var fired []int + for i := 1; i <= 160; i++ { + if i == 72 { + src.ip = "10.0.0.78" // the next move, just after the last retry + } + if d.addrWatchTick(st, src.fn, now.Add(time.Duration(i)*time.Second)) { + fired = append(fired, i) + } + } + // The move, its three retries, then the next move 10 s after the last + // retry. Its own retries follow from 91 s on. + want := []int{1, 11, 31, 71, 81} + if len(fired) < len(want) { + t.Fatalf("fired at %v, want it to start %v", fired, want) + } + for i, w := range want { + if fired[i] != w { + t.Fatalf("fired at %v, want it to start %v", fired, want) + } + } + reasons := []string{addrReasonLocal, addrReasonRegistry, addrReasonRegistry, addrReasonRegistry, addrReasonLocal} + for i, r := range reasons { + if (*calls)[i].reason != r { + t.Fatalf("recovery %d = %+v, want reason %s", i, (*calls)[i], r) + } + } + if got := (*calls)[4]; got.previous != "10.0.0.77" || got.current != "10.0.0.78" || !got.announce { + t.Fatalf("second move = %+v, want an announced move from .77 to .78", got) + } +} + // The production source: the kernel's choice of source address for a target. // Toward loopback that is loopback, every time; an unrelated interface // appearing, a link-local address or a rotating IPv6 temporary address cannot From faee422855385e99fc4c7125c9e6a0fd4b8d0045 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 11:51:56 +0300 Subject: [PATCH 2/6] daemon: Stop no longer waits on a registry redial or heartbeat backoff The registry dial retries slept through Stop. A recovery that was reconnecting to a registry that refused connections kept dialling for its whole backoff, ten attempts over 47.5 s: Stop gave up on it after its 5 s wait and logged a leaked goroutine. The heartbeat's sleep before a re-registration (up to 30 s) did the same. Both backoffs now end when Stop is called (sleepOrStop, which the address watcher's follow-up already used as a closure), and each dial attempt checks for Stop first. Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/addrwatch.go | 14 ++-------- pkg/daemon/daemon.go | 28 ++++++++++++++++++-- pkg/daemon/zz_registry_halfopen_test.go | 35 +++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 14 deletions(-) diff --git a/pkg/daemon/addrwatch.go b/pkg/daemon/addrwatch.go index 59504a37..47a5bddb 100644 --- a/pkg/daemon/addrwatch.go +++ b/pkg/daemon/addrwatch.go @@ -561,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() diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 0b3af0ff..2bd28bbb 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -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 } @@ -5779,7 +5801,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 diff --git a/pkg/daemon/zz_registry_halfopen_test.go b/pkg/daemon/zz_registry_halfopen_test.go index 22b0a0da..20b0324e 100644 --- a/pkg/daemon/zz_registry_halfopen_test.go +++ b/pkg/daemon/zz_registry_halfopen_test.go @@ -147,3 +147,38 @@ func TestForceReconnectRegistryAfterStopDoesNotReplaceConn(t *testing.T) { t.Fatal("forceReconnectRegistry replaced the registry conn after Stop began") } } + +// A recovery redialling a registry that refuses connections gives way to +// Stop. Its dial backoff used to sleep through Stop: ten attempts took 47.5 s, +// and Stop gave up on it after 5 s and reported a leaked goroutine. +func TestForceReconnectRegistryGivesWayToStop(t *testing.T) { + t.Parallel() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := ln.Addr().String() + ln.Close() // nothing listens there now: every dial is refused + + d := New(Config{RegistryAddr: addr}) + done := make(chan error, 1) + go func() { done <- d.forceReconnectRegistry() }() + time.Sleep(200 * time.Millisecond) // inside the dial backoff + close(d.stopCh) + stopped := time.Now() + + select { + case err := <-done: + if !errors.Is(err, errDaemonStopping) { + t.Fatalf("reconnect after stop: err = %v, want errDaemonStopping", err) + } + if took := time.Since(stopped); took > time.Second { + t.Fatalf("reconnect returned %v after Stop, want within 1s", took) + } + case <-time.After(3 * time.Second): + t.Fatal("reconnect still dialling the registry 3s after Stop") + } + if d.reg() != nil { + t.Fatal("a registry connection was installed after Stop") + } +} From 3b492525ce3e6fe85cb8c6a50353e80e99d81c2f Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 11:55:11 +0300 Subject: [PATCH 3/6] daemon: the heartbeat does not repeat a registry recovery that just ran The heartbeat's reconnect and re-registration took reestablishMu, but never skipped work a recovery had just done and never recorded their own. When the heartbeat's call timed out (8 s) on the connection a resume recovery was replacing, it waited for the recovery, replaced the fresh connection again and re-registered in full a second time: two registers and two ReportTrust per trusted peer. - reestablishedRecently and noteReestablished are now shared by reestablishTransport and the heartbeat. - The heartbeat's reconnect takes the connection its call timed out on and does nothing if that has already been replaced; the heartbeat then does not escalate to a re-registration. - The heartbeat's re-registration is skipped when a full one the registry accepted finished under 5 s ago (wall clock), and records its own, so a recovery right after it skips its registry half. - reestablishTransport registers with the beacon before deciding to skip: the recent run may now be the heartbeat's, which does not. Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/daemon.go | 48 ++++++++++++--- pkg/daemon/rxwatchdog.go | 48 ++++++++++----- pkg/daemon/zz_reestablish_transport_test.go | 66 ++++++++++++++++++++- 3 files changed, 139 insertions(+), 23 deletions(-) diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 2bd28bbb..4f971bcd 100644 --- a/pkg/daemon/daemon.go +++ b/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 @@ -5783,10 +5783,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 { + reconnected, rcErr := d.reconnectRegistrySerialised(rc) + switch { + case rcErr != nil: slog.Warn("registry force-reconnect failed", "error", rcErr) - } else { + case reconnected: consecutiveFailures = HeartbeatReregThresh + default: + // A recovery has replaced the connection the call + // waited on, and re-registers 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 @@ -5831,16 +5837,42 @@ 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, and reports whether it did. +func (d *Daemon) reconnectRegistrySerialised(stale *registry.Client) (bool, error) { d.reestablishMu.Lock() defer d.reestablishMu.Unlock() - return d.forceReconnectRegistry() + if d.reg() != stale { + return false, 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. 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 + } + if err := d.reRegister(); err != nil { + return err + } + d.noteReestablished(true) + return nil } // tunnelKeepaliveLoop (L4) refreshes the daemon's beacon registration so the diff --git a/pkg/daemon/rxwatchdog.go b/pkg/daemon/rxwatchdog.go index 3bab006e..17b4f78e 100644 --- a/pkg/daemon/rxwatchdog.go +++ b/pkg/daemon/rxwatchdog.go @@ -266,14 +266,38 @@ type reestablishOpts struct { // instead of the full reRegister with its visibility, hostname and // trust re-sync. endpointOnly bool - // always runs even right after another run. Without it the call is - // skipped when a run the registry accepted finished less than + // always runs even right after another run. Without it the registry + // half is skipped when a re-registration the registry accepted (a + // recovery's, or the heartbeat's) finished less than // reestablishCoalesce ago and did at least the same work: a full // re-registration for a full caller. The address watcher sets it: a run // that started before our address changed registered the old one. always bool } +// reestablishedRecently reports whether a re-registration the registry +// accepted finished less than reestablishCoalesce ago and covers what the +// caller needs (for full, a full re-registration), and how long ago it +// finished. The caller holds reestablishMu. +// +// Wall clock, not the monotonic one: on Linux the monotonic clock stops +// while the host is suspended, so a run from just before a suspend would +// look recent to the resume handler. +func (d *Daemon) reestablishedRecently(full bool) (time.Duration, bool) { + if d.reestablishOKWall == 0 || (full && !d.reestablishOKFull) { + return 0, false + } + age := time.Duration(time.Now().UnixNano() - d.reestablishOKWall) + return age, age >= 0 && age < reestablishCoalesce +} + +// noteReestablished records a re-registration the registry accepted, full +// or endpoint-only. The caller holds reestablishMu. +func (d *Daemon) noteReestablished(full bool) { + d.reestablishOKWall = time.Now().UnixNano() + d.reestablishOKFull = full +} + // reestablishTransport re-registers this node with the beacon and the // registry. It is the one recovery routine shared by the rx watchdog's soft // recovery, the host-resume handler and the address watcher (addrwatch.go). @@ -288,18 +312,15 @@ func (d *Daemon) reestablishTransport(cause string, o reestablishOpts) bool { d.reestablishMu.Lock() defer d.reestablishMu.Unlock() - // Wall clock, not the monotonic one: on Linux the monotonic clock stops - // while the host is suspended, so a run from just before a suspend would - // look recent to the resume handler. - covered := o.endpointOnly || d.reestablishOKFull - if age := time.Now().UnixNano() - d.reestablishOKWall; !o.always && covered && d.reestablishOKWall != 0 && - age >= 0 && age < int64(reestablishCoalesce) { - slog.Info("transport re-established moments ago — not repeating it", "cause", cause, - "age", time.Duration(age).Truncate(time.Millisecond).String()) + // The beacon registration is one datagram and runs every time: the + // recent run that lets the registry half be skipped may be the + // heartbeat's re-registration, which does not register with the beacon. + d.tunnels.RegisterWithBeacon() + if age, ok := d.reestablishedRecently(!o.endpointOnly); ok && !o.always { + slog.Info("registry re-registered moments ago — not repeating it", "cause", cause, + "age", age.Truncate(time.Millisecond).String()) return true } - - d.tunnels.RegisterWithBeacon() if d.reg() == nil { return false } @@ -316,8 +337,7 @@ func (d *Daemon) reestablishTransport(cause string, o reestablishOpts) bool { if err := register(); err != nil { return false // registerEndpoint logs what the registry answered } - d.reestablishOKWall = time.Now().UnixNano() - d.reestablishOKFull = !o.endpointOnly + d.noteReestablished(!o.endpointOnly) return true } diff --git a/pkg/daemon/zz_reestablish_transport_test.go b/pkg/daemon/zz_reestablish_transport_test.go index 105b9b6c..d0765193 100644 --- a/pkg/daemon/zz_reestablish_transport_test.go +++ b/pkg/daemon/zz_reestablish_transport_test.go @@ -334,11 +334,15 @@ func TestHeartbeatRegistryCallsWaitForARecovery(t *testing.T) { defer stop() d := newRegistryTestDaemon(t, addr, Config{}) + reconnect := func() error { + _, err := d.reconnectRegistrySerialised(d.reg()) + return err + } for _, call := range []struct { name string fn func() error }{ - {"reconnect", d.reconnectRegistrySerialised}, + {"reconnect", reconnect}, {"re-register", d.reRegisterSerialised}, } { before := d.reg() @@ -367,3 +371,63 @@ func TestHeartbeatRegistryCallsWaitForARecovery(t *testing.T) { } } } + +// The heartbeat does not repeat a recovery that has just run. Its call can +// time out on the connection a recovery replaced while the call waited (a +// resume whose own dial was slow). It then replaced the fresh connection +// again and re-registered in full a second time: two registers and two +// ReportTrust per trusted peer. Its own re-registration counts for a +// recovery right after it the same way. +func TestHeartbeatDoesNotRepeatARecovery(t *testing.T) { + t.Parallel() + var log registryLog + addr, stop := serveFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} { + typ, _ := req["type"].(string) + log.add(typ) + if typ == "register" { + return registerOK("") + } + return map[string]interface{}{"type": typ + "_ok"} + }) + defer stop() + d := newRegistryTestDaemon(t, addr, Config{}) + fs := installFakeHandshake(d) + fs.trustedRecs = []HandshakeTrustRecord{{NodeID: 41}, {NodeID: 42}} + + stale := d.reg() // the heartbeat's call is waiting on this connection + if !d.reestablishTransport("resume", reestablishOpts{freshConn: true}) { + t.Fatal("resume recovery failed") + } + fresh := d.reg() + + // The call times out: the heartbeat reconnects, then re-registers. + reconnected, err := d.reconnectRegistrySerialised(stale) + if err != nil { + t.Fatalf("heartbeat reconnect: %v", err) + } + if reconnected || d.reg() != fresh { + t.Fatal("the heartbeat replaced the connection the recovery had just opened") + } + if err := d.reRegisterSerialised(); err != nil { + t.Fatalf("heartbeat re-register: %v", err) + } + if log.get("register") != 1 || log.get("report_trust") != 2 { + t.Fatalf("resume then heartbeat sent %v, want one register and one report_trust per trusted peer", log.count) + } + + // The other way round: a resume right after the heartbeat's own + // re-registration skips its registry half. + d.reestablishMu.Lock() + d.reestablishOKWall = time.Now().Add(-2 * reestablishCoalesce).UnixNano() + d.reestablishMu.Unlock() + log.reset() + if err := d.reRegisterSerialised(); err != nil { + t.Fatalf("heartbeat re-register: %v", err) + } + if !d.reestablishTransport("resume", reestablishOpts{freshConn: true}) { + t.Fatal("resume recovery failed") + } + if log.get("register") != 1 || d.reg() != fresh { + t.Fatalf("heartbeat then resume sent %v and replaced the connection: %v, want one register", log.count, d.reg() != fresh) + } +} From 321d8e7e20022e4aa20cec6d36dfd0392e747fbc Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 11:55:30 +0300 Subject: [PATCH 4/6] CHANGELOG: address watcher follow-ups Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 400bb48b..9dacf167 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -386,6 +386,11 @@ 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 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 From e702a3a956efecfffd3cec240e1d5f08b0518961 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 12:27:36 +0300 Subject: [PATCH 5/6] daemon: a failed recovery or partial restore does not stand for a re-registration Two cases where the heartbeat's re-registration was skipped although nothing had restored the node: - A recovery replaced the connection the heartbeat's call waited on, but its own re-registration failed. The heartbeat took the replaced connection as a sign the recovery had re-registered and did not escalate. It now re-registers unless something registered after its call began. - A full re-registration whose visibility, hostname or trust writes partly failed was recorded as full, so a heartbeat re-registration within 5 s skipped re-syncing them. It now counts as endpoint-only. restoreRegistryState reports whether every write succeeded, and its log no longer says trust pairs were re-synced when some failed. Also corrects the comment on the beacon registration: the heartbeat's full re-registration does register with the beacon; an endpoint-only one does not. Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/daemon.go | 66 ++++++++++----- pkg/daemon/rxwatchdog.go | 25 ++++-- pkg/daemon/zz_reestablish_transport_test.go | 91 ++++++++++++++++++++- 3 files changed, 151 insertions(+), 31 deletions(-) diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 4f971bcd..a258c472 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -5775,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()) }) @@ -5783,15 +5784,15 @@ 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) - reconnected, rcErr := d.reconnectRegistrySerialised(rc) + rereg, rcErr := d.reconnectRegistrySerialised(rc, callStart) switch { case rcErr != nil: slog.Warn("registry force-reconnect failed", "error", rcErr) - case reconnected: + case rereg: consecutiveFailures = HeartbeatReregThresh default: // A recovery has replaced the connection the call - // waited on, and re-registers over the new one. + // waited on, and re-registered over the new one. slog.Info("registry connection already replaced by a recovery — not reconnecting again") } } @@ -5844,12 +5845,16 @@ func (d *Daemon) trustRepublishLoop() { // 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, and reports whether it did. -func (d *Daemon) reconnectRegistrySerialised(stale *registry.Client) (bool, error) { +// 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() if d.reg() != stale { - return false, nil + return d.reestablishOKWall < callStart.UnixNano(), nil } if err := d.forceReconnectRegistry(); err != nil { return false, err @@ -5859,7 +5864,8 @@ func (d *Daemon) reconnectRegistrySerialised(stale *registry.Client) (bool, erro // 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. +// 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() @@ -5868,10 +5874,11 @@ func (d *Daemon) reRegisterSerialised() error { "age", age.Truncate(time.Millisecond).String()) return nil } - if err := d.reRegister(); err != nil { + full, err := d.reRegisterFull() + if err != nil { return err } - d.noteReestablished(true) + d.noteReestablished(full) return nil } @@ -6020,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 @@ -6153,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)) } } @@ -6196,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 diff --git a/pkg/daemon/rxwatchdog.go b/pkg/daemon/rxwatchdog.go index 17b4f78e..ca134204 100644 --- a/pkg/daemon/rxwatchdog.go +++ b/pkg/daemon/rxwatchdog.go @@ -291,8 +291,10 @@ func (d *Daemon) reestablishedRecently(full bool) (time.Duration, bool) { return age, age >= 0 && age < reestablishCoalesce } -// noteReestablished records a re-registration the registry accepted, full -// or endpoint-only. The caller holds reestablishMu. +// noteReestablished records a re-registration the registry accepted: full +// when the registry also took back all of this node's state (visibility, +// hostname, every trust pair), so that a full one after it would repeat it. +// The caller holds reestablishMu. func (d *Daemon) noteReestablished(full bool) { d.reestablishOKWall = time.Now().UnixNano() d.reestablishOKFull = full @@ -312,9 +314,11 @@ func (d *Daemon) reestablishTransport(cause string, o reestablishOpts) bool { d.reestablishMu.Lock() defer d.reestablishMu.Unlock() - // The beacon registration is one datagram and runs every time: the - // recent run that lets the registry half be skipped may be the - // heartbeat's re-registration, which does not register with the beacon. + // The beacon registration is one datagram and runs every time, also + // when the registry half below is skipped as recently done: an + // endpoint-only re-registration does not register with the beacon, and + // the rx watchdog's soft recovery counts on the beacon's reply as + // inbound traffic. d.tunnels.RegisterWithBeacon() if age, ok := d.reestablishedRecently(!o.endpointOnly); ok && !o.always { slog.Info("registry re-registered moments ago — not repeating it", "cause", cause, @@ -330,14 +334,17 @@ func (d *Daemon) reestablishTransport(cause string, o reestablishOpts) bool { return false } } - register := d.reRegister + full := false + var err error if o.endpointOnly { - register = d.reRegisterEndpoint + err = d.reRegisterEndpoint() + } else { + full, err = d.reRegisterFull() } - if err := register(); err != nil { + if err != nil { return false // registerEndpoint logs what the registry answered } - d.noteReestablished(!o.endpointOnly) + d.noteReestablished(full) return true } diff --git a/pkg/daemon/zz_reestablish_transport_test.go b/pkg/daemon/zz_reestablish_transport_test.go index d0765193..495dcb4a 100644 --- a/pkg/daemon/zz_reestablish_transport_test.go +++ b/pkg/daemon/zz_reestablish_transport_test.go @@ -335,7 +335,7 @@ func TestHeartbeatRegistryCallsWaitForARecovery(t *testing.T) { d := newRegistryTestDaemon(t, addr, Config{}) reconnect := func() error { - _, err := d.reconnectRegistrySerialised(d.reg()) + _, err := d.reconnectRegistrySerialised(d.reg(), time.Now()) return err } for _, call := range []struct { @@ -395,17 +395,18 @@ func TestHeartbeatDoesNotRepeatARecovery(t *testing.T) { fs.trustedRecs = []HandshakeTrustRecord{{NodeID: 41}, {NodeID: 42}} stale := d.reg() // the heartbeat's call is waiting on this connection + callStart := time.Now() if !d.reestablishTransport("resume", reestablishOpts{freshConn: true}) { t.Fatal("resume recovery failed") } fresh := d.reg() // The call times out: the heartbeat reconnects, then re-registers. - reconnected, err := d.reconnectRegistrySerialised(stale) + rereg, err := d.reconnectRegistrySerialised(stale, callStart) if err != nil { t.Fatalf("heartbeat reconnect: %v", err) } - if reconnected || d.reg() != fresh { + if rereg || d.reg() != fresh { t.Fatal("the heartbeat replaced the connection the recovery had just opened") } if err := d.reRegisterSerialised(); err != nil { @@ -431,3 +432,87 @@ func TestHeartbeatDoesNotRepeatARecovery(t *testing.T) { t.Fatalf("heartbeat then resume sent %v and replaced the connection: %v, want one register", log.count, d.reg() != fresh) } } + +// A recovery that replaced the connection the heartbeat's call waited on, +// but whose re-registration failed, does not stand for one: the heartbeat +// re-registers over the new connection at once, as it would have after its +// own reconnect. +func TestHeartbeatReregistersWhenTheRecoveryDidNot(t *testing.T) { + t.Parallel() + var failRegister atomic.Bool + addr, stop := serveFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} { + typ, _ := req["type"].(string) + if typ == "register" { + if failRegister.Load() { + return map[string]interface{}{"type": "error", "error": "registry overloaded"} + } + return registerOK("") + } + return map[string]interface{}{"type": typ + "_ok"} + }) + defer stop() + d := newRegistryTestDaemon(t, addr, Config{}) + + stale := d.reg() + callStart := time.Now() + failRegister.Store(true) + if d.reestablishTransport("resume", reestablishOpts{freshConn: true}) { + t.Fatal("the resume's re-registration was refused, yet it reported success") + } + if d.reg() == stale { + t.Fatal("the resume did not replace the connection; nothing to test") + } + rereg, err := d.reconnectRegistrySerialised(stale, callStart) + if err != nil { + t.Fatalf("heartbeat reconnect: %v", err) + } + if !rereg { + t.Fatal("nothing registered since the heartbeat's call began, yet it would not re-register") + } +} + +// A full re-registration whose trust re-sync partly failed does not count +// as full: the heartbeat's re-registration right after it runs, and re-syncs +// the trust pairs, instead of being skipped as recently done. +func TestPartialRestoreDoesNotCountAsFull(t *testing.T) { + t.Parallel() + var log registryLog + var failTrust atomic.Bool + failTrust.Store(true) + addr, stop := serveFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} { + typ, _ := req["type"].(string) + log.add(typ) + switch { + case typ == "register": + return registerOK("") + case typ == "report_trust" && failTrust.Load(): + return map[string]interface{}{"type": "error", "error": "registry overloaded"} + } + return map[string]interface{}{"type": typ + "_ok"} + }) + defer stop() + d := newRegistryTestDaemon(t, addr, Config{}) + fs := installFakeHandshake(d) + fs.trustedRecs = []HandshakeTrustRecord{{NodeID: 41}, {NodeID: 42}} + + if !d.reestablishTransport("resume", reestablishOpts{}) { + t.Fatal("resume recovery failed") + } + failTrust.Store(false) + log.reset() + if err := d.reRegisterSerialised(); err != nil { + t.Fatalf("heartbeat re-register: %v", err) + } + if log.get("register") != 1 || log.get("report_trust") != 2 { + t.Fatalf("heartbeat after a partial restore sent %v, want a register and both trust pairs again", log.count) + } + + // Now complete, it counts: a resume right after it skips the registry. + log.reset() + if !d.reestablishTransport("resume", reestablishOpts{}) { + t.Fatal("resume recovery failed") + } + if log.get("register") != 0 { + t.Fatalf("resume right after a complete re-registration sent %v, want none", log.count) + } +} From d8677caa08981fb2fbf3b1db8056cb3a76a71ca8 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 12:44:31 +0300 Subject: [PATCH 6/6] CHANGELOG: a failed or partial recovery does not count Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9dacf167..1b9defad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -390,7 +390,9 @@ Detailed per-release notes are on the 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. + 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