From e3518564cb0288ce65131bfe2c7dc5aed9ca3172 Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Thu, 12 Mar 2026 14:50:01 -0600 Subject: [PATCH 1/2] node: Don't hold mutex on subscription channels in Spy --- node/cmd/spy/spy.go | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/node/cmd/spy/spy.go b/node/cmd/spy/spy.go index af496db16e4..97fc8018d3c 100644 --- a/node/cmd/spy/spy.go +++ b/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,22 @@ 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) } } } + s.subsSignedVaaMu.Unlock() + + for _, sub := range targets { + sub.ch <- message{vaaBytes: vaaBytes} //nolint:channelcheck // Don't want to drop incoming VAAs + } + return nil } From b217390cf65948b93580733e5b38dd6034e1b5e8 Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Tue, 16 Jun 2026 16:51:12 -0600 Subject: [PATCH 2/2] De-duplicate filters --- node/cmd/spy/spy.go | 19 +++++++++++++++-- node/cmd/spy/spy_test.go | 45 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/node/cmd/spy/spy.go b/node/cmd/spy/spy.go index 97fc8018d3c..9ff7ab601fc 100644 --- a/node/cmd/spy/spy.go +++ b/node/cmd/spy/spy.go @@ -154,6 +154,9 @@ func (s *spyServer) PublishSignedVAA(vaaBytes []byte) error { } } 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 } } @@ -196,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: @@ -206,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") } diff --git a/node/cmd/spy/spy_test.go b/node/cmd/spy/spy_test.go index 124df257287..bb46c808ae6 100644 --- a/node/cmd/spy/spy_test.go +++ b/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)