node: Don't hold mutex on subscription channels in Spy

node/cmd/spy/spy.go

@@ -110,8 +110,12 @@ func subscriptionId() string {
 }
 
 func (s *spyServer) PublishSignedVAA(vaaBytes []byte) error {
+	// Collect matching subscribers under the lock, then send outside the lock.
+	// This prevents a slow subscriber from blocking the mutex and stalling all
+	// other publishers and subscriber cleanup.
+	var targets []*subscriptionSignedVaa
+
 	s.subsSignedVaaMu.Lock()
-	defer s.subsSignedVaaMu.Unlock()
 
 	var v *vaa.VAA
 	var err error
@@ -123,16 +127,18 @@ func (s *spyServer) PublishSignedVAA(vaaBytes []byte) error {
 				verified = true
 				v, err = s.verifyVAA(v, vaaBytes)
 				if err != nil {
+					s.subsSignedVaaMu.Unlock()
 					return err
 				}
 			}
-			sub.ch <- message{vaaBytes: vaaBytes} //nolint:channelcheck // Don't want to drop incoming VAAs
+			targets = append(targets, sub)
 			continue
 		}
 
 		if v == nil {
 			v, err = vaa.Unmarshal(vaaBytes)
 			if err != nil {
+				s.subsSignedVaaMu.Unlock()
 				return err
 			}
 		}
@@ -143,15 +149,25 @@ func (s *spyServer) PublishSignedVAA(vaaBytes []byte) error {
 					verified = true
 					v, err = s.verifyVAA(v, vaaBytes)
 					if err != nil {
+						s.subsSignedVaaMu.Unlock()
 						return err
 					}
 				}
-				sub.ch <- message{vaaBytes: vaaBytes} //nolint:channelcheck // Don't want to drop incoming VAAs
+				targets = append(targets, sub)
+				// A VAA has a single (EmitterChain, EmitterAddress), so at most one filter can
+				// match. Stop here so a subscription is never collected more than once.
+				break
 			}
 		}
 
 	}
 
+	s.subsSignedVaaMu.Unlock()
+
+	for _, sub := range targets {
+		sub.ch <- message{vaaBytes: vaaBytes} //nolint:channelcheck // Don't want to drop incoming VAAs
+	}
+
 	return nil
 }
 
@@ -183,6 +199,13 @@ func (s *spyServer) verifyVAA(v *vaa.VAA, vaaBytes []byte) (*vaa.VAA, error) {
 func (s *spyServer) SubscribeSignedVAA(req *spyv1.SubscribeSignedVAARequest, resp spyv1.SpyRPCService_SubscribeSignedVAAServer) error {
 	var fi []filterSignedVaa
 	if req.Filters != nil {
+		// Deduplicate filters. A VAA has a single (EmitterChain, EmitterAddress), so only identical
+		// filters can match it. Without dedup, duplicate filters would cause the same subscription to
+		// be collected multiple times in PublishSignedVAA and the same VAA to be sent more than once
+		// to the subscription's (buffered, capacity 1) channel. Because that send is intentionally
+		// blocking (we don't drop VAAs), more than one queued send to a torn-down subscriber can block
+		// the single publisher goroutine forever, halting delivery to all subscribers.
+		seen := make(map[filterSignedVaa]struct{}, len(req.Filters))
 		for _, f := range req.Filters {
 			switch t := f.Filter.(type) {
 			case *spyv1.FilterEntry_EmitterFilter:
@@ -193,10 +216,15 @@ func (s *spyServer) SubscribeSignedVAA(req *spyv1.SubscribeSignedVAARequest, res
 				if t.EmitterFilter.GetChainId() > math.MaxUint16 {
 					return status.Error(codes.InvalidArgument, fmt.Sprintf("emitter chain id must be a valid 16 bit unsigned integer: %v", t.EmitterFilter.ChainId.Number()))
 				}
-				fi = append(fi, filterSignedVaa{
+				filter := filterSignedVaa{
 					chainId:     vaa.ChainID(t.EmitterFilter.ChainId), // #nosec G115 -- This is validated above
 					emitterAddr: addr,
-				})
+				}
+				if _, ok := seen[filter]; ok {
+					continue
+				}
+				seen[filter] = struct{}{}
+				fi = append(fi, filter)
 			default:
 				return status.Error(codes.InvalidArgument, "unsupported filter type")
 			}

node/cmd/spy/spy_test.go

@@ -153,6 +153,51 @@ func TestSpySubscribeEmitterFilter(t *testing.T) {
 	// just check the subscription can be created without returning an error
 }
 
+// Tests that duplicate (identical) filters supplied at subscribe time are deduplicated.
+// Without dedup, a single VAA would be collected once per duplicate filter in PublishSignedVAA
+// and sent multiple times to the subscription's capacity-1 channel; because that send is
+// intentionally blocking, more than one queued send to a torn-down subscriber could block the
+// single publisher goroutine forever.
+func TestSpySubscribeDeduplicatesFilters(t *testing.T) {
+	ctx, conn, client := grpcClientSetup(t)
+	defer conn.Close()
+
+	// Use a unique emitter so this subscription is unambiguous among any others in the shared server.
+	dedupEmitter := vaa.Address{0xde, 0xdb, 0xee, 0xff} // distinct from govEmitter and differentEmitter
+	emitterFilter := &spyv1.EmitterFilter{
+		ChainId:        publicrpcv1.ChainID(vaa.ChainIDEthereum),
+		EmitterAddress: dedupEmitter.String(),
+	}
+	filter := &spyv1.FilterEntry{Filter: &spyv1.FilterEntry_EmitterFilter{EmitterFilter: emitterFilter}}
+
+	// Subscribe with the same filter three times.
+	req := &spyv1.SubscribeSignedVAARequest{Filters: []*spyv1.FilterEntry{filter, filter, filter}}
+	if _, err := client.SubscribeSignedVAA(ctx, req); err != nil {
+		t.Fatalf("SubscribeSignedVAA failed: %v", err)
+	}
+
+	waitForClientSubscriptionInit(mockedSpyServer)
+
+	// Find the subscription with our unique emitter and assert its filters were deduplicated to one.
+	want := filterSignedVaa{chainId: vaa.ChainIDEthereum, emitterAddr: dedupEmitter}
+	mockedSpyServer.subsSignedVaaMu.Lock()
+	defer mockedSpyServer.subsSignedVaaMu.Unlock()
+	found := false
+	for _, sub := range mockedSpyServer.subsSignedVaa {
+		for _, f := range sub.filters {
+			if f == want {
+				found = true
+				if len(sub.filters) != 1 {
+					t.Fatalf("expected duplicate filters to be deduplicated to 1, got %d", len(sub.filters))
+				}
+			}
+		}
+	}
+	if !found {
+		t.Fatal("did not find the subscription with the deduplicated emitter filter")
+	}
+}
+
 // Tests a subscription to spySever with no filters will return message(s)
 func TestSpyHandleGossipVAA(t *testing.T) {
 	ctx, conn, client := grpcClientSetup(t)