process once per client / seqid

Author: ndyakov

maintnotifications/manager.go

@@ -69,6 +69,10 @@ type Manager struct {
 	// MOVING operation tracking - using sync.Map for better concurrent performance
 	activeMovingOps sync.Map // map[MovingOperationKey]*MovingOperation
 
+	// SMIGRATED notification deduplication - tracks processed SeqIDs
+	// Multiple connections may receive the same SMIGRATED notification
+	processedSMigratedSeqIDs sync.Map // map[int64]bool
+
 	// Atomic state tracking - no locks needed for state queries
 	activeOperationCount atomic.Int64 // Number of active operations
 	closed               atomic.Bool  // Manager closed state
@@ -238,6 +242,15 @@ func (hm *Manager) GetActiveOperationCount() int64 {
 	return hm.activeOperationCount.Load()
 }
 
+// MarkSMigratedSeqIDProcessed attempts to mark a SMIGRATED SeqID as processed.
+// Returns true if this is the first time processing this SeqID (should process),
+// false if it was already processed (should skip).
+// This prevents duplicate processing when multiple connections receive the same notification.
+func (hm *Manager) MarkSMigratedSeqIDProcessed(seqID int64) bool {
+	_, alreadyProcessed := hm.processedSMigratedSeqIDs.LoadOrStore(seqID, true)
+	return !alreadyProcessed // Return true if NOT already processed
+}
+
 // Close closes the manager.
 func (hm *Manager) Close() error {
 	// Use atomic operation for thread-safe close check

maintnotifications/push_notification_handler.go

@@ -341,6 +341,7 @@ func (snh *NotificationHandler) handleSMigrating(ctx context.Context, handlerCtx
 // SMIGRATED indicates that a cluster slot has finished migrating to a different node.
 // This is a cluster-level notification that triggers cluster state reload.
 // Expected format: ["SMIGRATED", SeqID, host:port, slot1/range1-range2, ...]
+// Note: Multiple connections may receive the same notification, so we deduplicate by SeqID.
 func (snh *NotificationHandler) handleSMigrated(ctx context.Context, handlerCtx push.NotificationHandlerContext, notification []interface{}) error {
 	if len(notification) < 4 {
 		internal.Logger.Printf(ctx, logs.InvalidNotification("SMIGRATED", notification))
@@ -354,6 +355,15 @@ func (snh *NotificationHandler) handleSMigrated(ctx context.Context, handlerCtx
 		return ErrInvalidNotification
 	}
 
+	// Deduplicate by SeqID - multiple connections may receive the same notification
+	if !snh.manager.MarkSMigratedSeqIDProcessed(seqID) {
+		// Already processed this SeqID, skip
+		if internal.LogLevel.DebugOrAbove() {
+			internal.Logger.Printf(ctx, "cluster: SMIGRATED notification with SeqID %d already processed, skipping", seqID)
+		}
+		return nil
+	}
+
 	// Extract host:port (position 2)
 	hostPort, ok := notification[2].(string)
 	if !ok {

maintnotifications/smigrating_test.go

@@ -201,6 +201,64 @@ func TestSMigratedNotificationHandler(t *testing.T) {
 		}
 	})
 
+t.Run("SMigratedNotification_Deduplication", func(t *testing.T) {
+	// Track callback invocations
+	var callbackCount atomic.Int32
+
+	// Create a mock manager with callback
+	manager := &Manager{
+		clusterStateReloadCallback: func(ctx context.Context, hostPort string, slotRanges []string) {
+			callbackCount.Add(1)
+		},
+	}
+
+	// Create notification handler
+	handler := &NotificationHandler{
+		manager:           manager,
+		operationsManager: manager,
+	}
+
+	// Create SMIGRATED notification with SeqID 456
+	notification := []interface{}{"SMIGRATED", int64(456), "127.0.0.1:6379", "1234"}
+
+	ctx := context.Background()
+	handlerCtx := push.NotificationHandlerContext{}
+
+	// Handle the notification first time
+	err := handler.handleSMigrated(ctx, handlerCtx, notification)
+	if err != nil {
+		t.Errorf("handleSMigrated should not error on first call: %v", err)
+	}
+
+	// Verify callback was called once
+	if callbackCount.Load() != 1 {
+		t.Errorf("Expected callback to be called once, got %d", callbackCount.Load())
+	}
+
+	// Handle the same notification again (simulating multiple connections)
+	err = handler.handleSMigrated(ctx, handlerCtx, notification)
+	if err != nil {
+		t.Errorf("handleSMigrated should not error on second call: %v", err)
+	}
+
+	// Verify callback was NOT called again (still 1)
+	if callbackCount.Load() != 1 {
+		t.Errorf("Expected callback to be called only once (deduplication), got %d", callbackCount.Load())
+	}
+
+	// Handle a different notification with different SeqID
+	notification2 := []interface{}{"SMIGRATED", int64(789), "127.0.0.1:6380", "5678"}
+	err = handler.handleSMigrated(ctx, handlerCtx, notification2)
+	if err != nil {
+		t.Errorf("handleSMigrated should not error on third call: %v", err)
+	}
+
+	// Verify callback was called again (now 2)
+	if callbackCount.Load() != 2 {
+		t.Errorf("Expected callback to be called twice (different SeqID), got %d", callbackCount.Load())
+	}
+})
+
 	t.Run("InvalidSMigratedNotification_TooShort", func(t *testing.T) {
 		manager := &Manager{}
 		handler := &NotificationHandler{