fix: prevent data loss during cross-version rolling upgrade

cmd/helper/commands/cluster/shutdown.go

@@ -203,12 +203,33 @@ func Shutdown(ctx context.Context, c *cli.Context, client *kubernetes.Clientset,
 	time.Sleep(time.Second * 10)
 
 	logger.Info("do shutdown node")
-	if _, err = valkeyClient.Do(ctx, "SHUTDOWN"); err != nil && !errors.Is(err, io.EOF) {
+	if err = shutdownNode(ctx, valkeyClient, logger); err != nil {
 		logger.Error(err, "graceful shutdown failed")
 	}
 	return nil
 }
 
+// shutdownNode re-checks dbsize before issuing SHUTDOWN.
+// If dbsize == 0 (e.g. memory was cleared by a failed cross-version fullsync),
+// SHUTDOWN NOSAVE is used to preserve the existing dump.rdb on disk.
+// Otherwise, normal SHUTDOWN is issued to persist the current dataset.
+func shutdownNode(ctx context.Context, valkeyClient valkey.ValkeyClient, logger logr.Logger) error {
+	dbsize, _ := valkey.Int64(valkeyClient.Do(ctx, "DBSIZE"))
+	if dbsize == 0 {
+		logger.Info("dbsize is 0, using SHUTDOWN NOSAVE to preserve existing dump.rdb")
+		_, err := valkeyClient.Do(ctx, "SHUTDOWN", "NOSAVE")
+		if err != nil && !errors.Is(err, io.EOF) {
+			return err
+		}
+		return nil
+	}
+	_, err := valkeyClient.Do(ctx, "SHUTDOWN")
+	if err != nil && !errors.Is(err, io.EOF) {
+		return err
+	}
+	return nil
+}
+
 func getPodAccessAddr(pod *v1.Pod) string {
 	addr := net.JoinHostPort(pod.Status.PodIP, "6379")
 	ipFamilyPrefer := os.Getenv("IP_FAMILY_PREFER")

cmd/helper/commands/cluster/shutdown_test.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2024 chideat.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cluster
+
+import (
+	"context"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/chideat/valkey-operator/pkg/valkey"
+	"github.com/go-logr/logr"
+)
+
+// Test_shutdownNode verifies the SHUTDOWN/SHUTDOWN NOSAVE decision in shutdownNode.
+// miniredis doesn't support SHUTDOWN, so both cases return an error — but each exercises
+// the correct branch (NOSAVE for dbsize==0, normal SHUTDOWN for dbsize>0).
+func Test_shutdownNode(t *testing.T) {
+	tests := []struct {
+		name    string
+		setup   func(s *miniredis.Miniredis)
+		wantErr bool
+	}{
+		{
+			name:    "empty database uses SHUTDOWN NOSAVE",
+			setup:   func(_ *miniredis.Miniredis) {},
+			wantErr: true, // miniredis returns ERR for SHUTDOWN NOSAVE
+		},
+		{
+			name: "non-empty database uses normal SHUTDOWN",
+			setup: func(s *miniredis.Miniredis) {
+				_ = s.Set("key1", "value1")
+				_ = s.Set("key2", "value2")
+			},
+			wantErr: true, // miniredis returns ERR for SHUTDOWN
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := miniredis.RunT(t)
+			tt.setup(s)
+			client := valkey.NewValkeyClient(s.Addr(), valkey.AuthInfo{})
+			defer client.Close()
+
+			err := shutdownNode(context.Background(), client, logr.Discard())
+			if (err != nil) != tt.wantErr {
+				t.Errorf("shutdownNode() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}

cmd/helper/commands/failover/shutdown.go

@@ -365,10 +365,32 @@ func Shutdown(ctx context.Context, c *cli.Context, client *kubernetes.Clientset,
 	time.Sleep(time.Second * 30)
 
 	logger.Info("do shutdown node")
+	if shutdownErr := shutdownNode(ctx, valkeyClient, logger); shutdownErr != nil {
+		logger.Error(shutdownErr, "graceful shutdown failed")
+	}
+	return err
+}
+
+// shutdownNode re-checks dbsize before issuing SHUTDOWN.
+// If dbsize == 0 (e.g. memory was cleared by a failed cross-version fullsync),
+// SHUTDOWN NOSAVE is used to preserve the existing dump.rdb on disk.
+// Otherwise, normal SHUTDOWN is issued to persist the current dataset.
+// Uses a 300s timeout to accommodate large RDB snapshots.
+func shutdownNode(ctx context.Context, valkeyClient valkey.ValkeyClient, logger logr.Logger) error {
+	dbsize, _ := valkey.Int64(valkeyClient.Do(ctx, "DBSIZE"))
+	if dbsize == 0 {
+		logger.Info("dbsize is 0, using SHUTDOWN NOSAVE to preserve existing dump.rdb")
+		_, err := valkeyClient.DoWithTimeout(ctx, time.Second*300, "SHUTDOWN", "NOSAVE")
+		if err != nil && !errors.Is(err, io.EOF) {
+			return err
+		}
+		return nil
+	}
 	// NOTE: here set timeout to 300s, which will try best to do a shutdown snapshot
 	// if the data is too large, this snapshot may not be completed
-	if _, err := valkeyClient.DoWithTimeout(ctx, time.Second*300, "SHUTDOWN"); err != nil && !errors.Is(err, io.EOF) {
-		logger.Error(err, "graceful shutdown failed")
+	_, err := valkeyClient.DoWithTimeout(ctx, time.Second*300, "SHUTDOWN")
+	if err != nil && !errors.Is(err, io.EOF) {
+		return err
 	}
-	return err
+	return nil
 }

cmd/helper/commands/failover/shutdown_test.go

@@ -17,9 +17,12 @@ limitations under the License.
 package failover
 
 import (
+	"context"
 	"os"
 	"testing"
 
+	"github.com/alicebob/miniredis/v2"
+	"github.com/chideat/valkey-operator/pkg/valkey"
 	"github.com/go-logr/logr"
 )
 
@@ -76,3 +79,42 @@ replica-announce-port 31095`,
 		})
 	}
 }
+
+// Test_shutdownNode verifies the SHUTDOWN/SHUTDOWN NOSAVE decision in shutdownNode.
+// miniredis doesn't support SHUTDOWN, so both cases return an error — but each exercises
+// the correct branch (NOSAVE for dbsize==0, normal SHUTDOWN for dbsize>0).
+func Test_shutdownNode(t *testing.T) {
+	tests := []struct {
+		name    string
+		setup   func(s *miniredis.Miniredis)
+		wantErr bool
+	}{
+		{
+			name:    "empty database uses SHUTDOWN NOSAVE",
+			setup:   func(_ *miniredis.Miniredis) {},
+			wantErr: true, // miniredis returns ERR for SHUTDOWN NOSAVE
+		},
+		{
+			name: "non-empty database uses normal SHUTDOWN",
+			setup: func(s *miniredis.Miniredis) {
+				_ = s.Set("key1", "value1")
+				_ = s.Set("key2", "value2")
+			},
+			wantErr: true, // miniredis returns ERR for SHUTDOWN
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := miniredis.RunT(t)
+			tt.setup(s)
+			client := valkey.NewValkeyClient(s.Addr(), valkey.AuthInfo{})
+			defer client.Close()
+
+			err := shutdownNode(context.Background(), client, logr.Discard())
+			if (err != nil) != tt.wantErr {
+				t.Errorf("shutdownNode() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}