rpcclient: manage subscriptions sequentially

[AnnaShaleva]

Close #3093.

@roman-khimov, it's the best option that I can offer based on #3093 (comment) since I still don't understand the source of the problem. The same effect can be reached with additional mutex without any subscriptionDispatcher routine, so probably I'm doing something wrong, need your review.

Also, I'm still trying to reproduce the original problem locally with privnet.

[AnnaShaleva]

The same effect can be reached with additional mutex

So if the idea of this PR is right, then an alternative (which looks simpler to me) is:

anna@kiwi:~/Documents/GitProjects/nspcc-dev/neo-go$ git diff
diff --git a/pkg/rpcclient/wsclient.go b/pkg/rpcclient/wsclient.go
index 887e6aa59..2119ef978 100644
--- a/pkg/rpcclient/wsclient.go
+++ b/pkg/rpcclient/wsclient.go
@@ -71,8 +71,9 @@ type WSClient struct {
        closeErrLock sync.RWMutex
        closeErr     error
 
-       subscriptionsLock sync.RWMutex
-       subscriptions     map[string]notificationReceiver
+       subscriptionsOrderLock sync.Mutex
+       subscriptionsLock      sync.RWMutex
+       subscriptions          map[string]notificationReceiver
        // receivers is a mapping from receiver channel to a set of corresponding subscription IDs.
        // It must be accessed with subscriptionsLock taken. Its keys must be used to deliver
        // notifications, if channel is not in the receivers list and corresponding subscription
@@ -783,6 +784,9 @@ func (c *WSClient) performSubscription(params []any, rcvr notificationReceiver)
                        return "", err
                }
        }
+
+       c.subscriptionsOrderLock.Lock()
+       defer c.subscriptionsOrderLock.Unlock()
        if err := c.performRequest("subscribe", params, &resp); err != nil {
                return "", err
        }
@@ -972,13 +976,8 @@ func (c *WSClient) UnsubscribeAll() error {
 // after WS RPC unsubscription request is completed. Until then the subscriber channel
 // may still receive WS notifications.
 func (c *WSClient) performUnsubscription(id string) error {
-       c.subscriptionsLock.RLock()
-       rcvrWas, ok := c.subscriptions[id]
-       c.subscriptionsLock.RUnlock()
-
-       if !ok {
-               return errors.New("no subscription with this ID")
-       }
+       c.subscriptionsOrderLock.Lock()
+       defer c.subscriptionsOrderLock.Unlock()
 
        var resp bool
        if err := c.performRequest("unsubscribe", []any{id}, &resp); err != nil {
@@ -996,14 +995,6 @@ func (c *WSClient) performUnsubscription(id string) error {
                return errors.New("no subscription with this ID")
        }
 
-       cleanUpSubscriptions := true
-       if rcvrWas.Receiver() != rcvr.Receiver() {
-               // concurrent subscription has been done and been overwritten; this
-               // is not this routine's subscription, cleanup only receivers map
-               rcvr = rcvrWas
-               cleanUpSubscriptions = false
-       }
-
        ch := rcvr.Receiver()
        ids := c.receivers[ch]
        for i, rcvrID := range ids {
@@ -1017,9 +1008,7 @@ func (c *WSClient) performUnsubscription(id string) error {
        } else {
                c.receivers[ch] = ids
        }
-       if cleanUpSubscriptions {
-               delete(c.subscriptions, id)
-       }
+       delete(c.subscriptions, id)
        return nil
 }

[roman-khimov]

Let's go with the simplest solution for now (aka mutex), we can get back to the wsReader one later if needed (it can handle concurrent sub/unsub requests more efficiently).

[AnnaShaleva]

Ready for review.

[AnnaShaleva]

Linter is failing, but not sue to changes in this PR:

Error: Failed to run: Error: Command failed: /home/runner/golangci-lint-1.64.8-linux-amd64/golangci-lint config verify --config=/home/runner/work/neo-go/neo-go/nspcc-gh/.golangci.yml
Failed executing command with error: [../../nspcc-gh/.golangci.yml] validate: compile schema: failing loading "https://golangci-lint.run/jsonschema/golangci.v1.64.jsonschema.json": Get "https://golangci-lint.run/jsonschema/golangci.v1.64.jsonschema.json": context deadline exceeded (Client.Timeout exceeded while awaiting headers)
, Error: Command failed: /home/runner/golangci-lint-1.64.8-linux-amd64/golangci-lint config verify --config=/home/runner/work/neo-go/neo-go/nspcc-gh/.golangci.yml
Failed executing command with error: [../../nspcc-gh/.golangci.yml] validate: compile schema: failing loading "https://golangci-lint.run/jsonschema/golangci.v1.64.jsonschema.json": Get "https://golangci-lint.run/jsonschema/golangci.v1.64.jsonschema.json": context deadline exceeded (Client.Timeout exceeded while awaiting headers)

    at genericNodeError (node:internal/errors:984:15)
    at wrappedFn (node:internal/errors:538:14)
    at ChildProcess.exithandler (node:child_process:422:12)
    at ChildProcess.emit (node:events:524:28)
    at maybeClose (node:internal/child_process:1104:16)
    at ChildProcess._handle.onexit (node:internal/child_process:304:5)
Error: Command failed: /home/runner/golangci-lint-1.64.8-linux-amd64/golangci-lint config verify --config=/home/runner/work/neo-go/neo-go/nspcc-gh/.golangci.yml
Failed executing command with error: [../../nspcc-gh/.golangci.yml] validate: compile schema: failing loading "https://golangci-lint.run/jsonschema/golangci.v1.64.jsonschema.json": Get "https://golangci-lint.run/jsonschema/golangci.v1.64.jsonschema.json": context deadline exceeded (Client.Timeout exceeded while awaiting headers)

[Copilot]

Pull Request Overview

This PR introduces a sequential processing mechanism for WebSocket subscription and unsubscription requests to avoid server-side subscription ID conflicts.

• Added subscriptionsOrderLock in WSClient to manage ordered subscribe/unsubscribe processing.
• Wrapped performSubscription and performUnsubscription operations with the new mutex for sequential execution.
• Removed redundant receiver comparison logic in performUnsubscription.

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 82.67%. Comparing base (e76ef6f) to head (80c8c4f).
Report is 2 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #3893      +/-   ##
==========================================
+ Coverage   82.66%   82.67%   +0.01%     
==========================================
  Files         342      342              
  Lines       48349    48347       -2     
==========================================
+ Hits        39967    39973       +6     
+ Misses       6732     6726       -6     
+ Partials     1650     1648       -2     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.