From 3b10815988098667c7266fb4547f65975e452d27 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Tue, 8 Sep 2026 16:22:48 +0900
Subject: [PATCH 1/4] snapshotoffload: wire the scheduler into the server
 runtime
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Completes M2: the scheduler now actually runs. Opt-in via
--snapshotOffloadBucket (S3) or --snapshotOffloadLocalDir, with the §7
configuration surface for region, endpoint, profile, path style,
server-side encryption, schedule, jitter, concurrency, spool dir and
source-cluster identity.

Each local Raft group contributes its own data dir plus both leadership
callbacks. Both read the engine through snapshotEngine(): the scheduler
outlives startup and races Close(), so a direct field read would be a
data race, and a runtime whose engine has been cleared reports "not
leader" rather than publishing.

A configured-but-unbuildable offload fails startup instead of logging
and continuing. An operator who set a backup destination and silently
received no backups is worse off than one whose node refused to start.
Bucket and local dir are mutually exclusive for the same reason:
ambiguity about which destination holds the artifacts is only
discovered when a restore is attempted.

Adds the scheduler's Prometheus metrics — published/skipped/failed
counters, last-published-index gauge, publish-duration and payload-size
histograms. The failure counter deliberately carries no error label:
messages are unbounded and one recurring failure would explode the
metric's cardinality. Skip reasons are normalized into the scheduler's
closed set.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 ...artial_physical_snapshot_object_offload.md |   6 +-
 main.go                                       |  10 +
 main_snapshot_offload.go                      | 210 ++++++++++++++++++
 main_snapshot_offload_test.go                 | 125 +++++++++++
 monitoring/registry.go                        |  12 +
 monitoring/snapshot_offload.go                | 157 +++++++++++++
 monitoring/snapshot_offload_test.go           |  96 ++++++++
 7 files changed, 615 insertions(+), 1 deletion(-)
 create mode 100644 main_snapshot_offload.go
 create mode 100644 main_snapshot_offload_test.go
 create mode 100644 monitoring/snapshot_offload.go
 create mode 100644 monitoring/snapshot_offload_test.go

diff --git a/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md b/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md
index 2956620aa..f338b0cca 100644
--- a/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md
+++ b/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md
@@ -48,7 +48,11 @@ The M1 object-store-neutral substrate now adds:
 - `cmd/elastickv-snapshot-offload publish` and `restore` for local and
   S3-backed operator workflows.
 
-The runtime scheduler and retention/GC remain pending.
+The runtime scheduler is implemented and wired into main.go, opt-in via
+`--snapshotOffloadBucket` (or `--snapshotOffloadLocalDir`). Retention/GC
+is implemented per §5. Restore drills and corruption tests are in place;
+multi-node acceptance, operator documentation, and the §7
+versioned-bucket decision remain pending.
 
 ## 2. Safety boundary
 
diff --git a/main.go b/main.go
index 87aae6b0c..f284e6082 100644
--- a/main.go
+++ b/main.go
@@ -805,6 +805,16 @@ func startDistributionStartup(in distributionStartupInput) (distributionStartup,
 	}
 	startMonitoringCollectors(in.ctx, in.metricsRegistry, in.runtimes, in.clock)
 	startFSMCompactorIfEnabled(in.ctx, in.eg, in.runtimes, in.readTracker)
+	// §4 physical snapshot offload. Opt-in, and a hard error when
+	// configured-but-unbuildable: an operator who set a backup
+	// destination and silently got no backups is worse off than one
+	// whose node refused to start.
+	if err := startSnapshotOffload(
+		in.ctx, in.eg, in.runtimes, *raftDir, in.raftID, in.cfg.multi,
+		in.metricsRegistry.SnapshotOffloadObserver(), slog.Default(),
+	); err != nil {
+		return distributionStartup{}, err
+	}
 	return distributionStartup{
 		defaultRuntime:   defaultRuntime,
 		distServer:       distServer,
diff --git a/main_snapshot_offload.go b/main_snapshot_offload.go
new file mode 100644
index 000000000..be9ef13e3
--- /dev/null
+++ b/main_snapshot_offload.go
@@ -0,0 +1,210 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log/slog"
+	"strings"
+
+	"github.com/bootjp/elastickv/internal/raftengine"
+	"github.com/bootjp/elastickv/internal/snapshotoffload"
+	"github.com/cockroachdb/errors"
+	"golang.org/x/sync/errgroup"
+)
+
+// Physical snapshot object offload (design doc §4 / §7). Opt-in: the
+// whole subsystem stays dormant unless --snapshotOffloadBucket (S3) or
+// --snapshotOffloadLocalDir (filesystem) is set.
+//
+// The design requires that only a group's current leader publishes, so
+// each group contributes both a cheap pre-check and a pre-commit
+// leadership re-verification; the scheduler bounds the latter itself.
+var (
+	snapshotOffloadBucket = flag.String("snapshotOffloadBucket", "",
+		"S3 bucket for physical snapshot offload; empty to disable")
+	snapshotOffloadLocalDir = flag.String("snapshotOffloadLocalDir", "",
+		"filesystem root for physical snapshot offload; an alternative to --snapshotOffloadBucket, mainly for testing")
+	snapshotOffloadPrefix = flag.String("snapshotOffloadPrefix", "",
+		"key prefix below which snapshot artifacts are written")
+	snapshotOffloadRegion = flag.String("snapshotOffloadRegion", "",
+		"AWS region for the snapshot offload bucket")
+	snapshotOffloadEndpoint = flag.String("snapshotOffloadEndpoint", "",
+		"custom S3 endpoint for snapshot offload; empty uses the AWS default")
+	snapshotOffloadProfile = flag.String("snapshotOffloadProfile", "",
+		"shared-credentials profile for snapshot offload")
+	snapshotOffloadForcePathStyle = flag.Bool("snapshotOffloadForcePathStyle", false,
+		"use path-style addressing for the snapshot offload endpoint")
+	snapshotOffloadSSE = flag.String("snapshotOffloadServerSideEncryption", "",
+		"server-side encryption mode for snapshot objects (AES256 or aws:kms)")
+	snapshotOffloadSSEKMSKeyID = flag.String("snapshotOffloadSSEKMSKeyId", "",
+		"KMS key ARN when --snapshotOffloadServerSideEncryption is aws:kms")
+	snapshotOffloadInterval = flag.Duration("snapshotOffloadInterval", snapshotoffload.DefaultSchedulerInterval,
+		"how often to scan local groups for a publishable snapshot")
+	snapshotOffloadJitter = flag.Duration("snapshotOffloadJitter", 0,
+		"random spread applied to the offload schedule; zero uses a quarter of the interval")
+	snapshotOffloadConcurrency = flag.Int("snapshotOffloadConcurrency", snapshotoffload.DefaultSchedulerConcurrency,
+		"maximum concurrent snapshot uploads for this process")
+	snapshotOffloadSpoolDir = flag.String("snapshotOffloadSpoolDir", "",
+		"directory for snapshot spool files; empty uses the data dir's filesystem")
+	snapshotOffloadSourceCluster = flag.String("snapshotOffloadSourceCluster", "",
+		"source cluster identity recorded in every manifest; required when offload is enabled")
+)
+
+// snapshotOffloadEnabled reports whether the operator configured a
+// destination. Checked before any other offload flag is validated so a
+// node that never opts in cannot fail startup on offload config.
+func snapshotOffloadEnabled() bool {
+	return strings.TrimSpace(*snapshotOffloadBucket) != "" ||
+		strings.TrimSpace(*snapshotOffloadLocalDir) != ""
+}
+
+// buildSnapshotOffloadStore constructs the configured object store.
+//
+// Bucket and local dir are mutually exclusive: accepting both would
+// leave which destination actually receives the artifacts ambiguous,
+// and a backup written to the wrong place is discovered only when a
+// restore is attempted.
+func buildSnapshotOffloadStore(ctx context.Context) (snapshotoffload.ObjectStore, error) {
+	bucket := strings.TrimSpace(*snapshotOffloadBucket)
+	localDir := strings.TrimSpace(*snapshotOffloadLocalDir)
+	if bucket != "" && localDir != "" {
+		return nil, errors.Wrap(snapshotoffload.ErrInvalidOptions,
+			"--snapshotOffloadBucket and --snapshotOffloadLocalDir are mutually exclusive")
+	}
+	if localDir != "" {
+		store, err := snapshotoffload.NewLocalStore(localDir)
+		if err != nil {
+			return nil, errors.Wrap(err, "snapshot offload: local store")
+		}
+		return store, nil
+	}
+	store, err := snapshotoffload.NewS3Store(ctx, snapshotoffload.S3StoreConfig{
+		Bucket:               bucket,
+		Region:               strings.TrimSpace(*snapshotOffloadRegion),
+		Endpoint:             strings.TrimSpace(*snapshotOffloadEndpoint),
+		Profile:              strings.TrimSpace(*snapshotOffloadProfile),
+		ForcePathStyle:       *snapshotOffloadForcePathStyle,
+		ServerSideEncryption: strings.TrimSpace(*snapshotOffloadSSE),
+		SSEKMSKeyID:          strings.TrimSpace(*snapshotOffloadSSEKMSKeyID),
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "snapshot offload: s3 store")
+	}
+	return store, nil
+}
+
+// snapshotOffloadGroups builds one OffloadGroup per local Raft group.
+//
+// Both leadership callbacks read the engine through snapshotEngine():
+// the scheduler outlives startup and races Close(), so a direct field
+// read would be a data race. A runtime whose engine has been cleared
+// reports "not leader", which fails closed.
+func snapshotOffloadGroups(
+	runtimes []*raftGroupRuntime, raftDir, raftID string, multi bool,
+) []snapshotoffload.OffloadGroup {
+	groups := make([]snapshotoffload.OffloadGroup, 0, len(runtimes))
+	for _, rt := range runtimes {
+		if rt == nil {
+			continue
+		}
+		groups = append(groups, snapshotoffload.OffloadGroup{
+			GroupID:      rt.spec.id,
+			DataDir:      groupDataDir(raftDir, raftID, rt.spec.id, multi),
+			IsLeader:     snapshotOffloadIsLeader(rt),
+			VerifyLeader: snapshotOffloadVerifyLeader(rt),
+		})
+	}
+	return groups
+}
+
+func snapshotOffloadIsLeader(rt *raftGroupRuntime) func() bool {
+	return func() bool {
+		engine := rt.snapshotEngine()
+		return engine != nil && engine.State() == raftengine.StateLeader
+	}
+}
+
+// snapshotOffloadVerifyLeader is the §4 pre-commit re-verification: a
+// multi-gigabyte spool takes long enough to lose an election, so
+// leadership must hold at the instant the manifest commits, not merely
+// when the snapshot was opened.
+func snapshotOffloadVerifyLeader(rt *raftGroupRuntime) func(context.Context) error {
+	return func(ctx context.Context) error {
+		engine := rt.snapshotEngine()
+		if engine == nil {
+			return errors.Wrap(snapshotoffload.ErrInvalidOptions,
+				"snapshot offload: raft engine closed")
+		}
+		verifier, ok := engine.(interface {
+			VerifyLeader(context.Context) error
+		})
+		if !ok {
+			return errors.Wrap(snapshotoffload.ErrInvalidOptions,
+				"snapshot offload: raft engine cannot verify leadership")
+		}
+		return errors.Wrap(verifier.VerifyLeader(ctx), "snapshot offload: verify leadership")
+	}
+}
+
+// startSnapshotOffload wires and starts the scheduler when offload is
+// configured. It returns an error rather than logging and continuing:
+// an operator who configured a backup destination and got no backups
+// is worse off than one whose node refused to start.
+func startSnapshotOffload(
+	ctx context.Context,
+	eg *errgroup.Group,
+	runtimes []*raftGroupRuntime,
+	raftDir, raftID string,
+	multi bool,
+	observer snapshotoffload.SchedulerObserver,
+	logger *slog.Logger,
+) error {
+	if !snapshotOffloadEnabled() {
+		return nil
+	}
+	store, err := buildSnapshotOffloadStore(ctx)
+	if err != nil {
+		return err
+	}
+
+	opts := []snapshotoffload.SchedulerOption{
+		snapshotoffload.WithSchedulerInterval(*snapshotOffloadInterval),
+		snapshotoffload.WithSchedulerConcurrency(*snapshotOffloadConcurrency),
+		snapshotoffload.WithSchedulerObserver(observer),
+		snapshotoffload.WithSchedulerLogger(logger),
+	}
+	if *snapshotOffloadJitter > 0 {
+		opts = append(opts, snapshotoffload.WithSchedulerJitter(*snapshotOffloadJitter))
+	}
+	if dir := strings.TrimSpace(*snapshotOffloadSpoolDir); dir != "" {
+		opts = append(opts, snapshotoffload.WithSchedulerSpoolDir(dir))
+	}
+
+	scheduler, err := snapshotoffload.NewScheduler(
+		store,
+		snapshotOffloadGroups(runtimes, raftDir, raftID, multi),
+		strings.TrimSpace(*snapshotOffloadPrefix),
+		strings.TrimSpace(*snapshotOffloadSourceCluster),
+		buildVersion(),
+		opts...,
+	)
+	if err != nil {
+		return errors.Wrap(err, "snapshot offload: scheduler")
+	}
+
+	logger.Info("snapshot offload enabled",
+		slog.Int("groups", len(runtimes)),
+		slog.Duration("interval", *snapshotOffloadInterval),
+		slog.Int("concurrency", *snapshotOffloadConcurrency))
+
+	eg.Go(func() error {
+		// Run returns only on context cancellation; a failing group is
+		// retried on the next tick rather than tearing the process
+		// down, because an object-store outage must not stop serving.
+		if err := scheduler.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
+			return errors.Wrap(err, "snapshot offload scheduler")
+		}
+		return nil
+	})
+	return nil
+}
diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go
new file mode 100644
index 000000000..aae595609
--- /dev/null
+++ b/main_snapshot_offload_test.go
@@ -0,0 +1,125 @@
+package main
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/snapshotoffload"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+// withOffloadFlags sets the offload flags for one test and restores
+// them afterwards. The flags are process globals, so a test that left
+// them set would enable offload for every later test in the package.
+func withOffloadFlags(t *testing.T, bucket, localDir string) {
+	t.Helper()
+	origBucket, origLocal := *snapshotOffloadBucket, *snapshotOffloadLocalDir
+	*snapshotOffloadBucket, *snapshotOffloadLocalDir = bucket, localDir
+	t.Cleanup(func() {
+		*snapshotOffloadBucket, *snapshotOffloadLocalDir = origBucket, origLocal
+	})
+}
+
+// TestSnapshotOffloadIsOptIn pins that a node which configured no
+// destination does no offload work and cannot fail startup on offload
+// configuration.
+func TestSnapshotOffloadIsOptIn(t *testing.T) {
+	withOffloadFlags(t, "", "")
+	require.False(t, snapshotOffloadEnabled())
+	require.NoError(t, startSnapshotOffload(
+		context.Background(), nil, nil, t.TempDir(), "n1", false, nil, testLogger(t)))
+}
+
+// TestSnapshotOffloadRejectsAmbiguousDestination guards against
+// accepting both a bucket and a local dir: which destination actually
+// receives the artifacts would be ambiguous, and a backup written to
+// the wrong place is discovered only when a restore is attempted.
+func TestSnapshotOffloadRejectsAmbiguousDestination(t *testing.T) {
+	withOffloadFlags(t, "some-bucket", t.TempDir())
+	require.True(t, snapshotOffloadEnabled())
+
+	_, err := buildSnapshotOffloadStore(context.Background())
+	require.Error(t, err)
+	require.True(t, errors.Is(err, snapshotoffload.ErrInvalidOptions))
+	require.ErrorContains(t, err, "mutually exclusive")
+}
+
+func TestSnapshotOffloadBuildsALocalStore(t *testing.T) {
+	root := t.TempDir()
+	withOffloadFlags(t, "", root)
+
+	store, err := buildSnapshotOffloadStore(context.Background())
+	require.NoError(t, err)
+	require.NotNil(t, store)
+	_, ok := store.(*snapshotoffload.LocalStore)
+	require.True(t, ok)
+}
+
+// TestSnapshotOffloadGroupsCarryPerGroupDataDirs pins that each group
+// is pointed at its own Raft data dir. Publishing a group's snapshot
+// from another group's directory would ship the wrong state under the
+// right manifest identity.
+func TestSnapshotOffloadGroupsCarryPerGroupDataDirs(t *testing.T) {
+	raftDir := t.TempDir()
+	runtimes := []*raftGroupRuntime{
+		{spec: groupSpec{id: 1}},
+		{spec: groupSpec{id: 2}},
+		nil, // a nil runtime must be skipped, not panic
+	}
+
+	groups := snapshotOffloadGroups(runtimes, raftDir, "n1", true)
+	require.Len(t, groups, 2)
+
+	seen := map[uint64]string{}
+	for _, g := range groups {
+		require.NotNil(t, g.IsLeader, "every group must carry both leadership callbacks")
+		require.NotNil(t, g.VerifyLeader)
+		seen[g.GroupID] = g.DataDir
+	}
+	require.Equal(t, filepath.Join(raftDir, "n1", "group-1"), seen[1])
+	require.Equal(t, filepath.Join(raftDir, "n1", "group-2"), seen[2])
+	require.NotEqual(t, seen[1], seen[2])
+}
+
+// TestSnapshotOffloadLeadershipFailsClosedOnAClosedEngine covers
+// shutdown: the scheduler outlives startup and races Close(), so a
+// runtime whose engine has been cleared must report "not leader"
+// rather than panic or, worse, publish.
+func TestSnapshotOffloadLeadershipFailsClosedOnAClosedEngine(t *testing.T) {
+	rt := &raftGroupRuntime{spec: groupSpec{id: 7}} // engine never set
+
+	require.False(t, snapshotOffloadIsLeader(rt)(),
+		"a closed engine must never look like a leader")
+
+	err := snapshotOffloadVerifyLeader(rt)(context.Background())
+	require.Error(t, err)
+	require.True(t, errors.Is(err, snapshotoffload.ErrInvalidOptions))
+}
+
+// TestStartSnapshotOffloadRejectsIncompleteConfiguration pins that a
+// configured-but-invalid offload fails startup rather than logging and
+// leaving the operator with no backups.
+func TestStartSnapshotOffloadRejectsIncompleteConfiguration(t *testing.T) {
+	withOffloadFlags(t, "", t.TempDir())
+	origCluster := *snapshotOffloadSourceCluster
+	*snapshotOffloadSourceCluster = "   " // whitespace-only: no identity
+	t.Cleanup(func() { *snapshotOffloadSourceCluster = origCluster })
+
+	err := startSnapshotOffload(
+		context.Background(), nil,
+		[]*raftGroupRuntime{{spec: groupSpec{id: 1}}},
+		t.TempDir(), "n1", false, nil, testLogger(t))
+	require.Error(t, err)
+	require.True(t, errors.Is(err, snapshotoffload.ErrInvalidOptions))
+}
+
+// testLogger discards output so a test that exercises the enabled path
+// does not spam the run.
+func testLogger(t *testing.T) *slog.Logger {
+	t.Helper()
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
diff --git a/monitoring/registry.go b/monitoring/registry.go
index 568bfeeb6..c25257c52 100644
--- a/monitoring/registry.go
+++ b/monitoring/registry.go
@@ -31,6 +31,7 @@ type Registry struct {
 	coldStartObs  *ColdStartObserver
 	tso           *TSOMetrics
 	tsoObserver   *TSOObserver
+	snapOffload   *SnapshotOffloadMetrics
 }
 
 // NewRegistry builds a registry with constant labels that identify the local node.
@@ -63,6 +64,7 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry {
 	r.coldStartObs = newColdStartObserver(r.coldStart)
 	r.tso = newTSOMetrics(registerer)
 	r.tsoObserver = newTSOObserver(r.tso)
+	r.snapOffload = newSnapshotOffloadMetrics(registerer)
 	return r
 }
 
@@ -292,3 +294,13 @@ func (r *Registry) TSOObserver() *TSOObserver {
 	}
 	return r.tsoObserver
 }
+
+// SnapshotOffloadObserver returns the physical snapshot offload
+// scheduler's metrics observer. Passed to the scheduler through
+// snapshotoffload.WithSchedulerObserver.
+func (r *Registry) SnapshotOffloadObserver() *SnapshotOffloadMetrics {
+	if r == nil {
+		return nil
+	}
+	return r.snapOffload
+}
diff --git a/monitoring/snapshot_offload.go b/monitoring/snapshot_offload.go
new file mode 100644
index 000000000..6670a2bc4
--- /dev/null
+++ b/monitoring/snapshot_offload.go
@@ -0,0 +1,157 @@
+package monitoring
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// snapshotPayloadBucketBase is the smallest payload-size histogram
+// bucket (1 MiB). Snapshots below it are rounding error next to the
+// multi-gigabyte cases the histogram exists to show.
+const (
+	snapshotPayloadBucketBase   = 1 << 20 // 1 MiB
+	snapshotPayloadBucketFactor = 4
+	snapshotPayloadBucketCount  = 8 // 1 MiB through ~16 GiB
+)
+
+// SnapshotOffloadMetrics exposes the physical snapshot offload
+// scheduler's outcomes (design doc §4).
+//
+// group_id is a label on every series: its cardinality is the number
+// of Raft groups this process hosts, which is bounded by deployment
+// topology rather than by traffic. skip reason is a closed set owned
+// by the scheduler.
+type SnapshotOffloadMetrics struct {
+	published        *prometheus.CounterVec
+	skipped          *prometheus.CounterVec
+	failed           *prometheus.CounterVec
+	lastPublishIndex *prometheus.GaugeVec
+	publishSeconds   *prometheus.HistogramVec
+	payloadBytes     *prometheus.HistogramVec
+}
+
+func newSnapshotOffloadMetrics(registerer prometheus.Registerer) *SnapshotOffloadMetrics {
+	m := &SnapshotOffloadMetrics{
+		published: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "elastickv_snapshot_offload_published_total",
+				Help: "Total physical snapshots published to the object store, by Raft group.",
+			},
+			[]string{"group_id"},
+		),
+		skipped: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "elastickv_snapshot_offload_skipped_total",
+				Help: "Total offload scans that published nothing, by Raft group and reason. Routine on a follower or an unchanged snapshot.",
+			},
+			[]string{"group_id", "reason"},
+		),
+		failed: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "elastickv_snapshot_offload_failed_total",
+				Help: "Total offload attempts that failed, by Raft group. A sustained rate means backups are not being taken.",
+			},
+			[]string{"group_id"},
+		),
+		lastPublishIndex: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "elastickv_snapshot_offload_last_published_index",
+				Help: "Raft index of the most recent snapshot this process published, by group. Staleness here is the backup-freshness signal.",
+			},
+			[]string{"group_id"},
+		),
+		publishSeconds: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name:    "elastickv_snapshot_offload_publish_seconds",
+				Help:    "Wall time to spool, upload and commit one snapshot.",
+				Buckets: []float64{0.5, 1, 5, 15, 30, 60, 300, 900, 1800, 3600},
+			},
+			[]string{"group_id"},
+		),
+		payloadBytes: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name: "elastickv_snapshot_offload_payload_bytes",
+				Help: "Size of each published snapshot payload.",
+				Buckets: prometheus.ExponentialBuckets(
+					snapshotPayloadBucketBase,
+					snapshotPayloadBucketFactor,
+					snapshotPayloadBucketCount,
+				),
+			},
+			[]string{"group_id"},
+		),
+	}
+	registerer.MustRegister(
+		m.published,
+		m.skipped,
+		m.failed,
+		m.lastPublishIndex,
+		m.publishSeconds,
+		m.payloadBytes,
+	)
+	return m
+}
+
+// ObserveSnapshotOffloadPublished records one successful publication.
+func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadPublished(
+	groupID, index uint64, payloadBytes int64, elapsed time.Duration,
+) {
+	if m == nil {
+		return
+	}
+	label := snapshotOffloadGroupLabel(groupID)
+	m.published.WithLabelValues(label).Inc()
+	m.lastPublishIndex.WithLabelValues(label).Set(float64(index))
+	m.publishSeconds.WithLabelValues(label).Observe(max(0, elapsed).Seconds())
+	m.payloadBytes.WithLabelValues(label).Observe(float64(max(int64(0), payloadBytes)))
+}
+
+// ObserveSnapshotOffloadSkipped records a scan that published nothing.
+func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadSkipped(groupID uint64, reason string) {
+	if m == nil {
+		return
+	}
+	m.skipped.WithLabelValues(snapshotOffloadGroupLabel(groupID), normalizeSnapshotOffloadSkip(reason)).Inc()
+}
+
+// ObserveSnapshotOffloadFailed records a failed attempt. The error is
+// deliberately not a label: its text is unbounded, and a per-message
+// series would let one recurring failure explode the metric's
+// cardinality. Diagnosis comes from the scheduler's log line.
+func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadFailed(groupID uint64, _ error) {
+	if m == nil {
+		return
+	}
+	m.failed.WithLabelValues(snapshotOffloadGroupLabel(groupID)).Inc()
+}
+
+func snapshotOffloadGroupLabel(groupID uint64) string {
+	return strconv.FormatUint(groupID, 10)
+}
+
+// Skip reasons emitted by the scheduler.
+const (
+	snapshotOffloadSkipNotLeader        = "not_leader"
+	snapshotOffloadSkipAlreadyPublished = "already_published"
+	snapshotOffloadSkipNoSnapshot       = "no_persisted_snapshot"
+	snapshotOffloadSkipInFlight         = "already_in_flight"
+	snapshotOffloadSkipUnknownLeader    = "leadership_unknown"
+	snapshotOffloadSkipUnknown          = "unknown"
+)
+
+// normalizeSnapshotOffloadSkip keeps the reason label inside the
+// scheduler's closed set.
+func normalizeSnapshotOffloadSkip(reason string) string {
+	switch reason {
+	case snapshotOffloadSkipNotLeader,
+		snapshotOffloadSkipAlreadyPublished,
+		snapshotOffloadSkipNoSnapshot,
+		snapshotOffloadSkipInFlight,
+		snapshotOffloadSkipUnknownLeader:
+		return reason
+	default:
+		return snapshotOffloadSkipUnknown
+	}
+}
diff --git a/monitoring/snapshot_offload_test.go b/monitoring/snapshot_offload_test.go
new file mode 100644
index 000000000..b2e68a822
--- /dev/null
+++ b/monitoring/snapshot_offload_test.go
@@ -0,0 +1,96 @@
+package monitoring
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSnapshotOffloadMetricsRecordOutcomes(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSnapshotOffloadMetrics(reg)
+
+	m.ObserveSnapshotOffloadPublished(7, 4211, 5<<20, 12*time.Second)
+	m.ObserveSnapshotOffloadSkipped(7, snapshotOffloadSkipNotLeader)
+	m.ObserveSnapshotOffloadSkipped(7, snapshotOffloadSkipAlreadyPublished)
+	m.ObserveSnapshotOffloadFailed(9, errors.New("object store unavailable"))
+
+	require.NoError(t, testutil.GatherAndCompare(
+		reg,
+		strings.NewReader(`
+# HELP elastickv_snapshot_offload_published_total Total physical snapshots published to the object store, by Raft group.
+# TYPE elastickv_snapshot_offload_published_total counter
+elastickv_snapshot_offload_published_total{group_id="7"} 1
+# HELP elastickv_snapshot_offload_last_published_index Raft index of the most recent snapshot this process published, by group. Staleness here is the backup-freshness signal.
+# TYPE elastickv_snapshot_offload_last_published_index gauge
+elastickv_snapshot_offload_last_published_index{group_id="7"} 4211
+# HELP elastickv_snapshot_offload_failed_total Total offload attempts that failed, by Raft group. A sustained rate means backups are not being taken.
+# TYPE elastickv_snapshot_offload_failed_total counter
+elastickv_snapshot_offload_failed_total{group_id="9"} 1
+# HELP elastickv_snapshot_offload_skipped_total Total offload scans that published nothing, by Raft group and reason. Routine on a follower or an unchanged snapshot.
+# TYPE elastickv_snapshot_offload_skipped_total counter
+elastickv_snapshot_offload_skipped_total{group_id="7",reason="already_published"} 1
+elastickv_snapshot_offload_skipped_total{group_id="7",reason="not_leader"} 1
+`),
+		"elastickv_snapshot_offload_published_total",
+		"elastickv_snapshot_offload_last_published_index",
+		"elastickv_snapshot_offload_failed_total",
+		"elastickv_snapshot_offload_skipped_total",
+	))
+}
+
+// TestSnapshotOffloadMetricsBoundTheSkipReasonLabel is the cardinality
+// guard: an unrecognised reason must collapse rather than mint a
+// series per distinct string.
+func TestSnapshotOffloadMetricsBoundTheSkipReasonLabel(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSnapshotOffloadMetrics(reg)
+
+	m.ObserveSnapshotOffloadSkipped(1, "something-new")
+	m.ObserveSnapshotOffloadSkipped(1, "something-else")
+	m.ObserveSnapshotOffloadSkipped(1, "")
+
+	require.Equal(t, 1, testutil.CollectAndCount(m.skipped))
+	require.InDelta(t, 3.0,
+		testutil.ToFloat64(m.skipped.WithLabelValues("1", snapshotOffloadSkipUnknown)), 0.0001)
+}
+
+// TestSnapshotOffloadMetricsDoNotLabelByError pins that the failure
+// counter carries no error text: messages are unbounded, and one
+// recurring failure would otherwise explode the metric's cardinality.
+func TestSnapshotOffloadMetricsDoNotLabelByError(t *testing.T) {
+	t.Parallel()
+
+	reg := prometheus.NewRegistry()
+	m := newSnapshotOffloadMetrics(reg)
+
+	for i := range 20 {
+		m.ObserveSnapshotOffloadFailed(1, errors.New(strings.Repeat("x", i+1)))
+	}
+	require.Equal(t, 1, testutil.CollectAndCount(m.failed),
+		"distinct error texts must not create distinct series")
+}
+
+func TestSnapshotOffloadMetricsNilReceiverIsInert(t *testing.T) {
+	t.Parallel()
+
+	var m *SnapshotOffloadMetrics
+	require.NotPanics(t, func() {
+		m.ObserveSnapshotOffloadPublished(1, 2, 3, time.Second)
+		m.ObserveSnapshotOffloadSkipped(1, "x")
+		m.ObserveSnapshotOffloadFailed(1, errors.New("boom"))
+	})
+	require.NotNil(t, NewRegistry("n1", "127.0.0.1:1").SnapshotOffloadObserver())
+
+	var nilRegistry *Registry
+	require.Nil(t, nilRegistry.SnapshotOffloadObserver())
+}

From 79ec2ed8bef36fe00f743f556e67f1846776b22a Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Tue, 8 Sep 2026 16:30:10 +0900
Subject: [PATCH 2/4] docs: add the snapshot offload operations runbook

Closes the M3 operator-documentation item: enabling offload, verifying
that backups are actually being produced, retention semantics, restore,
and failure modes.

Two things the runbook makes explicit because they are the ways an
operator gets silently burned:

  - a group whose last_published_index never advances has no backups
    even though nothing is failing, so staleness needs its own alert;
  - a versioned bucket without a noncurrent-version lifecycle rule
    grows without bound while retention reports success, because a
    keyed delete only writes a delete marker.

Every flag, metric name and skip reason in the runbook was
cross-checked against the source rather than written from memory.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 ...artial_physical_snapshot_object_offload.md |   5 +-
 docs/snapshot_offload_operations.md           | 203 ++++++++++++++++++
 2 files changed, 206 insertions(+), 2 deletions(-)
 create mode 100644 docs/snapshot_offload_operations.md

diff --git a/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md b/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md
index f338b0cca..7581ea7ce 100644
--- a/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md
+++ b/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md
@@ -51,8 +51,9 @@ The M1 object-store-neutral substrate now adds:
 The runtime scheduler is implemented and wired into main.go, opt-in via
 `--snapshotOffloadBucket` (or `--snapshotOffloadLocalDir`). Retention/GC
 is implemented per §5. Restore drills and corruption tests are in place;
-multi-node acceptance, operator documentation, and the §7
-versioned-bucket decision remain pending.
+multi-node acceptance and the §7 versioned-bucket decision remain
+pending; the operator runbook is at
+[`../snapshot_offload_operations.md`](../snapshot_offload_operations.md).
 
 ## 2. Safety boundary
 
diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md
new file mode 100644
index 000000000..592dbf86c
--- /dev/null
+++ b/docs/snapshot_offload_operations.md
@@ -0,0 +1,203 @@
+# Physical Snapshot Object Offload — Operations
+
+Runbook for the physical snapshot offload subsystem: continuous backup of
+Raft snapshots to an S3-compatible object store, and disaster recovery from
+those artifacts.
+
+Design: [`design/2026_07_19_partial_physical_snapshot_object_offload.md`](design/2026_07_19_partial_physical_snapshot_object_offload.md).
+
+> **Note:** §4 (Retention) describes the retention/GC subsystem, which lands
+> in a separate change. Everything else here is live once this change ships.
+
+## Scope
+
+Use this runbook to:
+
+1. enable continuous snapshot offload on a cluster,
+2. verify that backups are actually being produced,
+3. restore a node from a published snapshot,
+4. configure retention, and understand what it will and will not delete.
+
+This is **physical** backup: it ships the Raft snapshot the engine already
+produced. It does not force an extra state-machine snapshot, so backup
+freshness is bounded by the engine's own snapshot cadence.
+
+## 1. What gets written
+
+Two object kinds under the configured prefix:
+
+```
+<prefix>/v1/groups/<group>/snapshots/<index>-<term>.json   manifest
+<prefix>/v1/payloads/sha256/<xx>/<sha256>.fsm              payload
+```
+
+Payloads are **content-addressed and shared**: two groups (or two generations)
+whose snapshots hash identically converge on one object. This matters for
+retention — see §4.
+
+Manifests are immutable and self-hashing. A manifest names exactly one payload.
+
+## 2. Enabling offload
+
+Offload is opt-in. A node with no destination configured does no offload work
+and cannot fail startup on offload settings.
+
+```bash
+elastickv \
+  --snapshotOffloadBucket=my-backup-bucket \
+  --snapshotOffloadRegion=ap-northeast-1 \
+  --snapshotOffloadSourceCluster=prod-tokyo \
+  --snapshotOffloadPrefix=elastickv \
+  --snapshotOffloadServerSideEncryption=aws:kms \
+  --snapshotOffloadSSEKMSKeyId=arn:aws:kms:ap-northeast-1:123456789012:key/abcd
+```
+
+| Flag | Meaning |
+|---|---|
+| `--snapshotOffloadBucket` | S3 bucket. Enables offload. |
+| `--snapshotOffloadLocalDir` | Filesystem root instead of S3. **Mutually exclusive** with the bucket. |
+| `--snapshotOffloadSourceCluster` | Cluster identity recorded in every manifest. Required. |
+| `--snapshotOffloadPrefix` | Key prefix for all artifacts. |
+| `--snapshotOffloadRegion` / `--snapshotOffloadEndpoint` / `--snapshotOffloadProfile` / `--snapshotOffloadForcePathStyle` | S3 addressing and credentials. |
+| `--snapshotOffloadServerSideEncryption` / `--snapshotOffloadSSEKMSKeyId` | `AES256` or `aws:kms`. KMS aliases are rejected; pass an ARN or bare key ID. |
+| `--snapshotOffloadInterval` | Scan cadence. Default 15m. |
+| `--snapshotOffloadJitter` | Spread across groups. Default: a quarter of the interval. |
+| `--snapshotOffloadConcurrency` | Concurrent uploads per process. Default 1. |
+| `--snapshotOffloadSpoolDir` | Where payloads are spooled before upload. Needs room for the largest snapshot. |
+
+**A misconfigured offload refuses to start the node.** That is deliberate: an
+operator who configured a backup destination and silently received no backups
+is worse off than one whose node failed loudly.
+
+### Security requirements
+
+The bucket holds physical keys and metadata. Storage-envelope encryption
+protects *values*, not all keys and metadata, so the bucket itself must be
+protected:
+
+- private ACLs — anonymous read or write is a deployment failure,
+- TLS,
+- server-side encryption (SSE-S3 or SSE-KMS),
+- credentials scoped to `list`/`get`/`put`/`delete` **below the prefix only**,
+- secrets supplied by file or environment, never in process arguments.
+
+## 3. Verifying that backups are happening
+
+Only the current leader of a group publishes; followers skip. On a healthy
+three-node group, exactly one node reports publishes and two report
+`not_leader`.
+
+```promql
+# Backup freshness — the number that matters. Alert if it stops advancing.
+elastickv_snapshot_offload_last_published_index
+
+# Backups are failing. Any sustained rate is paging-grade.
+rate(elastickv_snapshot_offload_failed_total[15m])
+
+# Routine skips. Expected on followers and unchanged snapshots.
+rate(elastickv_snapshot_offload_skipped_total[15m])
+```
+
+Skip reasons and what they mean:
+
+| Reason | Meaning | Action |
+|---|---|---|
+| `not_leader` | This node does not lead the group. | None — expected on followers. |
+| `already_published` | Snapshot unchanged since this process last published it. | None. |
+| `no_persisted_snapshot` | The group has not produced a snapshot yet. | None on a young cluster. Investigate if it persists on a busy group. |
+| `already_in_flight` | Another scan is publishing this group. | None. |
+| `leadership_unknown` | Engine unavailable, typically during shutdown. | None if the node is stopping. |
+
+**A group whose `last_published_index` never advances has no backups**, even
+though nothing is failing. Alert on staleness, not only on errors.
+
+## 4. Retention
+
+Retention is per group, and runs in two phases.
+
+**Phase 1 — manifests.** Keeps `MinGenerations` newest per group plus anything
+inside `MaxAge`, and always keeps a group's newest valid manifest regardless of
+both. A group can never be left with no restore point.
+
+**Phase 2 — payloads.** Rebuilds the live set from **every surviving manifest
+in the whole prefix** — not per group, because payloads are shared — then
+reclaims only unreferenced objects, using **two-pass mark-and-sweep**: a pass
+marks an eligible payload, and only a later pass, with the object unchanged and
+the mark older than `MinMarkAge`, deletes it.
+
+The second pass exists because a publisher reusing a payload rewrites identical
+bytes, which no general-purpose S3 precondition can detect (`If-Match` compares
+a content-derived ETag; `IfMatchLastModifiedTime` is directory-buckets only).
+**`MinMarkAge` must exceed your longest plausible publish.**
+
+Retention refuses to delete anything when it cannot prove the live set:
+
+- a malformed manifest anywhere in the prefix → payload reclamation is skipped
+  entirely, and the malformed object is preserved for inspection,
+- a listing or pagination failure → no deletes at all,
+- an object under the payload prefix that does not parse as a payload key →
+  left alone.
+
+If `PayloadPhaseSkipped` is set with malformed manifests reported, fix or
+remove the malformed object; storage will not be reclaimed until you do.
+
+### Versioned buckets
+
+Retention deletes by key. On a bucket with **S3 versioning enabled**, a keyed
+delete only writes a delete marker: the bytes survive as a noncurrent version
+that later listings cannot see, so GC reports successful reclamation while
+storage grows without bound.
+
+**A versioned backup bucket requires a noncurrent-version expiration lifecycle
+rule.** Whether to instead enumerate versions directly, or refuse versioned
+buckets at startup, is an open decision.
+
+## 5. Restore
+
+Restore is **offline** and targets an **absent** data directory. It refuses to
+overwrite an existing one — that guard is what protects an operator who
+mistakenly points a restore at a live node.
+
+```bash
+# 1. Find the generation to restore.
+elastickv-snapshot-offload publish --help   # same store flags as below
+
+# 2. Restore into a fresh directory.
+elastickv-snapshot-offload restore \
+  --store=s3 --s3-bucket=my-backup-bucket --s3-region=ap-northeast-1 \
+  --manifest-key='elastickv/v1/groups/1/snapshots/00000000000000004211-00000000000000000007.json' \
+  --data-dir=/var/lib/elastickv/n1 \
+  --peers='n1=10.0.0.1:50051,n2=10.0.0.2:50051,n3=10.0.0.3:50051'
+
+# 3. Start the node normally against the restored directory.
+```
+
+Restore verifies exact length and SHA-256 before the payload is accepted, then
+fsyncs and atomically renames it into place. Any integrity failure leaves the
+destination **absent** rather than half-written.
+
+Target membership (`--peers`) is explicit operator input, not copied from the
+source. That is what makes recovery onto replacement addresses possible, while
+the source membership stays in the manifest for audit.
+
+Exit codes: `0` success, `1` invalid invocation, `2` missing or invalid
+snapshot data. Automation should distinguish these.
+
+## 6. Failure modes
+
+| Symptom | Cause | Action |
+|---|---|---|
+| `last_published_index` frozen, no failures | Node is not the leader, or the engine has produced no new snapshot. | Confirm which node leads the group; check the engine's snapshot cadence. |
+| Sustained `failed_total` | Object store unreachable, credentials expired, bucket policy denies writes. | Check the scheduler's log line — it carries the error the metric deliberately omits. |
+| Storage grows despite retention | Versioned bucket without a lifecycle rule (§4), or reclamation blocked by a malformed manifest. | Add the lifecycle rule; inspect reported malformed manifests. |
+| Restore fails with an integrity error | Payload truncated, over-length, or the manifest was edited. | Restore an older generation; the destination was left absent, so nothing was damaged. |
+| Restore refuses to run | Destination directory already exists. | Restore into a fresh path. Never delete a live data dir to make room. |
+
+## 7. Limits
+
+- Backup freshness is bounded by the Raft engine's snapshot cadence; offload
+  never forces an extra snapshot.
+- Losing a group's leadership mid-publish can leave an unreferenced payload,
+  which retention reclaims. It can never leave a committed manifest.
+- Mark state is per-process and in memory. A restart delays reclamation by one
+  pass; it never advances it.

From e349dee1bd2203c5a2056fe9ab51d4b9223992d2 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Wed, 9 Sep 2026 13:16:32 +0900
Subject: [PATCH 3/4] docs: correct the multi-group restore path in the offload
 runbook
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The restore example pointed --data-dir at the node's raft dir, but the
server derives a per-group directory: a multi-group node opens
<raftDir>/<raftID>/group-<G>. An operator following the runbook during
disaster recovery would restore into a path the server never opens,
find the per-group directories empty at startup, and have the restore
silently ignored — the worst possible moment for a documentation bug.

Adds the full derivation table (multi-group, single group, single-node
group 0) and states that a multi-group recovery needs one restore
invocation per group.

A table-driven test pins every documented path against groupDataDir so
the runbook cannot drift from the function the server actually uses.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 docs/snapshot_offload_operations.md | 24 +++++++++++++++++---
 main_snapshot_offload_test.go       | 34 +++++++++++++++++++++++++++++
 2 files changed, 55 insertions(+), 3 deletions(-)

diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md
index 592dbf86c..e6af6d476 100644
--- a/docs/snapshot_offload_operations.md
+++ b/docs/snapshot_offload_operations.md
@@ -162,16 +162,34 @@ mistakenly points a restore at a live node.
 # 1. Find the generation to restore.
 elastickv-snapshot-offload publish --help   # same store flags as below
 
-# 2. Restore into a fresh directory.
+# 2. Restore each group into ITS OWN directory (see the path rule below).
 elastickv-snapshot-offload restore \
   --store=s3 --s3-bucket=my-backup-bucket --s3-region=ap-northeast-1 \
   --manifest-key='elastickv/v1/groups/1/snapshots/00000000000000004211-00000000000000000007.json' \
-  --data-dir=/var/lib/elastickv/n1 \
+  --data-dir=/var/lib/elastickv/n1/group-1 \
   --peers='n1=10.0.0.1:50051,n2=10.0.0.2:50051,n3=10.0.0.3:50051'
 
-# 3. Start the node normally against the restored directory.
+# 3. Repeat for every group the node hosts, then start it normally.
 ```
 
+### The `--data-dir` path must match what the server will open
+
+`--data-dir` is the **per-group** directory, not the node's `--raftDir`.
+The server derives it as:
+
+| Deployment | Group | Directory |
+|---|---|---|
+| multi-group (`--raftRedisMap` etc.) | any group *G* | `<raftDir>/<raftID>/group-<G>` |
+| single group | the default group | `<raftDir>/<raftID>` |
+| single-node, group 0 | 0 | `<raftDir>/<raftID>/group-0` |
+
+Restoring a multi-group node into `<raftDir>/<raftID>` puts the data
+where the server never looks: startup finds the per-group directories
+empty and the restore is silently ignored. **A multi-group recovery
+must restore every group's manifest into its own `group-<G>`
+directory** — one `restore` invocation per group — or the node comes
+back with only the groups you happened to place correctly.
+
 Restore verifies exact length and SHA-256 before the payload is accepted, then
 fsyncs and atomically renames it into place. Any integrity failure leaves the
 destination **absent** rather than half-written.
diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go
index aae595609..f879f172a 100644
--- a/main_snapshot_offload_test.go
+++ b/main_snapshot_offload_test.go
@@ -123,3 +123,37 @@ func testLogger(t *testing.T) *slog.Logger {
 	t.Helper()
 	return slog.New(slog.NewTextHandler(io.Discard, nil))
 }
+
+// TestRunbookRestorePathsMatchGroupDataDir keeps the operations
+// runbook's `--data-dir` table honest against the function the server
+// actually uses.
+//
+// A wrong path here is not a cosmetic doc bug: an operator following
+// it during disaster recovery restores into a directory the server
+// never opens, startup finds the per-group directories empty, and the
+// restore is silently ignored.
+func TestRunbookRestorePathsMatchGroupDataDir(t *testing.T) {
+	t.Parallel()
+
+	const raftDir = "/var/lib/elastickv"
+	const raftID = "n1"
+
+	tests := []struct {
+		name    string
+		groupID uint64
+		multi   bool
+		want    string
+	}{
+		{name: "multi-group", groupID: 1, multi: true, want: "/var/lib/elastickv/n1/group-1"},
+		{name: "multi-group higher id", groupID: 7, multi: true, want: "/var/lib/elastickv/n1/group-7"},
+		{name: "single group", groupID: 1, multi: false, want: "/var/lib/elastickv/n1"},
+		{name: "single node group zero", groupID: 0, multi: false, want: "/var/lib/elastickv/n1/group-0"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			require.Equal(t, tc.want, groupDataDir(raftDir, raftID, tc.groupID, tc.multi),
+				"docs/snapshot_offload_operations.md documents this path for restore")
+		})
+	}
+}

From b1738d4c3bc3f8b85ac3613c83c41c3fbfe47f12 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Sat, 12 Sep 2026 14:49:50 +0900
Subject: [PATCH 4/4] snapshotoffload: require an expected group id on restore
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three review findings on the restore runbook, two P1.

P1 — restore accepted another group's manifest. RestorePhysicalSnapshot
read Manifest.GroupID but never compared it with anything the operator
supplied, and nothing downstream carries the group's identity: the
prepared artifacts record index, term, peers and payload hash, while
startup derives the group from the directory layout. An operator
repeating the command across groups and pasting the wrong manifest key
therefore produced a valid-looking directory that startup loaded under a
different group's routing identity, reporting no error at any point.

RestoreOptions.ExpectGroupID is now required and checked before the
payload download and before the destination is created, so a mistaken key
costs nothing and leaves nothing behind. It is a *uint64 because group 0
is a real group (the dedicated TSO group) and cannot double as "unset";
the CLI takes --expect-group, also required. The mismatch gets its own
sentinel, ErrRestoreGroupMismatch, so the operator sees "wrong group"
rather than a generic invalid-options error, and classifies as
exitUserErr: the snapshot data is intact, the invocation named the wrong
manifest.

P1 — the runbook's path table was wrong for a mixed TSO topology.
dataGroupsNeedMultiDirs counts DATA groups and excludes group 0, so a
node running the dedicated TSO group alongside a single data group has
two --raftGroups entries but is not multi-dir: group 0 lands in group-0
while the data group opens <raftDir>/<raftID> directly. The table's
"multi-group / any group" row sent that data group to group-1, a
directory startup never opens -- and an empty group is not an error, so
the restore was silently ignored. The table now enumerates the mixed
topology in both directions and states the actual rule.

The existing TestRunbookRestorePathsMatchGroupDataDir could not catch
this because it takes `multi` as an input, so it cannot see a reader
deriving the wrong `multi` in the first place. The new test starts from
[]groupSpec and composes effectiveMultiDataDirs with groupDataDir, which
is the path an operator actually follows.

P2 — --peers must be that group's peer map. Each group has its own
listener addresses from --raftGroups / --raftGroupPeers and restore
persists the supplied peers into that group's data directory, so copying
the example's group-1 endpoints into every invocation leaves the other
groups unable to form a quorum. Documented alongside the other
per-group flags.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 cmd/elastickv-snapshot-offload/main.go      |  27 ++++-
 cmd/elastickv-snapshot-offload/main_test.go | 120 ++++++++++++++++++++
 docs/snapshot_offload_operations.md         |  63 ++++++++--
 internal/snapshotoffload/manifest.go        |  12 +-
 internal/snapshotoffload/offload_test.go    |  45 +++++---
 internal/snapshotoffload/restore.go         |  38 +++++++
 internal/snapshotoffload/s3_store_test.go   |   1 +
 main_snapshot_offload_test.go               |  81 +++++++++++++
 8 files changed, 353 insertions(+), 34 deletions(-)

diff --git a/cmd/elastickv-snapshot-offload/main.go b/cmd/elastickv-snapshot-offload/main.go
index efdd1640d..0bf1bba33 100644
--- a/cmd/elastickv-snapshot-offload/main.go
+++ b/cmd/elastickv-snapshot-offload/main.go
@@ -10,6 +10,7 @@ import (
 	"log/slog"
 	"os"
 	"sort"
+	"strconv"
 	"strings"
 
 	"github.com/bootjp/elastickv/internal/raftengine/etcd"
@@ -58,6 +59,11 @@ 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
 }
 
 func main() {
@@ -151,6 +157,10 @@ 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.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.")
 	if err := fs.Parse(argv); err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -166,6 +176,14 @@ 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")
+	}
+	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
 	}
@@ -259,10 +277,11 @@ func runRestore(ctx context.Context, cfg *restoreConfig, logger *slog.Logger) er
 		return err
 	}
 	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,
 	})
 	if err != nil {
 		return errors.Wrap(err, "restore physical snapshot")
diff --git a/cmd/elastickv-snapshot-offload/main_test.go b/cmd/elastickv-snapshot-offload/main_test.go
index a9dedf22b..e2fc4839c 100644
--- a/cmd/elastickv-snapshot-offload/main_test.go
+++ b/cmd/elastickv-snapshot-offload/main_test.go
@@ -50,6 +50,9 @@ 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",
 	}, io.Discard, logger)
 	require.NoError(t, err)
 	require.Equal(t, exitSuccess, code)
@@ -168,6 +171,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 +188,112 @@ 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",
+	}, 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",
+	})
+	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",
+	})
+	require.Error(t, err)
+}
diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md
index e6af6d476..f28daba98 100644
--- a/docs/snapshot_offload_operations.md
+++ b/docs/snapshot_offload_operations.md
@@ -167,28 +167,69 @@ elastickv-snapshot-offload restore \
   --store=s3 --s3-bucket=my-backup-bucket --s3-region=ap-northeast-1 \
   --manifest-key='elastickv/v1/groups/1/snapshots/00000000000000004211-00000000000000000007.json' \
   --data-dir=/var/lib/elastickv/n1/group-1 \
+  --expect-group=1 \
   --peers='n1=10.0.0.1:50051,n2=10.0.0.2:50051,n3=10.0.0.3:50051'
 
 # 3. Repeat for every group the node hosts, then start it normally.
 ```
 
+### Every flag that says "group" must say the SAME group
+
+`--manifest-key`, `--data-dir`, `--expect-group` and `--peers` are four
+independent statements about which group you are restoring. Three of them
+used to be uncheckable against each other:
+
+- **`--expect-group` is required and is the only cross-check.** Nothing in
+  the restored directory records which group the data came from — the
+  artifacts carry index, term, peers and payload hash, and startup derives
+  the group from the directory layout. So pasting group 2's manifest key
+  into a group-1 restore produced a valid-looking directory that startup
+  then loaded as group 1: the wrong physical FSM under another group's
+  routing identity, with no error anywhere. The restore is now refused,
+  before the download and before the destination is created, so a mistaken
+  key costs nothing and leaves nothing behind.
+- **`--peers` must be that group's peer map, not group 1's.** Each group
+  has its own listener addresses from `--raftGroups` / `--raftGroupPeers`,
+  and restore persists the supplied peers into that group's data
+  directory. Copying the `:50051` endpoints from the example above into
+  every invocation leaves the other groups trying to reach the wrong Raft
+  endpoints, and they never form a quorum. Re-derive the peer list per
+  group the same way you re-derive the manifest key and the data dir.
+
 ### The `--data-dir` path must match what the server will open
 
 `--data-dir` is the **per-group** directory, not the node's `--raftDir`.
 The server derives it as:
 
-| Deployment | Group | Directory |
+The rule the server actually applies (`groupDataDir`) is:
+
+- **group 0 always** gets `<raftDir>/<raftID>/group-0`.
+- **every other group** gets `group-<G>` **only when the node hosts more
+  than one _data_ group**, and `<raftDir>/<raftID>` otherwise.
+
+"More than one data group" is the exact condition, because
+`dataGroupsNeedMultiDirs` counts data groups and **excludes group 0**:
+
+| Deployment (`--raftGroups`) | Group | Directory |
 |---|---|---|
-| multi-group (`--raftRedisMap` etc.) | any group *G* | `<raftDir>/<raftID>/group-<G>` |
-| single group | the default group | `<raftDir>/<raftID>` |
-| single-node, group 0 | 0 | `<raftDir>/<raftID>/group-0` |
-
-Restoring a multi-group node into `<raftDir>/<raftID>` puts the data
-where the server never looks: startup finds the per-group directories
-empty and the restore is silently ignored. **A multi-group recovery
-must restore every group's manifest into its own `group-<G>`
-directory** — one `restore` invocation per group — or the node comes
-back with only the groups you happened to place correctly.
+| two or more data groups, e.g. `1,2` | any group *G* | `<raftDir>/<raftID>/group-<G>` |
+| a single data group, e.g. `1` | 1 | `<raftDir>/<raftID>` |
+| dedicated TSO **plus one** data group, e.g. `0,1` | 0 | `<raftDir>/<raftID>/group-0` |
+| dedicated TSO **plus one** data group, e.g. `0,1` | 1 | `<raftDir>/<raftID>` — **not** `group-1` |
+| dedicated TSO plus two or more data groups, e.g. `0,1,2` | any group *G* | `<raftDir>/<raftID>/group-<G>` |
+
+The fourth row is the one that catches people out. A node running the
+dedicated TSO group alongside a single data group *looks* multi-group —
+two entries in `--raftGroups` — but only group 0 is parked under
+`group-0`; the data group still opens `<raftDir>/<raftID>` directly.
+Restoring it into `group-1` puts the data where startup never looks.
+
+That is the general failure in both directions: a directory the server
+does not open is not an error, it is an empty group. Startup finds
+nothing, the restore is silently ignored, and the node comes back with
+only the groups you happened to place correctly. **Check the table per
+group before each invocation** rather than assuming one rule for the
+whole node.
 
 Restore verifies exact length and SHA-256 before the payload is accepted, then
 fsyncs and atomically renames it into place. Any integrity failure leaves the
diff --git a/internal/snapshotoffload/manifest.go b/internal/snapshotoffload/manifest.go
index ca30a7686..a3e6a09df 100644
--- a/internal/snapshotoffload/manifest.go
+++ b/internal/snapshotoffload/manifest.go
@@ -19,9 +19,15 @@ const (
 
 var (
 	ErrInvalidOptions = errors.New("snapshot offload: invalid options")
-	ErrIntegrity      = errors.New("snapshot offload: integrity check failed")
-	ErrObjectConflict = errors.New("snapshot offload: object conflict")
-	ErrObjectNotFound = errors.New("snapshot offload: object not found")
+
+	// ErrRestoreGroupMismatch reports a restore whose manifest belongs to a
+	// different Raft group than the operator named. It is deliberately its
+	// own sentinel: an operator repeating the restore command across groups
+	// needs to see "wrong group", not a generic invalid-options error.
+	ErrRestoreGroupMismatch = errors.New("snapshot offload: manifest group does not match the requested group")
+	ErrIntegrity            = errors.New("snapshot offload: integrity check failed")
+	ErrObjectConflict       = errors.New("snapshot offload: object conflict")
+	ErrObjectNotFound       = errors.New("snapshot offload: object not found")
 
 	// ErrNoPersistedSnapshot reports that the LOCAL data dir has no
 	// persisted snapshot yet. It is deliberately distinct from
diff --git a/internal/snapshotoffload/offload_test.go b/internal/snapshotoffload/offload_test.go
index a8942cadd..8a54bcf7a 100644
--- a/internal/snapshotoffload/offload_test.go
+++ b/internal/snapshotoffload/offload_test.go
@@ -58,6 +58,7 @@ func TestPublishAndRestorePhysicalSnapshotRoundTrip(t *testing.T) {
 		Peers: []etcdraftengine.Peer{
 			{NodeID: 9, ID: "n9", Address: "127.0.0.1:19009"},
 		},
+		ExpectGroupID: expectGroup(manifest.GroupID),
 	})
 	require.NoError(t, err)
 	require.Equal(t, int64(len(payload)), result.PayloadBytes)
@@ -95,10 +96,11 @@ func TestRestoreRejectsCorruptPayloadAndLeavesDestinationAbsent(t *testing.T) {
 
 	restoreDataDir := filepath.Join(root, "restored")
 	_, err = RestorePhysicalSnapshot(ctx, RestoreOptions{
-		Store:       store,
-		ManifestKey: manifest.ManifestKey,
-		DataDir:     restoreDataDir,
-		Peers:       singlePeer(),
+		Store:         store,
+		ManifestKey:   manifest.ManifestKey,
+		DataDir:       restoreDataDir,
+		Peers:         singlePeer(),
+		ExpectGroupID: expectGroup(manifest.GroupID),
 	})
 	require.ErrorIs(t, err, ErrIntegrity)
 	_, statErr := os.Stat(restoreDataDir)
@@ -376,10 +378,11 @@ func TestRestoreInlineManifestRejectsStaleSelfHashBeforePayloadDownload(t *testi
 	tracked := &countingObjectStore{ObjectStore: store}
 
 	_, err = RestorePhysicalSnapshot(ctx, RestoreOptions{
-		Store:    tracked,
-		Manifest: &tampered,
-		DataDir:  filepath.Join(root, "restored"),
-		Peers:    singlePeer(),
+		Store:         tracked,
+		Manifest:      &tampered,
+		DataDir:       filepath.Join(root, "restored"),
+		Peers:         singlePeer(),
+		ExpectGroupID: expectGroup(tampered.GroupID),
 	})
 	require.ErrorIs(t, err, ErrIntegrity)
 	require.Zero(t, tracked.getObjectCalls)
@@ -480,10 +483,11 @@ func TestRestorePreflightsExistingDestinationBeforePayloadDownload(t *testing.T)
 	require.NoError(t, os.Mkdir(restoreDataDir, 0o755))
 
 	_, err = RestorePhysicalSnapshot(ctx, RestoreOptions{
-		Store:       store,
-		ManifestKey: manifest.ManifestKey,
-		DataDir:     restoreDataDir,
-		Peers:       singlePeer(),
+		Store:         store,
+		ManifestKey:   manifest.ManifestKey,
+		DataDir:       restoreDataDir,
+		Peers:         singlePeer(),
+		ExpectGroupID: expectGroup(manifest.GroupID),
 	})
 	require.ErrorIs(t, err, etcdraftengine.ErrExternalSnapshotRestoreExists)
 }
@@ -511,6 +515,7 @@ func TestRestoreRejectsInvalidPeersBeforePayloadDownload(t *testing.T) {
 		Peers: []etcdraftengine.Peer{
 			{NodeID: 0, ID: "n0", Address: "127.0.0.1:12000"},
 		},
+		ExpectGroupID: expectGroup(manifest.GroupID),
 	})
 	require.ErrorIs(t, err, ErrInvalidOptions)
 	require.Zero(t, tracked.getObjectCalls)
@@ -534,10 +539,11 @@ func TestRestoreHonorsCancelledContextBeforePayloadDownload(t *testing.T) {
 	tracked := &countingObjectStore{ObjectStore: store}
 
 	_, err = RestorePhysicalSnapshot(ctx, RestoreOptions{
-		Store:    tracked,
-		Manifest: manifest,
-		DataDir:  filepath.Join(root, "restored"),
-		Peers:    singlePeer(),
+		Store:         tracked,
+		Manifest:      manifest,
+		DataDir:       filepath.Join(root, "restored"),
+		Peers:         singlePeer(),
+		ExpectGroupID: expectGroup(manifest.GroupID),
 	})
 	require.ErrorIs(t, err, context.Canceled)
 	require.Zero(t, tracked.getObjectCalls)
@@ -724,3 +730,10 @@ func TestPublishVerifiesLeadershipAfterTheManifestAbsenceProbe(t *testing.T) {
 	require.NotContains(t, verifiedAfter, "put-manifest",
 		"and before the manifest object is created")
 }
+
+// expectGroup is the RestoreOptions.ExpectGroupID helper. The field is a
+// pointer because group 0 is a real group (the dedicated TSO group), so zero
+// cannot double as "not supplied".
+func expectGroup(groupID uint64) *uint64 {
+	return &groupID
+}
diff --git a/internal/snapshotoffload/restore.go b/internal/snapshotoffload/restore.go
index 0a45e819b..5d7008a57 100644
--- a/internal/snapshotoffload/restore.go
+++ b/internal/snapshotoffload/restore.go
@@ -20,6 +20,23 @@ type RestoreOptions struct {
 	Manifest    *Manifest
 	DataDir     string
 	Peers       []etcdraftengine.Peer
+
+	// ExpectGroupID is the Raft group the operator believes this data
+	// directory belongs to. Required.
+	//
+	// Nothing downstream carries the group's identity: the restored
+	// artifacts record index, term, peers and payload hash, but the group
+	// comes only from the manifest, and startup derives the group from the
+	// directory layout instead. So a group-2 manifest restored into a
+	// group-1 data directory produces a perfectly valid-looking
+	// directory that startup then loads as group 1 -- the wrong physical
+	// FSM under another group's routing identity, with no error anywhere.
+	// The only place that mistake can be caught is here, against what the
+	// operator says they intended.
+	//
+	// A pointer because group 0 is a real group (the dedicated TSO group),
+	// so zero cannot double as "unset".
+	ExpectGroupID *uint64
 }
 
 const (
@@ -89,6 +106,11 @@ func prepareRestorePayload(ctx context.Context, opts RestoreOptions) (Manifest,
 	if err := validateManifest(manifest); err != nil {
 		return Manifest{}, "", nil, err
 	}
+	// Before the download and before the destination exists, so a
+	// mistaken manifest key costs nothing and leaves nothing behind.
+	if err := checkRestoreGroup(manifest, opts.ExpectGroupID); err != nil {
+		return Manifest{}, "", nil, err
+	}
 	if err := checkRestorePreflight(ctx, opts.DataDir); err != nil {
 		return Manifest{}, "", nil, err
 	}
@@ -210,11 +232,27 @@ func validateRestoreOptions(opts RestoreOptions) error {
 		return errors.Wrap(ErrInvalidOptions, "data dir is required")
 	case len(opts.Peers) == 0:
 		return errors.Wrap(ErrInvalidOptions, "restore peers are required")
+	case opts.ExpectGroupID == nil:
+		return errors.Wrap(ErrInvalidOptions, "expected raft group id is required")
 	default:
 		return validateRestorePeers(opts.Peers)
 	}
 }
 
+// checkRestoreGroup rejects a manifest belonging to a different group than
+// the operator asked to restore.
+func checkRestoreGroup(manifest Manifest, expect *uint64) error {
+	if expect == nil {
+		return errors.Wrap(ErrInvalidOptions, "expected raft group id is required")
+	}
+	if manifest.GroupID != *expect {
+		return errors.Wrapf(ErrRestoreGroupMismatch,
+			"manifest %s belongs to group %d, not the requested group %d",
+			manifest.ManifestKey, manifest.GroupID, *expect)
+	}
+	return nil
+}
+
 func validateRestorePeers(peers []etcdraftengine.Peer) error {
 	seenNodeIDs := make(map[uint64]struct{}, len(peers))
 	seenIDs := make(map[string]struct{}, len(peers))
diff --git a/internal/snapshotoffload/s3_store_test.go b/internal/snapshotoffload/s3_store_test.go
index 010f65981..f795abf7c 100644
--- a/internal/snapshotoffload/s3_store_test.go
+++ b/internal/snapshotoffload/s3_store_test.go
@@ -46,6 +46,7 @@ func TestPublishAndRestorePhysicalSnapshotRoundTripWithS3Store(t *testing.T) {
 		Peers: []etcdraftengine.Peer{
 			{NodeID: 2, ID: "n2", Address: "127.0.0.1:12002"},
 		},
+		ExpectGroupID: expectGroup(manifest.GroupID),
 	})
 	require.NoError(t, err)
 	require.Equal(t, manifest.Payload.SHA256, result.PayloadSHA256)
diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go
index f879f172a..ad2e28bda 100644
--- a/main_snapshot_offload_test.go
+++ b/main_snapshot_offload_test.go
@@ -157,3 +157,84 @@ func TestRunbookRestorePathsMatchGroupDataDir(t *testing.T) {
 		})
 	}
 }
+
+// TestRunbookRestorePathsFollowFromTheGroupTopology pins the runbook table the
+// way an operator actually reads it: from --raftGroups to a directory.
+//
+// TestRunbookRestorePathsMatchGroupDataDir above takes `multi` as an input, so
+// it cannot catch the case where a reader derives the wrong `multi` in the first
+// place — and that is the case that bites. dataGroupsNeedMultiDirs counts DATA
+// groups and excludes group 0, so a node running the dedicated TSO group
+// alongside a single data group has two entries in --raftGroups but is NOT
+// multi-dir: group 0 lands in group-0 while the data group opens
+// <raftDir>/<raftID> directly. Restoring that data group into group-1 puts it
+// where startup never looks, and an empty group is not an error.
+func TestRunbookRestorePathsFollowFromTheGroupTopology(t *testing.T) {
+	t.Parallel()
+
+	const (
+		raftDir = "/var/lib/elastickv"
+		raftID  = "n1"
+	)
+	spec := func(ids ...uint64) []groupSpec {
+		out := make([]groupSpec, 0, len(ids))
+		for _, id := range ids {
+			out = append(out, groupSpec{id: id, address: "127.0.0.1:50051"})
+		}
+		return out
+	}
+
+	for _, tc := range []struct {
+		name    string
+		groups  []groupSpec
+		groupID uint64
+		want    string
+	}{
+		{
+			name:   "two data groups: each gets its own dir",
+			groups: spec(1, 2), groupID: 1,
+			want: "/var/lib/elastickv/n1/group-1",
+		},
+		{
+			name:   "two data groups: the second one too",
+			groups: spec(1, 2), groupID: 2,
+			want: "/var/lib/elastickv/n1/group-2",
+		},
+		{
+			name:   "a single data group opens the node dir",
+			groups: spec(1), groupID: 1,
+			want: "/var/lib/elastickv/n1",
+		},
+		{
+			name:   "dedicated TSO plus one data group: group 0 is always group-0",
+			groups: spec(0, 1), groupID: 0,
+			want: "/var/lib/elastickv/n1/group-0",
+		},
+		{
+			// The row that catches people out: two --raftGroups entries but
+			// only one DATA group, so this is not a multi-dir deployment.
+			name:   "dedicated TSO plus one data group: the data group is NOT group-1",
+			groups: spec(0, 1), groupID: 1,
+			want: "/var/lib/elastickv/n1",
+		},
+		{
+			name:   "dedicated TSO plus two data groups is multi-dir again",
+			groups: spec(0, 1, 2), groupID: 1,
+			want: "/var/lib/elastickv/n1/group-1",
+		},
+		{
+			name:   "dedicated TSO plus two data groups: group 0 unchanged",
+			groups: spec(0, 1, 2), groupID: 0,
+			want: "/var/lib/elastickv/n1/group-0",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			// requested=true is the operator asking for per-group dirs; the
+			// topology decides whether that takes effect.
+			multi := effectiveMultiDataDirs(tc.groups, true)
+			require.Equal(t, tc.want, groupDataDir(raftDir, raftID, tc.groupID, multi),
+				"docs/snapshot_offload_operations.md documents this path for restore")
+		})
+	}
+}