deps: common v0.5.15; registry reconnect cannot race Stop

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

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=

pkg/daemon/daemon.go

@@ -73,6 +73,10 @@
 
 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) 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 @@
 	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
 }

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")
+	}
+}