[#706] morph: Use backoff strategy for UnsubscribeAll()

* Sometimes the morph client cannot unsubscribe from events
  because websocket client may be got down and neo-go will
  never the request for unsubscription.

Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>

cmd/frostfs-node/netmap.go

@@ -282,11 +282,62 @@ func initNetmapState(c *cfg) {
 	c.handleLocalNodeInfo(ni)
 }
 
-func sameNodeInfo(a, b *netmapSDK.NodeInfo) bool {
+func needsUpdate(local, remote *netmapSDK.NodeInfo) bool {
-	// Suboptimal, but we do this once on the node startup.
+	return bytes.Equal(local.PublicKey(), remote.PublicKey()) && equalEndpoints(local, remote) && equalAttributes(local, remote)
-	rawA := a.Marshal()
+}
-	rawB := b.Marshal()
+
-	return bytes.Equal(rawA, rawB)
+func equalAttributes(local, remote *netmapSDK.NodeInfo) bool {
+	asA := make(map[string]string)
+	local.IterateAttributes(func(k, v string) {
+		asA[k] = v
+	})
+
+	allMatched := true
+	count := 0
+	remote.IterateAttributes(func(k, vb string) {
+		// IR adds new attributes derived from the locode, they should be skipped.
+		if isLocodeAttribute(k) {
+			return
+		}
+		if va, ok := asA[k]; !ok || va != vb {
+			allMatched = false
+			return
+		}
+		count++
+	})
+	return allMatched && count == len(asA)
+}
+
+func isLocodeAttribute(k string) bool {
+	// See https://git.frostfs.info/TrueCloudLab/frostfs-api/src/branch/master/netmap/types.proto#L171
+	switch k {
+	case "Continent", "Country", "CountryCode", "Location", "SubDiv", "SubDivCode":
+		return true
+	default:
+		return false
+	}
+}
+
+func equalEndpoints(a, b *netmapSDK.NodeInfo) bool {
+	var esA, esB []string
+	a.IterateNetworkEndpoints(func(e string) bool {
+		esA = append(esA, e)
+		return false
+	})
+	b.IterateNetworkEndpoints(func(e string) bool {
+		esB = append(esB, e)
+		return false
+	})
+
+	if len(esA) != len(esB) {
+		return false
+	}
+	for i := range esA {
+		if esA[i] != esB[i] {
+			return false
+		}
+	}
+	return true
 }
 
 func nodeState(ni *netmapSDK.NodeInfo) string {
@@ -314,7 +365,7 @@ func (c *cfg) netmapInitLocalNodeState(epoch uint64) (*netmapSDK.NodeInfo, bool,
 	for i := range nmNodes {
 		if bytes.Equal(nmNodes[i].PublicKey(), c.binPublicKey) {
 			candidate = &nmNodes[i]
-			alreadyBootstraped = candidate.IsOnline() && sameNodeInfo(&c.cfgNodeInfo.localInfo, candidate)
+			alreadyBootstraped = candidate.IsOnline() && needsUpdate(&c.cfgNodeInfo.localInfo, candidate)
 			break
 		}
 	}

go.mod

@@ -6,9 +6,10 @@ require (
 	git.frostfs.info/TrueCloudLab/frostfs-api-go/v2 v2.16.0
 	git.frostfs.info/TrueCloudLab/frostfs-contract v0.18.0
 	git.frostfs.info/TrueCloudLab/frostfs-observability v0.0.0-20230531082742-c97d21411eb6
-	git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20230911122224-ac8fc6d4400c
+	git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20230915114754-555ccc63b255
 	git.frostfs.info/TrueCloudLab/hrw v1.2.1
 	git.frostfs.info/TrueCloudLab/tzhash v1.8.0
+	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cheggaaa/pb v1.0.29
 	github.com/chzyer/readline v1.5.1
 	github.com/dgraph-io/ristretto v0.1.1

internal/logs/logs.go

@@ -195,6 +195,8 @@ const (
 	SubscriberCantCastBlockEventValueToBlock                                = "can't cast block event value to block"
 	SubscriberCantCastNotifyEventValueToTheNotaryRequestStruct              = "can't cast notify event value to the notary request struct"
 	SubscriberUnsupportedNotificationFromTheChain                           = "unsupported notification from the chain"
+	SubscriberCouldNotSwitchRPCDuringUnsubscriptionFromEvents               = "could not switch rpc during the unsubscription from events"
+	SubscriberCouldNotUnsubscribeFromEventsOnBackoffPolicy                  = "could not unsubscribe from events on backoff policy"
 	BlobovniczaCreatingDirectoryForBoltDB                                   = "creating directory for BoltDB"
 	BlobovniczaOpeningBoltDB                                                = "opening BoltDB"
 	BlobovniczaInitializing                                                 = "initializing..."

pkg/morph/client/client.go

@@ -74,6 +74,9 @@ type Client struct {
 	// channel for internal stop
 	closeChan chan struct{}
 
+	// channel to indicate that close is done
+	closeDone chan struct{}
+
 	// indicates that Client is not able to
 	// establish connection to any of the
 	// provided RPC endpoints

pkg/morph/client/constructor.go

@@ -120,6 +120,7 @@ func New(ctx context.Context, key *keys.PrivateKey, opts ...Option) (*Client, er
 		accAddr:   accAddr,
 		cfg:       *cfg,
 		closeChan: make(chan struct{}),
+		closeDone: make(chan struct{}),
 	}
 
 	cli.endpoints.init(cfg.endpoints)
@@ -169,6 +170,7 @@ func (c *Client) newCli(ctx context.Context, endpoint string) (*rpcclient.WSClie
 		Options: rpcclient.Options{
 			DialTimeout: c.cfg.dialTimeout,
 		},
+		CloseNotificationChannelIfFull: true,
 	})
 	if err != nil {
 		return nil, nil, fmt.Errorf("WS client creation: %w", err)

pkg/morph/client/multi.go

@@ -79,8 +79,10 @@ func (c *Client) closeWaiter(ctx context.Context) {
 	case <-ctx.Done():
 	case <-c.closeChan:
 	}
+	//nolint:contextcheck
 	_ = c.UnsubscribeAll()
 	c.close()
+	close(c.closeDone)
 }
 
 func (c *Client) switchToMostPrioritized(ctx context.Context) {

pkg/morph/client/notifications.go

@@ -1,11 +1,18 @@
 package client
 
 import (
+	"context"
+	"fmt"
+	"time"
+
+	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
+	"github.com/cenkalti/backoff"
 	"github.com/nspcc-dev/neo-go/pkg/core/block"
 	"github.com/nspcc-dev/neo-go/pkg/core/state"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc/result"
 	"github.com/nspcc-dev/neo-go/pkg/util"
+	"go.uber.org/zap"
 )
 
 // Close closes connection to the remote side making
@@ -17,6 +24,11 @@ func (c *Client) Close() {
 	// to prevent switching to another RPC node
 	// in the notification loop
 	close(c.closeChan)
+
+	// closeWaiter performs asynchronously and thus
+	// we may abrupt all process related to close process
+	// if we do not wait for closing process finish.
+	<-c.closeDone
 }
 
 // ReceiveExecutionNotifications performs subscription for notifications
@@ -100,6 +112,27 @@ func (c *Client) UnsubscribeAll() error {
 		return ErrConnectionLost
 	}
 
-	err := c.client.UnsubscribeAll()
+	if err := c.client.UnsubscribeAll(); err != nil {
-	return err
+		// TODO (aarifullin): consider the situation when the morph client
+		// failed to subscribe for events because of websocket client problems
+		// "under hood". After failed subscription the client invokes Close()
+		// that invokes UnsubscribeAll(). This requires to push new request
+		// via websocket to neo-go but the websocket client may be down.
+		// Therefore, morph will never request neo-go to unsubscribe from events, but
+		// we can try to fix this by reconnecting to neo-go.
+		backoffSettings := backoff.NewExponentialBackOff()
+		backoffSettings.MaxElapsedTime = 30 * time.Second
+		return backoff.Retry(func() error {
+			if !c.SwitchRPC(context.TODO()) {
+				c.logger.Warn(logs.SubscriberCouldNotSwitchRPCDuringUnsubscriptionFromEvents)
+				return fmt.Errorf("could not switch rpc")
+			}
+			err := c.client.UnsubscribeAll()
+			if err != nil {
+				c.logger.Warn(logs.SubscriberCouldNotUnsubscribeFromEventsOnBackoffPolicy, zap.Error(err))
+			}
+			return err
+		}, backoffSettings)
+	}
+	return nil
 }