diff --git a/go.mod b/go.mod index 13fd5bf7..f943da36 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/coder/websocket v1.8.15 github.com/pilot-protocol/app-store v1.0.3 github.com/pilot-protocol/beacon v0.2.9 - github.com/pilot-protocol/common v0.5.13 + github.com/pilot-protocol/common v0.5.15 github.com/pilot-protocol/dataexchange v0.2.2 github.com/pilot-protocol/eventstream v0.2.4 github.com/pilot-protocol/handshake v0.2.8 diff --git a/go.sum b/go.sum index fd601b11..76434d13 100644 --- a/go.sum +++ b/go.sum @@ -237,8 +237,8 @@ github.com/pilot-protocol/app-store v1.0.3 h1:LfY6iZW6awTPV9nnQn+u/sTG+2UfJPxzS+ github.com/pilot-protocol/app-store v1.0.3/go.mod h1:Sz1vIZ92zNMWEnTJTXNxw7QmlbPRXs73qY/duNC0tgs= github.com/pilot-protocol/beacon v0.2.9 h1:VqXAtRKl4YhZVkDmJzuNhS1bfpK0n9qbGKOGLfVYSyo= github.com/pilot-protocol/beacon v0.2.9/go.mod h1:DE8masXGku/IwfHL8lN/4CUTe5YsxMXFe2fRGJcH7w8= -github.com/pilot-protocol/common v0.5.13 h1:h9NmPh37ZZujktpSAwgq2H2ai7jG240nQTetGNWtZ3U= -github.com/pilot-protocol/common v0.5.13/go.mod h1:Ybc6f1A37s3ShoEh1nBMVL9DPyYlxvkqPTvtbxaNWg4= +github.com/pilot-protocol/common v0.5.15 h1:ds6DNWlth97pMcXZ7fMgfy9/s553la3/YuYOkwoXjjE= +github.com/pilot-protocol/common v0.5.15/go.mod h1:OTXD84ScrBbKTAhawgFlLw9HtisgNvr83Zs1yDAuG9k= github.com/pilot-protocol/dataexchange v0.2.2 h1:h1VJFqFCdMDtX1E2E8zxTQPw+9rWnMohXN8b6HpB2R0= github.com/pilot-protocol/dataexchange v0.2.2/go.mod h1:TUj2QtNMZ4oMnOag1j5k5qDxz2457S8uaB2FJif1Ulk= github.com/pilot-protocol/eventstream v0.2.4 h1:SyB64wqo+Qpz0hIAuzHGkI9Npwk5SFSg1aqNFLueUIQ= diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 8983525f..8a42c7c1 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -73,6 +73,10 @@ const registryCallDeadline = 8 * time.Second var errRegistryCallTimedOut = errors.New("registry: call timed out, connection likely half-open") +// errDaemonStopping is returned by background work that declines to start +// (or undoes itself) because Stop has begun. +var errDaemonStopping = errors.New("daemon is stopping") + func withRegistryDeadline(timeout time.Duration, fn func() (map[string]interface{}, error)) (map[string]interface{}, error) { type registryCallResult struct { resp map[string]interface{} @@ -2274,6 +2278,12 @@ func (d *Daemon) dialRegistryClient() (*registry.Client, error) { } func (d *Daemon) forceReconnectRegistry() error { + // The heartbeat and rx-watchdog call this without coordinating with + // Stop. Once Stop has begun, a reconnect would only leak a fresh pool + // that nothing closes. + if d.stopping() { + return errDaemonStopping + } newConn, err := d.dialRegistryClient() if err != nil { return err @@ -2282,6 +2292,12 @@ func (d *Daemon) forceReconnectRegistry() error { if old != nil { go old.Close() } + // Stop may have closed the registry between the check above and the + // swap; if so it closed the old conn, so close the one we installed. + if d.stopping() { + newConn.Close() + return errDaemonStopping + } slog.Warn("registry connection force-reconnected after half-open detection", "addr", d.config.RegistryAddr) return nil } diff --git a/pkg/daemon/zz_registry_halfopen_test.go b/pkg/daemon/zz_registry_halfopen_test.go index 651ba2d8..22b0a0da 100644 --- a/pkg/daemon/zz_registry_halfopen_test.go +++ b/pkg/daemon/zz_registry_halfopen_test.go @@ -126,3 +126,24 @@ func TestForceReconnectRegistryRecoversFromHalfOpenConn(t *testing.T) { t.Fatalf("register response missing node_id: %v", resp) } } + +// A reconnect requested after Stop began (the heartbeat and rx-watchdog +// call forceReconnectRegistry without coordinating with Stop) must not +// install a fresh registry pool that nothing will ever close. +func TestForceReconnectRegistryAfterStopDoesNotReplaceConn(t *testing.T) { + t.Parallel() + reg, liveRC := startTestRegistry(t) + defer reg.Close() + defer liveRC.Close() + + d := New(Config{RegistryAddr: reg.Addr().String()}) + d.regConn.Store(liveRC) + close(d.stopCh) + + if err := d.forceReconnectRegistry(); !errors.Is(err, errDaemonStopping) { + t.Fatalf("forceReconnectRegistry after stop: err = %v, want errDaemonStopping", err) + } + if got := d.reg(); got != liveRC { + t.Fatal("forceReconnectRegistry replaced the registry conn after Stop began") + } +}