diff --git a/CHANGELOG.md b/CHANGELOG.md index 400bb48b..1b9defad 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/pkg/daemon/addrwatch.go b/pkg/daemon/addrwatch.go index f380048a..47a5bddb 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 } @@ -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() diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 0b3af0ff..a258c472 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 @@ -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 diff --git a/pkg/daemon/rxwatchdog.go b/pkg/daemon/rxwatchdog.go index 3bab006e..ca134204 100644 --- a/pkg/daemon/rxwatchdog.go +++ b/pkg/daemon/rxwatchdog.go @@ -266,14 +266,40 @@ 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 +// 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 +} + // 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 +314,17 @@ 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, 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, + "age", age.Truncate(time.Millisecond).String()) return true } - - d.tunnels.RegisterWithBeacon() if d.reg() == nil { return false } @@ -309,15 +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.reestablishOKWall = time.Now().UnixNano() - d.reestablishOKFull = !o.endpointOnly + d.noteReestablished(full) return true } 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 diff --git a/pkg/daemon/zz_reestablish_transport_test.go b/pkg/daemon/zz_reestablish_transport_test.go index 105b9b6c..495dcb4a 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(), time.Now()) + 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,148 @@ 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 + 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. + rereg, err := d.reconnectRegistrySerialised(stale, callStart) + if err != nil { + t.Fatalf("heartbeat reconnect: %v", err) + } + if rereg || 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) + } +} + +// 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) + } +} 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") + } +}