snapshotoffload: add the M2 leader-only publish scheduler

cmd/elastickv-snapshot-offload/main.go

@@ -101,6 +101,11 @@ func classifyError(err error) int {
 	switch {
 	case errors.Is(err, snapshotoffload.ErrIntegrity),
 		errors.Is(err, snapshotoffload.ErrObjectNotFound),
+		// Splitting ErrNoPersistedSnapshot out of ErrObjectNotFound
+		// must not change the CLI contract: automation distinguishes
+		// "missing/invalid snapshot data" (2) from "bad invocation"
+		// (1), and a data dir with no snapshot is the former.
+		errors.Is(err, snapshotoffload.ErrNoPersistedSnapshot),
 		errors.Is(err, etcd.ErrExternalSnapshotRestoreInvalid),
 		errors.Is(err, etcd.ErrExternalSnapshotRestoreSHA256):
 		return exitDataErr

cmd/elastickv-snapshot-offload/main_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/bootjp/elastickv/internal/raftengine/etcd"
 	"github.com/bootjp/elastickv/internal/snapshotoffload"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -131,3 +132,48 @@ func seedCLISnapshot(t *testing.T, root string, payload []byte, index uint64, te
 	require.NoError(t, err)
 	return dataDir
 }
+
+// TestClassifyErrorKeepsMissingSnapshotAsADataError pins the CLI exit
+// contract across the ErrNoPersistedSnapshot split.
+//
+// Automation distinguishes "missing or invalid snapshot data" (2) from
+// "bad invocation" (1). Giving the missing-local-snapshot case its own
+// sentinel — so the scheduler could stop treating a vanished remote
+// object as a routine skip — must not silently move it to exit 1.
+func TestClassifyErrorKeepsMissingSnapshotAsADataError(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  error
+		want int
+	}{
+		{
+			name: "data dir has no persisted snapshot",
+			err:  errors.Wrap(snapshotoffload.ErrNoPersistedSnapshot, "publish"),
+			want: exitDataErr,
+		},
+		{
+			name: "object absent from the store",
+			err:  errors.Wrap(snapshotoffload.ErrObjectNotFound, "publish"),
+			want: exitDataErr,
+		},
+		{
+			name: "integrity failure",
+			err:  errors.Wrap(snapshotoffload.ErrIntegrity, "restore"),
+			want: exitDataErr,
+		},
+		{
+			name: "invalid invocation",
+			err:  errors.Wrap(snapshotoffload.ErrInvalidOptions, "publish"),
+			want: exitUserErr,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			require.Equal(t, tc.want, classifyError(tc.err))
+		})
+	}
+}

docs/design/2026_07_19_partial_physical_snapshot_object_offload.md

@@ -1,6 +1,6 @@
 # Physical Snapshot Object Offload
 
-Status: Partial — M0/M1 implemented; M2/M3 pending
+Status: Partial — M0/M1/M2 implemented; M3 pending
 Author: bootjp
 Date: 2026-07-19
 Updated: 2026-07-23
@@ -164,7 +164,7 @@ permissions below the configured prefix.
 |---|---|---|
 | M0 | Persisted snapshot export handle, complete-payload restore preparation, focused design | Implemented in the first substrate PR |
 | M1 | Object client interface, S3-compatible implementation, immutable payload/manifest publication, download verification, operator CLI | Implemented: local and S3 stores, manifest schema, payload-first publish, verified restore, and publish/restore CLI |
-| M2 | Leader-only per-group scheduler, metrics, jitter, concurrency bounds, cancellation and restart idempotency | Pending |
+| M2 | Leader-only per-group scheduler, metrics, jitter, concurrency bounds, cancellation and restart idempotency | Implemented: `internal/snapshotoffload/scheduler.go`. Leadership is checked before the snapshot is opened and re-checked immediately before the manifest commit via `PublishOptions.VerifyLeader`; uploads are bounded (default one per process) with interval jitter; cancellation is treated as shutdown rather than publish failure; restart idempotency comes from the object store, since publish reuses a matching committed manifest. Not yet wired into `main.go` — the runtime flags are M3. |
 | M3 | Retention/GC, restore drills, corruption tests, multi-node acceptance, operational documentation | Pending |
 
 The filename and header remain `partial` until M1-M3 complete the central

internal/snapshotoffload/manifest.go

@@ -22,6 +22,20 @@ var (
 	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
+	// ErrObjectNotFound: a young group that has not snapshotted is a
+	// normal scan outcome, whereas an object vanishing from the store
+	// mid-publish is a real failure, and collapsing the two would
+	// silence the second.
+	ErrNoPersistedSnapshot = errors.New("snapshot offload: no persisted snapshot available")
+
+	// ErrSnapshotNotNewer reports that the persisted snapshot is not
+	// newer than the caller's high-water mark, so nothing was
+	// published. It is a normal outcome for a scheduler tick over an
+	// unchanged snapshot, not a failure.
+	ErrSnapshotNotNewer = errors.New("snapshot offload: persisted snapshot is not newer than the last published index")
 )
 
 type Manifest struct {

internal/snapshotoffload/offload_test.go

@@ -244,7 +244,7 @@ func TestPutManifestReusesExistingManifestAfterCreateConflict(t *testing.T) {
 	candidate := existing
 	candidate.CreatedAt = time.Unix(401, 0).UTC()
 	racingStore := &headMissOnceStore{ObjectStore: store, key: key}
-	require.NoError(t, putManifest(ctx, racingStore, &candidate, true))
+	require.NoError(t, putManifest(ctx, racingStore, &candidate, true, nil))
 	require.Equal(t, existing.CreatedAt, candidate.CreatedAt)
 	require.NotEmpty(t, candidate.ManifestSHA256)
 }
@@ -662,3 +662,65 @@ func (s *headMissOnceStore) HeadObject(ctx context.Context, key string) (ObjectI
 func singlePeer() []etcdraftengine.Peer {
 	return []etcdraftengine.Peer{{NodeID: 1, ID: "n1", Address: "127.0.0.1:12001"}}
 }
+
+// headOrderingStore records the order of remote calls so a test can
+// prove the leadership recheck happens after the manifest absence
+// probe rather than before it.
+type headOrderingStore struct {
+	ObjectStore
+	manifestKey string
+	calls       []string
+}
+
+func (s *headOrderingStore) HeadObject(ctx context.Context, key string) (ObjectInfo, bool, error) {
+	if key == s.manifestKey {
+		s.calls = append(s.calls, "head-manifest")
+	}
+	return s.ObjectStore.HeadObject(ctx, key)
+}
+
+func (s *headOrderingStore) PutObject(
+	ctx context.Context, key string, body io.Reader, opts PutOptions,
+) (ObjectInfo, error) {
+	if key == s.manifestKey {
+		s.calls = append(s.calls, "put-manifest")
+	}
+	return s.ObjectStore.PutObject(ctx, key, body, opts)
+}
+
+// TestPublishVerifiesLeadershipAfterTheManifestAbsenceProbe pins the §4
+// ordering. The absence probe is a remote read with latency nothing in
+// the caller controls; checking leadership before it leaves a window in
+// which a node demoted during that read still commits a manifest —
+// exactly the guarantee the scheduler exists to provide.
+func TestPublishVerifiesLeadershipAfterTheManifestAbsenceProbe(t *testing.T) {
+	ctx := context.Background()
+	root := t.TempDir()
+	payload := []byte("EKVTHLC1ordering-payload")
+	sourceDataDir := seedPhysicalSnapshot(t, root, payload, 31, 6, singlePeer())
+	local := newTestLocalStore(t, filepath.Join(root, "objects"))
+
+	key, err := manifestKey("cluster-a", 1, 31, 6)
+	require.NoError(t, err)
+	ordering := &headOrderingStore{ObjectStore: local, manifestKey: key}
+
+	var verifiedAfter []string
+	_, err = PublishPersistedSnapshot(ctx, PublishOptions{
+		Store:         ordering,
+		DataDir:       sourceDataDir,
+		Prefix:        "cluster-a",
+		GroupID:       1,
+		SourceCluster: "cluster-a",
+		VerifyLeader: func(context.Context) error {
+			// Snapshot the calls seen so far at verification time.
+			verifiedAfter = append([]string(nil), ordering.calls...)
+			return nil
+		},
+	})
+	require.NoError(t, err)
+
+	require.Contains(t, verifiedAfter, "head-manifest",
+		"leadership must be re-verified AFTER the manifest absence probe")
+	require.NotContains(t, verifiedAfter, "put-manifest",
+		"and before the manifest object is created")
+}

internal/snapshotoffload/publish.go

@@ -25,6 +25,24 @@ type PublishOptions struct {
 	BinaryVersion string
 	CreatedAt     time.Time
 	SpoolDir      string
+	// VerifyLeader, when set, is re-checked immediately before the manifest is
+	// committed. §4 requires leadership to hold at that instant, not merely when
+	// the snapshot was opened: spooling a multi-gigabyte payload takes long
+	// enough to lose an election. Failing here can leave an unreferenced
+	// content-addressed payload, which GC reclaims, but never a committed
+	// manifest naming a snapshot this node no longer had the right to publish.
+	VerifyLeader func(context.Context) error
+	// SkipIfNotNewerThan suppresses the publish when the persisted
+	// snapshot's index is not greater than this value. Zero disables
+	// the check.
+	//
+	// The comparison happens after the export is opened but BEFORE
+	// the payload is spooled, which is the whole point: a scheduler
+	// that ticks every 15 minutes over an unchanged snapshot would
+	// otherwise re-read, re-hash and re-fsync a multi-gigabyte
+	// payload every tick just to discover the object store already
+	// has it.
+	SkipIfNotNewerThan uint64
 }
 
 func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manifest, error) {
@@ -38,6 +56,10 @@ func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manife
 	defer func() { _ = export.Close() }()
 
 	metadata := export.Metadata()
+	if opts.SkipIfNotNewerThan > 0 && metadata.Index <= opts.SkipIfNotNewerThan {
+		return nil, errors.Wrapf(ErrSnapshotNotNewer,
+			"persisted snapshot index %d is not newer than %d", metadata.Index, opts.SkipIfNotNewerThan)
+	}
 	payloadFile, payloadSHA, payloadBytes, err := spoolExport(ctx, export, publishSpoolDir(opts))
 	if err != nil {
 		return nil, err
@@ -56,14 +78,27 @@ func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manife
 	if err := putPayload(ctx, opts.Store, payloadObjectKey, payloadFile, payloadBytes, payloadSHA); err != nil {
 		return nil, err
 	}
+	return commitManifest(ctx, opts, metadata, payloadObjectKey, payloadSHA)
+}
+
+// commitManifest builds, validates and commits the manifest once the payload is
+// durable. Split out of PublishPersistedSnapshot to keep that function inside
+// the cyclop budget after the leadership re-check landed.
+func commitManifest(
+	ctx context.Context,
+	opts PublishOptions,
+	metadata etcdraftengine.PersistedSnapshotExportMetadata,
+	payloadObjectKey string,
+	payloadSHA string,
+) (*Manifest, error) {
 	manifest, err := buildManifest(opts, metadata, payloadObjectKey, payloadSHA)
 	if err != nil {
 		return nil, err
 	}
 	if err := validateManifest(*manifest); err != nil {
 		return nil, err
 	}
-	if err := putManifest(ctx, opts.Store, manifest, opts.CreatedAt.IsZero()); err != nil {
+	if err := putManifest(ctx, opts.Store, manifest, opts.CreatedAt.IsZero(), opts.VerifyLeader); err != nil {
 		return nil, err
 	}
 	return manifest, nil
@@ -75,7 +110,7 @@ func openPublishExport(dataDir string) (*etcdraftengine.PersistedSnapshotExport,
 		return nil, errors.Wrap(err, "open persisted snapshot export")
 	}
 	if !ok {
-		return nil, errors.Wrap(ErrObjectNotFound, "no persisted snapshot available")
+		return nil, errors.WithStack(ErrNoPersistedSnapshot)
 	}
 	return export, nil
 }
@@ -113,7 +148,13 @@ func buildManifest(
 	}, nil
 }
 
-func putManifest(ctx context.Context, store ObjectStore, manifest *Manifest, reuseExistingCreatedAt bool) error {
+func putManifest(
+	ctx context.Context,
+	store ObjectStore,
+	manifest *Manifest,
+	reuseExistingCreatedAt bool,
+	verifyLeader func(context.Context) error,
+) error {
 	data, manifestSHA, err := manifest.MarshalCanonical()
 	if err != nil {
 		return err
@@ -125,6 +166,17 @@ func putManifest(ctx context.Context, store ObjectStore, manifest *Manifest, reu
 	} else if exists {
 		return nil
 	}
+	// §4: leadership must hold at the instant the manifest is created,
+	// not merely before the absence probe above. That probe is a remote
+	// read whose latency is unbounded by anything the caller controls,
+	// so checking before it leaves a window in which a demoted node
+	// still commits a manifest — precisely the guarantee this
+	// scheduler exists to provide.
+	if verifyLeader != nil {
+		if err := verifyLeader(ctx); err != nil {
+			return errors.Wrap(err, "snapshot offload: leadership lost before manifest commit")
+		}
+	}
 	if err := createManifestObject(ctx, store, manifest, data, size, objectSHA, reuseExistingCreatedAt); err != nil {
 		return err
 	}
@@ -231,9 +283,22 @@ func manifestMatchesCandidate(existing Manifest, candidate Manifest, reuseExisti
 	return reflect.DeepEqual(existing, candidate)
 }
 
+// sameManifestExceptCreation compares a retry's candidate against the
+// committed manifest, ignoring the fields that legitimately differ
+// between two publishes of the SAME snapshot.
+//
+// BinaryVersion is one of them. It records which binary published the
+// artifact, not anything about the snapshot itself, so after an
+// upgrade a process that republishes an index it has not yet published
+// locally would otherwise conflict with the manifest the previous
+// binary committed — and keep failing every scan until Raft happens to
+// produce a new snapshot. The committed manifest keeps the original
+// publisher's version, which is the correct audit record for the
+// bytes that actually exist.
 func sameManifestExceptCreation(existing Manifest, candidate Manifest) bool {
 	candidate.CreatedAt = existing.CreatedAt
 	candidate.ManifestSHA256 = existing.ManifestSHA256
+	candidate.BinaryVersion = existing.BinaryVersion
 	return reflect.DeepEqual(existing, candidate)
 }