snapshotoffload: integrate publication and gated retention runtime

cmd/elastickv-snapshot-offload/main.go

@@ -10,6 +10,7 @@ import (
 	"log/slog"
 	"os"
 	"sort"
+	"strconv"
 	"strings"
 
 	"github.com/bootjp/elastickv/internal/raftengine/etcd"
@@ -59,6 +60,13 @@ type restoreConfig struct {
 	manifestKey string
 	dataDir     string
 	peerCSV     string
+	// expectGroupRaw is the flag text; empty means the operator did not
+	// supply one, which is an error. Parsed into expectGroupID because
+	// group 0 is a real group and cannot double as "unset".
+	expectGroupRaw               string
+	expectGroupID                uint64
+	expectCluster                string
+	allowLegacySourceClusterless bool
 }
 
 func main() {
@@ -123,7 +131,7 @@ func parsePublishFlags(argv []string) (*publishConfig, error) {
 	fs.StringVar(&cfg.dataDir, "data-dir", "", "Source raft data directory containing a persisted snapshot (required)")
 	fs.StringVar(&cfg.prefix, "prefix", "", "Object key prefix for published snapshot objects")
 	fs.Uint64Var(&cfg.groupID, "group-id", 0, "Raft group ID recorded in the manifest")
-	fs.StringVar(&cfg.sourceCluster, "source-cluster", "", "Source cluster identifier (required for group 0 manifests)")
+	fs.StringVar(&cfg.sourceCluster, "source-cluster", "", "Source cluster identifier (required; restore matches it against the manifest)")
 	fs.StringVar(&cfg.binaryVersion, "binary-version", "", "Binary version recorded in the manifest")
 	fs.StringVar(&cfg.spoolDir, "spool-dir", "", "Temporary spool directory for the payload stream")
 	if err := fs.Parse(argv); err != nil {
@@ -135,8 +143,14 @@ func parsePublishFlags(argv []string) (*publishConfig, error) {
 	if strings.TrimSpace(cfg.dataDir) == "" {
 		return nil, errors.New("--data-dir is required")
 	}
-	if cfg.groupID == 0 && strings.TrimSpace(cfg.sourceCluster) == "" {
-		return nil, errors.New("--source-cluster is required when --group-id is 0")
+	// Required for EVERY group, not just group 0. restore rejects an empty
+	// --expect-source-cluster outright and then compares it against the
+	// manifest's own value, so a manifest published without one can never
+	// match: a nonzero-group backup taken with this CLI would be
+	// unrestorable by it. Relaxing the restore side instead would give back
+	// the wrong-cluster restore that check exists to prevent.
+	if strings.TrimSpace(cfg.sourceCluster) == "" {
+		return nil, errors.New("--source-cluster is required")
 	}
 	if err := validateStoreFlags(cfg.store); err != nil {
 		return nil, err
@@ -152,6 +166,16 @@ func parseRestoreFlags(argv []string) (*restoreConfig, error) {
 	fs.StringVar(&cfg.manifestKey, "manifest-key", "", "Object key of the snapshot manifest to restore (required)")
 	fs.StringVar(&cfg.dataDir, "data-dir", "", "Fresh target raft data directory to create (required; must not already exist)")
 	fs.StringVar(&cfg.peerCSV, "peers", "", "Comma-separated raft peers id=addr,id=addr (required)")
+	fs.StringVar(&cfg.expectCluster, "expect-source-cluster", "",
+		"Source cluster this data dir is for (required). The restore is refused if the "+
+			"manifest was published by a different cluster -- which the group check alone "+
+			"cannot catch when one bucket holds backups from several clusters.")
+	fs.StringVar(&cfg.expectGroupRaw, "expect-group", "",
+		"Raft group id this data dir is for (required). The restore is refused if the manifest "+
+			"belongs to a different group, which is otherwise undetectable: nothing downstream "+
+			"records the group, so the wrong group's FSM would start under this group's identity.")
+	fs.BoolVar(&cfg.allowLegacySourceClusterless, "allow-legacy-source-clusterless", false,
+		"allow an older nonzero-group manifest without source_cluster; source identity cannot be verified")
 	if err := fs.Parse(argv); err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -167,6 +191,17 @@ func parseRestoreFlags(argv []string) (*restoreConfig, error) {
 	if strings.TrimSpace(cfg.peerCSV) == "" {
 		return nil, errors.New("--peers is required")
 	}
+	if strings.TrimSpace(cfg.expectGroupRaw) == "" {
+		return nil, errors.New("--expect-group is required")
+	}
+	if strings.TrimSpace(cfg.expectCluster) == "" {
+		return nil, errors.New("--expect-source-cluster is required")
+	}
+	groupID, err := strconv.ParseUint(strings.TrimSpace(cfg.expectGroupRaw), 10, 64)
+	if err != nil {
+		return nil, errors.Wrapf(err, "parse --expect-group %q", cfg.expectGroupRaw)
+	}
+	cfg.expectGroupID = groupID
 	if err := validateStoreFlags(cfg.store); err != nil {
 		return nil, err
 	}
@@ -263,11 +298,17 @@ func runRestore(ctx context.Context, cfg *restoreConfig, logger *slog.Logger) (r
 	if err != nil {
 		return err
 	}
+	if cfg.allowLegacySourceClusterless {
+		logger.Warn("restoring with legacy source-cluster compatibility; manifest source identity may be unverifiable")
+	}
 	result, err := snapshotoffload.RestorePhysicalSnapshot(ctx, snapshotoffload.RestoreOptions{
-		Store:       store,
-		ManifestKey: cfg.manifestKey,
-		DataDir:     cfg.dataDir,
-		Peers:       peers,
+		Store:                        store,
+		ManifestKey:                  cfg.manifestKey,
+		DataDir:                      cfg.dataDir,
+		Peers:                        peers,
+		ExpectGroupID:                &cfg.expectGroupID,
+		ExpectSourceCluster:          cfg.expectCluster,
+		AllowLegacySourceClusterless: cfg.allowLegacySourceClusterless,
 	})
 	if err != nil {
 		return errors.Wrap(err, "restore physical snapshot")

cmd/elastickv-snapshot-offload/main_test.go

@@ -50,6 +50,10 @@ func TestSnapshotOffloadCLIPublishAndRestoreLocal(t *testing.T) {
 		"--manifest-key", manifest.ManifestKey,
 		"--data-dir", restoreDataDir,
 		"--peers", "n2=127.0.0.1:12002",
+		// The publish above used --group-id 2, so this is the group this
+		// data dir is for.
+		"--expect-group", "2",
+		"--expect-source-cluster", "cluster-cli",
 	}, io.Discard, logger)
 	require.NoError(t, err)
 	require.Equal(t, exitSuccess, code)
@@ -67,6 +71,7 @@ func TestSnapshotOffloadCLIRequiresLocalRoot(t *testing.T) {
 		"--store", storeLocal,
 		"--data-dir", "data",
 		"--group-id", "1",
+		"--source-cluster", "prod-tokyo",
 	}, io.Discard, slog.New(slog.NewTextHandler(io.Discard, nil)))
 	require.ErrorContains(t, err, "--local-root is required")
 	require.Equal(t, exitUserErr, code)
@@ -101,6 +106,7 @@ func TestSnapshotOffloadCLIS3KMSRequiresAWSKMS(t *testing.T) {
 		"--s3-kms-key-id", "key-id",
 		"--data-dir", "data",
 		"--group-id", "1",
+		"--source-cluster", "prod-tokyo",
 	})
 	require.NoError(t, err)
 
@@ -111,6 +117,7 @@ func TestSnapshotOffloadCLIS3KMSRequiresAWSKMS(t *testing.T) {
 		"--s3-kms-key-id", "key-id",
 		"--data-dir", "data",
 		"--group-id", "1",
+		"--source-cluster", "prod-tokyo",
 	})
 	require.ErrorContains(t, err, "s3 KMS key id requires aws:kms encryption")
 }
@@ -168,6 +175,14 @@ func TestClassifyErrorKeepsMissingSnapshotAsADataError(t *testing.T) {
 			err:  errors.Wrap(snapshotoffload.ErrInvalidOptions, "publish"),
 			want: exitUserErr,
 		},
+		{
+			// The snapshot is intact; the operator named the wrong
+			// manifest for this data dir. Automation keys off the
+			// difference, so this must not be reported as bad data.
+			name: "wrong group's manifest",
+			err:  errors.Wrap(snapshotoffload.ErrRestoreGroupMismatch, "restore"),
+			want: exitUserErr,
+		},
 	}
 
 	for _, tc := range tests {
@@ -177,3 +192,238 @@ func TestClassifyErrorKeepsMissingSnapshotAsADataError(t *testing.T) {
 		})
 	}
 }
+
+// TestSnapshotOffloadCLIRestoreRefusesAnotherGroupsManifest is the regression
+// test for a silent mis-restore.
+//
+// Nothing downstream of the restore records which group the data belongs to:
+// the prepared artifacts carry index, term, peers and payload hash, and startup
+// derives the group from the directory layout. So an operator repeating this
+// command across groups and pasting the wrong manifest key produced a
+// valid-looking directory that startup then loaded under a different group's
+// routing identity, with no error at any point. --expect-group is the only
+// place that mistake is detectable.
+func TestSnapshotOffloadCLIRestoreRefusesAnotherGroupsManifest(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	root := t.TempDir()
+	objectRoot := filepath.Join(root, "objects")
+	sourceDataDir := seedCLISnapshot(t, root, []byte("EKVTHLC1group-2-payload"), 60, 9)
+
+	var stdout bytes.Buffer
+	code, err := run(ctx, []string{
+		commandPublish,
+		"--store", storeLocal,
+		"--local-root", objectRoot,
+		"--data-dir", sourceDataDir,
+		"--prefix", "cluster-cli",
+		"--group-id", "2",
+		"--source-cluster", "cluster-cli",
+		"--binary-version", "test-version",
+	}, &stdout, logger)
+	require.NoError(t, err)
+	require.Equal(t, exitSuccess, code)
+
+	manifest, err := snapshotoffload.DecodeManifest(stdout.Bytes())
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), manifest.GroupID)
+
+	restoreDataDir := filepath.Join(root, "restored-as-group-1")
+	code, err = run(ctx, []string{
+		commandRestore,
+		"--store", storeLocal,
+		"--local-root", objectRoot,
+		"--manifest-key", manifest.ManifestKey,
+		"--data-dir", restoreDataDir,
+		"--peers", "n1=127.0.0.1:12001",
+		// The operator means group 1, but pasted group 2's manifest key.
+		"--expect-group", "1",
+		"--expect-source-cluster", "cluster-cli",
+	}, io.Discard, logger)
+	require.Error(t, err)
+	require.ErrorIs(t, err, snapshotoffload.ErrRestoreGroupMismatch)
+	// exitUserErr, not exitDataErr: the snapshot data is fine, the
+	// invocation named the wrong manifest. Automation distinguishes the two.
+	require.Equal(t, exitUserErr, code)
+
+	// Nothing may be left behind: the check runs before the download and
+	// before the destination is created, so a mistaken key costs nothing.
+	_, statErr := os.Stat(restoreDataDir)
+	require.True(t, os.IsNotExist(statErr),
+		"a refused restore must not create the destination")
+}
+
+// --expect-group is required, because defaulting it would silently accept
+// whatever group the manifest happens to name -- which is the behaviour the
+// flag exists to remove.
+func TestSnapshotOffloadCLIRestoreRequiresAnExpectedGroup(t *testing.T) {
+	t.Parallel()
+
+	_, err := parseRestoreFlags([]string{
+		"--store", storeLocal,
+		"--local-root", "/tmp/objects",
+		"--manifest-key", "k",
+		"--data-dir", "/tmp/restored",
+		"--peers", "n1=127.0.0.1:12001",
+	})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "--expect-group is required")
+}
+
+// Group 0 is the dedicated TSO group, so it has to be accepted as an explicit
+// value rather than treated as "unset".
+func TestSnapshotOffloadCLIRestoreAcceptsGroupZero(t *testing.T) {
+	t.Parallel()
+
+	cfg, err := parseRestoreFlags([]string{
+		"--store", storeLocal,
+		"--local-root", "/tmp/objects",
+		"--manifest-key", "k",
+		"--data-dir", "/tmp/restored",
+		"--peers", "n1=127.0.0.1:12001",
+		"--expect-group", "0",
+		"--expect-source-cluster", "cluster-cli",
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint64(0), cfg.expectGroupID)
+}
+
+func TestSnapshotOffloadCLIRestoreRejectsANonNumericGroup(t *testing.T) {
+	t.Parallel()
+
+	_, err := parseRestoreFlags([]string{
+		"--store", storeLocal,
+		"--local-root", "/tmp/objects",
+		"--manifest-key", "k",
+		"--data-dir", "/tmp/restored",
+		"--peers", "n1=127.0.0.1:12001",
+		"--expect-group", "one",
+		"--expect-source-cluster", "cluster-cli",
+	})
+	require.Error(t, err)
+}
+
+// TestSnapshotOffloadCLIRestoreRefusesAnotherClustersManifest covers the gap
+// --expect-group alone leaves.
+//
+// One bucket can hold backups from several clusters, even under different
+// prefixes. A manifest from another cluster for the SAME group id passes the
+// group check, and the restore then produces a structurally valid directory
+// that startup loads as this cluster's FSM. Nothing downstream records the
+// source cluster, so this is the only place the mistake is detectable.
+func TestSnapshotOffloadCLIRestoreRefusesAnotherClustersManifest(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	root := t.TempDir()
+	objectRoot := filepath.Join(root, "objects")
+	sourceDataDir := seedCLISnapshot(t, root, []byte("EKVTHLC1other-cluster-payload"), 60, 9)
+
+	var stdout bytes.Buffer
+	code, err := run(ctx, []string{
+		commandPublish,
+		"--store", storeLocal,
+		"--local-root", objectRoot,
+		"--data-dir", sourceDataDir,
+		"--prefix", "shared-bucket",
+		"--group-id", "1",
+		"--source-cluster", "cluster-b",
+		"--binary-version", "test-version",
+	}, &stdout, logger)
+	require.NoError(t, err)
+	require.Equal(t, exitSuccess, code)
+
+	manifest, err := snapshotoffload.DecodeManifest(stdout.Bytes())
+	require.NoError(t, err)
+	require.Equal(t, "cluster-b", manifest.SourceCluster)
+	require.Equal(t, uint64(1), manifest.GroupID)
+
+	restoreDataDir := filepath.Join(root, "restored-into-cluster-a")
+	code, err = run(ctx, []string{
+		commandRestore,
+		"--store", storeLocal,
+		"--local-root", objectRoot,
+		"--manifest-key", manifest.ManifestKey,
+		"--data-dir", restoreDataDir,
+		"--peers", "n1=127.0.0.1:12001",
+		// The group matches, so only the cluster identity can catch this.
+		"--expect-group", "1",
+		"--expect-source-cluster", "cluster-a",
+	}, io.Discard, logger)
+	require.Error(t, err)
+	require.ErrorIs(t, err, snapshotoffload.ErrRestoreSourceClusterMismatch)
+	require.Equal(t, exitUserErr, code)
+
+	_, statErr := os.Stat(restoreDataDir)
+	require.True(t, os.IsNotExist(statErr),
+		"a refused restore must not create the destination")
+}
+
+func TestSnapshotOffloadCLIRestoreRequiresAnExpectedSourceCluster(t *testing.T) {
+	t.Parallel()
+
+	_, err := parseRestoreFlags([]string{
+		"--store", storeLocal,
+		"--local-root", "/tmp/objects",
+		"--manifest-key", "k",
+		"--data-dir", "/tmp/restored",
+		"--peers", "n1=127.0.0.1:12001",
+		"--expect-group", "1",
+	})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "--expect-source-cluster is required")
+}
+
+func TestSnapshotOffloadCLIParsesLegacySourceClusterlessOverride(t *testing.T) {
+	t.Parallel()
+
+	cfg, err := parseRestoreFlags([]string{
+		"--store", storeLocal,
+		"--local-root", "/tmp/objects",
+		"--manifest-key", "legacy/v1/groups/1/snapshots/manifest.json",
+		"--data-dir", "/tmp/restored",
+		"--peers", "n1=127.0.0.1:12001",
+		"--expect-group", "1",
+		"--expect-source-cluster", "cluster-a",
+		"--allow-legacy-source-clusterless",
+	})
+	require.NoError(t, err)
+	require.True(t, cfg.allowLegacySourceClusterless)
+}
+
+// TestSnapshotOffloadCLIPublishRequiresASourceClusterForEveryGroup pins that
+// publish and restore agree on this flag.
+//
+// It used to be required only for group 0. restore rejects an empty
+// --expect-source-cluster outright and then compares it against the manifest's
+// own SourceCluster, so a manifest published without one can never match: a
+// nonzero-group backup taken with this CLI was unrestorable by it. Relaxing
+// the restore side instead would give back the wrong-cluster restore that
+// check exists to prevent.
+func TestSnapshotOffloadCLIPublishRequiresASourceClusterForEveryGroup(t *testing.T) {
+	t.Parallel()
+
+	for _, group := range []string{"0", "1", "7"} {
+		_, err := parsePublishFlags([]string{
+			"--store", storeLocal,
+			"--local-root", "/tmp/objects",
+			"--data-dir", "data",
+			"--group-id", group,
+		})
+		require.ErrorContains(t, err, "--source-cluster is required",
+			"group %s must not be able to publish an unrestorable manifest", group)
+
+		cfg, err := parsePublishFlags([]string{
+			"--store", storeLocal,
+			"--local-root", "/tmp/objects",
+			"--data-dir", "data",
+			"--group-id", group,
+			"--source-cluster", "prod-tokyo",
+		})
+		require.NoError(t, err)
+		require.Equal(t, "prod-tokyo", cfg.sourceCluster)
+	}
+}