[client] Avoid Signal reconnect listener shutdown deadlock (#8257)

* [client] Reproduce Signal reconnect listener shutdown deadlock

* [client] Invoke Signal reconnect listeners outside the client mutex

Capture the listener with the registration state, then release the mutex before calling it. Watcher shutdown can now unregister a listener while an in-flight callback waits for the watcher lock.

Keep callback invocation synchronous and document that unregistration does not drain an already captured callback. Record callback completion inside the regression test's external lock.

shared/signal/client/grpc.go

@@ -299,7 +299,6 @@ func (c *GrpcClient) notifyStreamDisconnected() {
 
 func (c *GrpcClient) notifyStreamConnected() {
 	c.mux.Lock()
-	defer c.mux.Unlock()
 
 	c.status = StreamConnected
 	if c.connectedCh != nil {
@@ -308,8 +307,12 @@ func (c *GrpcClient) notifyStreamConnected() {
 		c.connectedCh = nil
 	}
 
-	if c.onReconnectedListenerFn != nil {
-		c.onReconnectedListenerFn()
+	listener := c.onReconnectedListenerFn
+	c.mux.Unlock()
+
+	// Listeners can hold their own locks while unregistering from this client.
+	if listener != nil {
+		listener()
 	}
 }
 
@@ -409,6 +412,7 @@ func (c *GrpcClient) WaitStreamConnected(ctx context.Context) {
 	}
 }
 
+// SetOnReconnectedListener replaces the listener without waiting for an in-flight callback.
 func (c *GrpcClient) SetOnReconnectedListener(fn func()) {
 	c.mux.Lock()
 	defer c.mux.Unlock()

shared/signal/client/reconnect_test.go

@@ -0,0 +1,98 @@
+package client
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReconnectListenerCanBeClearedWhileCallbackIsBlocked(t *testing.T) {
+	client := &GrpcClient{}
+	var listenerMu sync.Mutex
+	callbackStarted := make(chan struct{})
+	callbackFinished := make(chan struct{})
+	callbackCalls := 0
+	client.SetOnReconnectedListener(func() {
+		callbackCalls++
+		close(callbackStarted)
+		listenerMu.Lock()
+		close(callbackFinished)
+		listenerMu.Unlock()
+	})
+
+	// SRWatcher.Close holds its own mutex while removing the listener. A
+	// reconnect callback waiting for that mutex must not block its removal.
+	listenerMu.Lock()
+	notificationDone := make(chan struct{})
+	go func() {
+		defer close(notificationDone)
+		client.notifyStreamConnected()
+	}()
+	var unregisterDone chan struct{}
+	t.Cleanup(func() {
+		// Release the callback even when the regression assertion fails, so
+		// the unfixed implementation does not leave deadlocked goroutines.
+		listenerMu.Unlock()
+		select {
+		case <-notificationDone:
+			select {
+			case <-callbackFinished:
+			default:
+				t.Error("in-flight callback did not finish after listener removal")
+			}
+		case <-time.After(5 * time.Second):
+			t.Error("reconnect notification did not finish after releasing the listener")
+		}
+		if unregisterDone != nil {
+			select {
+			case <-unregisterDone:
+			case <-time.After(5 * time.Second):
+				t.Error("listener removal did not finish after releasing the listener")
+			}
+		}
+	})
+
+	select {
+	case <-callbackStarted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("reconnect callback did not start")
+	}
+
+	unregisterDone = make(chan struct{})
+	go func() {
+		client.SetOnReconnectedListener(nil)
+		close(unregisterDone)
+	}()
+	select {
+	case <-unregisterDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("listener removal blocked behind the reconnect callback")
+	}
+
+	// Registration still becomes visible, and future notifications do not
+	// invoke the removed listener even while its first callback is in flight.
+	assert.True(t, client.StreamConnected(), "listener removal must preserve stream registration")
+	client.notifyStreamConnected()
+	assert.Equal(t, 1, callbackCalls, "future notifications must not invoke the removed listener")
+}
+
+func TestReconnectListenerRunsAfterRegistrationWaitersAreReleased(t *testing.T) {
+	connected := make(chan struct{})
+	client := &GrpcClient{connectedCh: connected}
+	called := false
+	client.SetOnReconnectedListener(func() {
+		called = true
+		select {
+		case <-connected:
+		default:
+			t.Error("reconnect callback ran before registration waiters were released")
+		}
+	})
+
+	client.notifyStreamConnected()
+
+	assert.True(t, called, "reconnect notification must invoke its listener synchronously")
+	assert.True(t, client.StreamConnected(), "stream must remain registered after the callback")
+}