From ded73abbef576d0eadfe9a81e6ac2163619b87a7 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Sun, 6 Sep 2026 15:33:11 +0900
Subject: [PATCH 1/7] snapshotoffload: add the M2 leader-only publish scheduler
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

M0/M1 left the substrate reachable only from the CLI: manifest, publish
and restore existed, but nothing scanned groups or published on its own.
This is M2 from docs/design/2026_07_19_partial_physical_snapshot_object_offload.md.

Per §4:

- Only the current group leader publishes. IsLeader is the cheap
  pre-check so a follower never opens the snapshot, and VerifyLeader is
  re-run immediately before the manifest commit. Spooling a multi-GB
  payload is long enough to lose an election, and the design is explicit
  that losing it may strand a content-addressed payload -- which GC
  reclaims -- but must never commit a manifest.
- Uploads are bounded, one per process by default, so a node hosting many
  groups cannot saturate its uplink.
- Interval jitter spreads multi-group work off a common tick.
- Cancellation is shutdown, not a publish failure: it is neither reported
  to the observer nor logged as an error.
- Restart idempotency comes from the object store rather than local
  state. A fresh process republishing the same index reuses the committed
  manifest, which publish already handled; the in-memory index map is an
  optimisation only, and the test asserts a restarted scheduler succeeds
  with no local record.
- The scheduler never asks the state machine for a snapshot. Cadence
  stays owned by the Raft engine, which §10 lists as a non-goal.

Metrics go through a SchedulerObserver interface so the monitoring
registry can implement it in M3 without this package importing it.

Not yet wired into main.go: the operator flags and the retention/GC half
are M3, and the doc now records M2 as implemented rather than pending.
---
 ...artial_physical_snapshot_object_offload.md |   4 +-
 internal/snapshotoffload/publish.go           |  25 ++
 internal/snapshotoffload/scheduler.go         | 261 ++++++++++++++++++
 internal/snapshotoffload/scheduler_test.go    | 209 ++++++++++++++
 4 files changed, 497 insertions(+), 2 deletions(-)
 create mode 100644 internal/snapshotoffload/scheduler.go
 create mode 100644 internal/snapshotoffload/scheduler_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 7bf1765a6..2956620aa 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
@@ -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
diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go
index 734359d20..aef47f40c 100644
--- a/internal/snapshotoffload/publish.go
+++ b/internal/snapshotoffload/publish.go
@@ -25,6 +25,13 @@ 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
 }
 
 func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manifest, error) {
@@ -56,6 +63,19 @@ 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
@@ -63,6 +83,11 @@ func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manife
 	if err := validateManifest(*manifest); err != nil {
 		return nil, err
 	}
+	if opts.VerifyLeader != nil {
+		if err := opts.VerifyLeader(ctx); err != nil {
+			return nil, errors.Wrap(err, "snapshot offload: leadership lost before manifest commit")
+		}
+	}
 	if err := putManifest(ctx, opts.Store, manifest, opts.CreatedAt.IsZero()); err != nil {
 		return nil, err
 	}
diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
new file mode 100644
index 000000000..e7494fcff
--- /dev/null
+++ b/internal/snapshotoffload/scheduler.go
@@ -0,0 +1,261 @@
+package snapshotoffload
+
+import (
+	"context"
+	"log/slog"
+	"math/rand/v2"
+	"sync"
+	"time"
+
+	"github.com/cockroachdb/errors"
+)
+
+// Scheduler is the §4 leader-only publisher: each process scans its own Raft
+// groups on an interval and offloads a persisted snapshot when one exists that
+// has not been published yet.
+//
+// It never asks the state machine for a snapshot. Snapshot cadence stays owned
+// by the Raft engine, so this milestone can only publish what the engine has
+// already persisted -- an offload that forced its own snapshot would change
+// compaction behaviour, which §10 lists as a non-goal.
+type Scheduler struct {
+	groups      []OffloadGroup
+	store       ObjectStore
+	prefix      string
+	sourceName  string
+	binVersion  string
+	spoolDir    string
+	interval    time.Duration
+	jitter      time.Duration
+	concurrency int
+	observer    SchedulerObserver
+	logger      *slog.Logger
+	now         func() time.Time
+	// published caches the highest index this process has published per group.
+	// It is an optimisation only: restart idempotency comes from the object
+	// store, not from this map.
+	mu        sync.Mutex
+	published map[uint64]uint64
+}
+
+// OffloadGroup is one local Raft group the scheduler may publish for.
+type OffloadGroup struct {
+	GroupID uint64
+	DataDir string
+	// IsLeader is the cheap pre-check made before opening the snapshot.
+	IsLeader func() bool
+	// VerifyLeader is the authoritative check, re-run immediately before the
+	// manifest is committed. See PublishOptions.VerifyLeader.
+	VerifyLeader func(context.Context) error
+}
+
+// SchedulerObserver receives per-attempt outcomes for metrics.
+type SchedulerObserver interface {
+	ObserveSnapshotOffloadPublished(groupID, index uint64, payloadBytes int64, elapsed time.Duration)
+	ObserveSnapshotOffloadSkipped(groupID uint64, reason string)
+	ObserveSnapshotOffloadFailed(groupID uint64, err error)
+}
+
+type nopSchedulerObserver struct{}
+
+func (nopSchedulerObserver) ObserveSnapshotOffloadPublished(uint64, uint64, int64, time.Duration) {}
+func (nopSchedulerObserver) ObserveSnapshotOffloadSkipped(uint64, string)                         {}
+func (nopSchedulerObserver) ObserveSnapshotOffloadFailed(uint64, error)                           {}
+
+// Default scheduling parameters from §4.
+const (
+	DefaultSchedulerInterval    = 15 * time.Minute
+	DefaultSchedulerConcurrency = 1
+)
+
+type SchedulerOption func(*Scheduler)
+
+func WithSchedulerInterval(d time.Duration) SchedulerOption {
+	return func(s *Scheduler) {
+		if d > 0 {
+			s.interval = d
+		}
+	}
+}
+
+// WithSchedulerJitter spreads multi-group work so every group in a process does
+// not contend for the upload slot on the same tick.
+func WithSchedulerJitter(d time.Duration) SchedulerOption {
+	return func(s *Scheduler) {
+		if d >= 0 {
+			s.jitter = d
+		}
+	}
+}
+
+func WithSchedulerConcurrency(n int) SchedulerOption {
+	return func(s *Scheduler) {
+		if n > 0 {
+			s.concurrency = n
+		}
+	}
+}
+
+func WithSchedulerObserver(o SchedulerObserver) SchedulerOption {
+	return func(s *Scheduler) {
+		if o != nil {
+			s.observer = o
+		}
+	}
+}
+
+func WithSchedulerLogger(l *slog.Logger) SchedulerOption {
+	return func(s *Scheduler) {
+		if l != nil {
+			s.logger = l
+		}
+	}
+}
+
+func WithSchedulerClock(now func() time.Time) SchedulerOption {
+	return func(s *Scheduler) {
+		if now != nil {
+			s.now = now
+		}
+	}
+}
+
+func WithSchedulerSpoolDir(dir string) SchedulerOption {
+	return func(s *Scheduler) { s.spoolDir = dir }
+}
+
+// NewScheduler builds the offload scheduler. It is opt-in: callers construct it
+// only when object offload is configured.
+func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluster, binaryVersion string, opts ...SchedulerOption) *Scheduler {
+	s := &Scheduler{
+		groups:      groups,
+		store:       store,
+		prefix:      prefix,
+		sourceName:  sourceCluster,
+		binVersion:  binaryVersion,
+		interval:    DefaultSchedulerInterval,
+		jitter:      DefaultSchedulerInterval / 4, //nolint:mnd // a quarter interval spreads groups without doubling the period.
+		concurrency: DefaultSchedulerConcurrency,
+		observer:    nopSchedulerObserver{},
+		logger:      slog.Default().With(slog.String("component", "snapshot-offload")),
+		now:         time.Now,
+		published:   make(map[uint64]uint64),
+	}
+	for _, opt := range opts {
+		opt(s)
+	}
+	return s
+}
+
+func (s *Scheduler) validate() error {
+	switch {
+	case s.store == nil:
+		return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires an object store")
+	case s.sourceName == "":
+		return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires a source cluster name")
+	}
+	return nil
+}
+
+// Run scans on the configured interval until ctx is cancelled. Cancellation is
+// the only stop condition; a failing group is retried on the next tick rather
+// than tearing the loop down, because an object store outage must not stop the
+// process.
+func (s *Scheduler) Run(ctx context.Context) error {
+	if ctx == nil {
+		return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler context is required")
+	}
+	if err := s.validate(); err != nil {
+		return err
+	}
+	timer := time.NewTimer(s.nextDelay())
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-timer.C:
+			s.SyncOnce(ctx)
+			timer.Reset(s.nextDelay())
+		}
+	}
+}
+
+func (s *Scheduler) nextDelay() time.Duration {
+	if s.jitter <= 0 {
+		return s.interval
+	}
+	return s.interval + time.Duration(rand.Int64N(int64(s.jitter))) //nolint:gosec // scheduling jitter, not a security decision.
+}
+
+// SyncOnce runs one scan across every local group, bounded by the upload
+// concurrency limit. Exported so tests and operators can force a pass.
+func (s *Scheduler) SyncOnce(ctx context.Context) {
+	sem := make(chan struct{}, s.concurrency)
+	var wg sync.WaitGroup
+	for _, group := range s.groups {
+		if ctx.Err() != nil {
+			break
+		}
+		wg.Add(1)
+		go func(g OffloadGroup) {
+			defer wg.Done()
+			select {
+			case sem <- struct{}{}:
+			case <-ctx.Done():
+				return
+			}
+			defer func() { <-sem }()
+			s.publishGroup(ctx, g)
+		}(group)
+	}
+	wg.Wait()
+}
+
+func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
+	// Cheap pre-check first: a follower must not even open the snapshot.
+	if group.IsLeader != nil && !group.IsLeader() {
+		s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "not_leader")
+		return
+	}
+	started := s.now()
+	manifest, err := PublishPersistedSnapshot(ctx, PublishOptions{
+		Store:         s.store,
+		DataDir:       group.DataDir,
+		Prefix:        s.prefix,
+		GroupID:       group.GroupID,
+		SourceCluster: s.sourceName,
+		BinaryVersion: s.binVersion,
+		SpoolDir:      s.spoolDir,
+		VerifyLeader:  group.VerifyLeader,
+	})
+	if err != nil {
+		if errors.Is(err, context.Canceled) || ctx.Err() != nil {
+			return
+		}
+		s.observer.ObserveSnapshotOffloadFailed(group.GroupID, err)
+		s.logger.WarnContext(ctx, "snapshot offload publish failed",
+			slog.Uint64("group_id", group.GroupID), slog.String("error", err.Error()))
+		return
+	}
+	s.markPublished(group.GroupID, manifest.SnapshotIndex)
+	s.observer.ObserveSnapshotOffloadPublished(
+		group.GroupID, manifest.SnapshotIndex, manifest.Payload.Bytes, s.now().Sub(started))
+}
+
+func (s *Scheduler) markPublished(groupID, index uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if index > s.published[groupID] {
+		s.published[groupID] = index
+	}
+}
+
+// LastPublishedIndex reports the highest index this process has published for a
+// group. Zero means "nothing published by this process", not "nothing
+// published": another node or a previous run may hold newer manifests.
+func (s *Scheduler) LastPublishedIndex(groupID uint64) uint64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.published[groupID]
+}
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
new file mode 100644
index 000000000..52e7efb93
--- /dev/null
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -0,0 +1,209 @@
+package snapshotoffload
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+type recordingObserver struct {
+	mu        sync.Mutex
+	published []uint64
+	skipped   []string
+	failed    []error
+}
+
+func (o *recordingObserver) ObserveSnapshotOffloadPublished(groupID, _ uint64, _ int64, _ time.Duration) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.published = append(o.published, groupID)
+}
+
+func (o *recordingObserver) ObserveSnapshotOffloadSkipped(_ uint64, reason string) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.skipped = append(o.skipped, reason)
+}
+
+func (o *recordingObserver) ObserveSnapshotOffloadFailed(_ uint64, err error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.failed = append(o.failed, err)
+}
+
+func (o *recordingObserver) snapshot() ([]uint64, []string, []error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return append([]uint64(nil), o.published...), append([]string(nil), o.skipped...), append([]error(nil), o.failed...)
+}
+
+const schedulerTestIndex = 42
+
+func seedSchedulerGroup(t *testing.T, root, name string) string {
+	t.Helper()
+	dir := filepath.Join(root, name)
+	require.NoError(t, os.MkdirAll(dir, 0o750))
+	return seedPhysicalSnapshot(t, dir, []byte("EKVTHLC1scheduler-payload"), schedulerTestIndex, 3,
+		[]etcdraftengine.Peer{{NodeID: 1, ID: "n1", Address: "127.0.0.1:1"}})
+}
+
+// §4: only the current group leader may publish, and a follower must not even
+// open the snapshot.
+func TestSchedulerSkipsGroupsThisNodeDoesNotLead(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "g")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+
+	s := NewScheduler(store, []OffloadGroup{{
+		GroupID:  7,
+		DataDir:  dataDir,
+		IsLeader: func() bool { return false },
+	}}, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+
+	s.SyncOnce(context.Background())
+
+	published, skipped, failed := obs.snapshot()
+	require.Empty(t, published, "a follower must publish nothing")
+	require.Empty(t, failed)
+	require.Equal(t, []string{"not_leader"}, skipped)
+	require.Zero(t, s.LastPublishedIndex(7))
+}
+
+// §4: leadership is re-checked immediately before the manifest. Losing it in
+// the window may strand a content-addressed payload, which GC reclaims, but
+// must never commit a manifest.
+func TestSchedulerDoesNotCommitManifestWhenLeadershipIsLostWhileSpooling(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "g")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+	lost := errors.New("leadership lost")
+
+	s := NewScheduler(store, []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return lost },
+	}}, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+
+	s.SyncOnce(context.Background())
+
+	published, _, failed := obs.snapshot()
+	require.Empty(t, published)
+	require.Len(t, failed, 1)
+	require.ErrorIs(t, failed[0], lost)
+
+	manifestObjectKey, err := manifestKey("cluster-a", 7, 42, 3)
+	require.NoError(t, err)
+	_, ok, err := store.HeadObject(context.Background(), manifestObjectKey)
+	require.NoError(t, err)
+	require.False(t, ok, "no manifest may be committed after leadership is lost")
+}
+
+// A leader publishes, and re-running the scan is idempotent: the second pass
+// reuses the committed manifest rather than producing a second one. Restart
+// safety comes from the object store, so a fresh Scheduler behaves the same.
+func TestSchedulerPublishesOnceAndIsIdempotentAcrossRestart(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "g")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+	groups := []OffloadGroup{{GroupID: 7, DataDir: dataDir, IsLeader: func() bool { return true }}}
+
+	s := NewScheduler(store, groups, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	s.SyncOnce(context.Background())
+	require.Equal(t, uint64(42), s.LastPublishedIndex(7))
+
+	// Same process, second scan.
+	s.SyncOnce(context.Background())
+	// A different process that has published nothing itself.
+	restarted := NewScheduler(store, groups, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	require.Zero(t, restarted.LastPublishedIndex(7), "a fresh process starts with no local record")
+	restarted.SyncOnce(context.Background())
+
+	_, _, failed := obs.snapshot()
+	require.Empty(t, failed, "republishing the same index must reuse the manifest, not fail")
+	require.Equal(t, uint64(42), restarted.LastPublishedIndex(7))
+}
+
+// §4 bounds uploads to one at a time per process by default, so a process
+// hosting many groups cannot saturate its uplink.
+func TestSchedulerBoundsConcurrentUploads(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	var inFlight, peak atomic.Int64
+	groups := make([]OffloadGroup, 0, 4)
+	for i := range 4 {
+		groupID := uint64(i) + 1 //nolint:gosec // loop index over a 4-element fixture.
+		dir := seedSchedulerGroup(t, root, "g"+string(rune('a'+i)))
+		groups = append(groups, OffloadGroup{
+			GroupID: groupID,
+			DataDir: dir,
+			IsLeader: func() bool {
+				cur := inFlight.Add(1)
+				for {
+					old := peak.Load()
+					if cur <= old || peak.CompareAndSwap(old, cur) {
+						break
+					}
+				}
+				time.Sleep(time.Millisecond)
+				inFlight.Add(-1)
+				return true
+			},
+		})
+	}
+
+	s := NewScheduler(store, groups, "cluster-a", "cluster-a", "test")
+	s.SyncOnce(context.Background())
+
+	require.Equal(t, int64(1), peak.Load(), "default concurrency is one upload per process")
+}
+
+// Cancellation must stop the scan rather than being reported as a publish
+// failure: a cancelled context is a shutdown, not an object-store problem.
+func TestSchedulerTreatsCancellationAsShutdown(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "g")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	s := NewScheduler(store, []OffloadGroup{{
+		GroupID: 7, DataDir: dataDir, IsLeader: func() bool { return true },
+	}}, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	s.SyncOnce(ctx)
+
+	published, _, failed := obs.snapshot()
+	require.Empty(t, published)
+	require.Empty(t, failed, "cancellation is shutdown, not a publish failure")
+}
+
+func TestSchedulerRunRequiresStoreAndSourceCluster(t *testing.T) {
+	t.Parallel()
+
+	require.ErrorIs(t, NewScheduler(nil, nil, "p", "c", "v").Run(context.Background()), ErrInvalidOptions)
+	store := newTestLocalStore(t, t.TempDir())
+	require.ErrorIs(t, NewScheduler(store, nil, "p", "", "v").Run(context.Background()), ErrInvalidOptions)
+}

From d1001398432ef54feb08a47cfd6e7aee9015e685 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Mon, 7 Sep 2026 21:57:56 +0900
Subject: [PATCH 2/7] snapshotoffload: require leadership callbacks and skip
 republished indexes
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two P1s from review.

A nil IsLeader/VerifyLeader previously meant "publishable": validate()
checked neither, so a miswired scheduler would commit a manifest from a
follower — silently, and against the one guarantee the scheduler
exists to provide. NewScheduler now validates eagerly and returns an
error, because SyncOnce is exported and does not re-validate; an
invalid scheduler can no longer be constructed. publishGroup also
treats unknown leadership as "not leader" rather than as permission,
so a Scheduler built by some other route still fails closed.

The published high-water mark was written but never read, so an
unchanged snapshot was fully re-read, re-hashed and re-fsynced on every
tick. PublishOptions.SkipIfNotNewerThan now short-circuits after the
export metadata is read but BEFORE the payload is spooled, which is
where the cost is; the scheduler reports it as an "already_published"
skip rather than a publish.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 internal/snapshotoffload/manifest.go       |   6 ++
 internal/snapshotoffload/publish.go        |  15 +++
 internal/snapshotoffload/scheduler.go      |  57 +++++++++-
 internal/snapshotoffload/scheduler_test.go | 116 ++++++++++++++++++---
 4 files changed, 174 insertions(+), 20 deletions(-)

diff --git a/internal/snapshotoffload/manifest.go b/internal/snapshotoffload/manifest.go
index f750ce0ba..5b6264064 100644
--- a/internal/snapshotoffload/manifest.go
+++ b/internal/snapshotoffload/manifest.go
@@ -22,6 +22,12 @@ var (
 	ErrIntegrity      = errors.New("snapshot offload: integrity check failed")
 	ErrObjectConflict = errors.New("snapshot offload: object conflict")
 	ErrObjectNotFound = errors.New("snapshot offload: object not found")
+
+	// 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 {
diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go
index aef47f40c..a6d174b51 100644
--- a/internal/snapshotoffload/publish.go
+++ b/internal/snapshotoffload/publish.go
@@ -32,6 +32,17 @@ type PublishOptions struct {
 	// 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) {
@@ -45,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
diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
index e7494fcff..be811c108 100644
--- a/internal/snapshotoffload/scheduler.go
+++ b/internal/snapshotoffload/scheduler.go
@@ -126,7 +126,12 @@ func WithSchedulerSpoolDir(dir string) SchedulerOption {
 
 // NewScheduler builds the offload scheduler. It is opt-in: callers construct it
 // only when object offload is configured.
-func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluster, binaryVersion string, opts ...SchedulerOption) *Scheduler {
+// NewScheduler validates its configuration eagerly so an invalid
+// scheduler cannot be constructed at all. In particular every group
+// must supply both leadership callbacks: SyncOnce is exported and does
+// not re-validate, so a nil callback that survived construction would
+// be a follower publishing a manifest.
+func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluster, binaryVersion string, opts ...SchedulerOption) (*Scheduler, error) {
 	s := &Scheduler{
 		groups:      groups,
 		store:       store,
@@ -144,7 +149,10 @@ func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluste
 	for _, opt := range opts {
 		opt(s)
 	}
-	return s
+	if err := s.validate(); err != nil {
+		return nil, err
+	}
+	return s, nil
 }
 
 func (s *Scheduler) validate() error {
@@ -154,6 +162,21 @@ func (s *Scheduler) validate() error {
 	case s.sourceName == "":
 		return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires a source cluster name")
 	}
+	// Both leadership callbacks are mandatory. Treating a nil callback
+	// as "leader" would let a miswired scheduler publish from a
+	// follower, which is the one thing this scheduler exists to
+	// prevent — and it would do so silently. Fail at construction
+	// instead, where the operator sees it.
+	for _, group := range s.groups {
+		switch {
+		case group.IsLeader == nil:
+			return errors.Wrapf(ErrInvalidOptions,
+				"snapshot offload group %d requires an IsLeader callback", group.GroupID)
+		case group.VerifyLeader == nil:
+			return errors.Wrapf(ErrInvalidOptions,
+				"snapshot offload group %d requires a VerifyLeader callback", group.GroupID)
+		}
+	}
 	return nil
 }
 
@@ -213,8 +236,16 @@ func (s *Scheduler) SyncOnce(ctx context.Context) {
 }
 
 func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
-	// Cheap pre-check first: a follower must not even open the snapshot.
-	if group.IsLeader != nil && !group.IsLeader() {
+	// Cheap pre-check first: a follower must not even open the
+	// snapshot. NewScheduler rejects a nil callback, so this is
+	// defence in depth for a Scheduler built by some other route:
+	// unknown leadership is treated as "not leader", never as
+	// permission to publish.
+	if group.IsLeader == nil || group.VerifyLeader == nil {
+		s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "leadership_unknown")
+		return
+	}
+	if !group.IsLeader() {
 		s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "not_leader")
 		return
 	}
@@ -228,11 +259,19 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 		BinaryVersion: s.binVersion,
 		SpoolDir:      s.spoolDir,
 		VerifyLeader:  group.VerifyLeader,
+		// Suppress the whole spool when this node has already
+		// published this index. Without it an unchanged snapshot is
+		// fully re-read and re-hashed on every tick.
+		SkipIfNotNewerThan: s.publishedIndex(group.GroupID),
 	})
 	if err != nil {
 		if errors.Is(err, context.Canceled) || ctx.Err() != nil {
 			return
 		}
+		if errors.Is(err, ErrSnapshotNotNewer) {
+			s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "already_published")
+			return
+		}
 		s.observer.ObserveSnapshotOffloadFailed(group.GroupID, err)
 		s.logger.WarnContext(ctx, "snapshot offload publish failed",
 			slog.Uint64("group_id", group.GroupID), slog.String("error", err.Error()))
@@ -243,6 +282,16 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 		group.GroupID, manifest.SnapshotIndex, manifest.Payload.Bytes, s.now().Sub(started))
 }
 
+// publishedIndex returns this process's high-water mark for a group.
+// It is intentionally in-memory only: a restart re-publishes once,
+// which the object store's content addressing makes cheap and which
+// keeps the scheduler from needing durable state of its own.
+func (s *Scheduler) publishedIndex(groupID uint64) uint64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.published[groupID]
+}
+
 func (s *Scheduler) markPublished(groupID, index uint64) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
index 52e7efb93..715bbfe6a 100644
--- a/internal/snapshotoffload/scheduler_test.go
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -65,11 +65,12 @@ func TestSchedulerSkipsGroupsThisNodeDoesNotLead(t *testing.T) {
 	store := newTestLocalStore(t, filepath.Join(root, "objects"))
 	obs := &recordingObserver{}
 
-	s := NewScheduler(store, []OffloadGroup{{
-		GroupID:  7,
-		DataDir:  dataDir,
-		IsLeader: func() bool { return false },
-	}}, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	s := newTestScheduler(t, store, []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return false },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}, WithSchedulerObserver(obs))
 
 	s.SyncOnce(context.Background())
 
@@ -92,12 +93,12 @@ func TestSchedulerDoesNotCommitManifestWhenLeadershipIsLostWhileSpooling(t *test
 	obs := &recordingObserver{}
 	lost := errors.New("leadership lost")
 
-	s := NewScheduler(store, []OffloadGroup{{
+	s := newTestScheduler(t, store, []OffloadGroup{{
 		GroupID:      7,
 		DataDir:      dataDir,
 		IsLeader:     func() bool { return true },
 		VerifyLeader: func(context.Context) error { return lost },
-	}}, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	}}, WithSchedulerObserver(obs))
 
 	s.SyncOnce(context.Background())
 
@@ -123,16 +124,21 @@ func TestSchedulerPublishesOnceAndIsIdempotentAcrossRestart(t *testing.T) {
 	dataDir := seedSchedulerGroup(t, root, "g")
 	store := newTestLocalStore(t, filepath.Join(root, "objects"))
 	obs := &recordingObserver{}
-	groups := []OffloadGroup{{GroupID: 7, DataDir: dataDir, IsLeader: func() bool { return true }}}
+	groups := []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}
 
-	s := NewScheduler(store, groups, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	s := newTestScheduler(t, store, groups, WithSchedulerObserver(obs))
 	s.SyncOnce(context.Background())
 	require.Equal(t, uint64(42), s.LastPublishedIndex(7))
 
 	// Same process, second scan.
 	s.SyncOnce(context.Background())
 	// A different process that has published nothing itself.
-	restarted := NewScheduler(store, groups, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	restarted := newTestScheduler(t, store, groups, WithSchedulerObserver(obs))
 	require.Zero(t, restarted.LastPublishedIndex(7), "a fresh process starts with no local record")
 	restarted.SyncOnce(context.Background())
 
@@ -168,10 +174,11 @@ func TestSchedulerBoundsConcurrentUploads(t *testing.T) {
 				inFlight.Add(-1)
 				return true
 			},
+			VerifyLeader: func(context.Context) error { return nil },
 		})
 	}
 
-	s := NewScheduler(store, groups, "cluster-a", "cluster-a", "test")
+	s := newTestScheduler(t, store, groups)
 	s.SyncOnce(context.Background())
 
 	require.Equal(t, int64(1), peak.Load(), "default concurrency is one upload per process")
@@ -190,9 +197,12 @@ func TestSchedulerTreatsCancellationAsShutdown(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
 
-	s := NewScheduler(store, []OffloadGroup{{
-		GroupID: 7, DataDir: dataDir, IsLeader: func() bool { return true },
-	}}, "cluster-a", "cluster-a", "test", WithSchedulerObserver(obs))
+	s := newTestScheduler(t, store, []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}, WithSchedulerObserver(obs))
 	s.SyncOnce(ctx)
 
 	published, _, failed := obs.snapshot()
@@ -203,7 +213,81 @@ func TestSchedulerTreatsCancellationAsShutdown(t *testing.T) {
 func TestSchedulerRunRequiresStoreAndSourceCluster(t *testing.T) {
 	t.Parallel()
 
-	require.ErrorIs(t, NewScheduler(nil, nil, "p", "c", "v").Run(context.Background()), ErrInvalidOptions)
+	_, err := NewScheduler(nil, nil, "p", "c", "v")
+	require.ErrorIs(t, err, ErrInvalidOptions)
 	store := newTestLocalStore(t, t.TempDir())
-	require.ErrorIs(t, NewScheduler(store, nil, "p", "", "v").Run(context.Background()), ErrInvalidOptions)
+	_, err = NewScheduler(store, nil, "p", "", "v")
+	require.ErrorIs(t, err, ErrInvalidOptions)
+}
+
+// newTestScheduler builds a valid scheduler and fails the test if the
+// configuration is rejected.
+func newTestScheduler(
+	t *testing.T, store ObjectStore, groups []OffloadGroup, opts ...SchedulerOption,
+) *Scheduler {
+	t.Helper()
+	s, err := NewScheduler(store, groups, "cluster-a", "cluster-a", "test", opts...)
+	require.NoError(t, err)
+	return s
+}
+
+// TestSchedulerRejectsGroupsMissingALeadershipCallback is the P1 guard:
+// a nil callback previously meant "publishable", so a miswired
+// scheduler would publish a manifest from a follower — silently, and
+// exactly against the guarantee the scheduler exists to provide.
+func TestSchedulerRejectsGroupsMissingALeadershipCallback(t *testing.T) {
+	t.Parallel()
+
+	store := newTestLocalStore(t, t.TempDir())
+	valid := OffloadGroup{
+		GroupID:      7,
+		DataDir:      t.TempDir(),
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}
+
+	missingIsLeader := valid
+	missingIsLeader.IsLeader = nil
+	_, err := NewScheduler(store, []OffloadGroup{missingIsLeader}, "p", "c", "v")
+	require.ErrorIs(t, err, ErrInvalidOptions)
+	require.ErrorContains(t, err, "IsLeader")
+
+	missingVerify := valid
+	missingVerify.VerifyLeader = nil
+	_, err = NewScheduler(store, []OffloadGroup{missingVerify}, "p", "c", "v")
+	require.ErrorIs(t, err, ErrInvalidOptions)
+	require.ErrorContains(t, err, "VerifyLeader")
+
+	// The fully-wired group is accepted.
+	_, err = NewScheduler(store, []OffloadGroup{valid}, "p", "c", "v")
+	require.NoError(t, err)
+}
+
+// TestSchedulerSkipsRepublishingAnUnchangedSnapshot is the P1
+// efficiency guard: an unchanged snapshot must not be re-spooled and
+// re-hashed on every tick. The skip has to happen before the payload
+// is read, so it is observable as a skip rather than a publish.
+func TestSchedulerSkipsRepublishingAnUnchangedSnapshot(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "g")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+	groups := []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}
+
+	s := newTestScheduler(t, store, groups, WithSchedulerObserver(obs))
+	s.SyncOnce(context.Background())
+	s.SyncOnce(context.Background())
+	s.SyncOnce(context.Background())
+
+	published, skipped, failed := obs.snapshot()
+	require.Empty(t, failed)
+	require.Len(t, published, 1, "an unchanged snapshot must be published exactly once")
+	require.Equal(t, []string{"already_published", "already_published"}, skipped)
 }

From 40a4963c7d8c5969993f4c9c518c60ac6e1f4ec2 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Mon, 7 Sep 2026 22:01:17 +0900
Subject: [PATCH 3/7] snapshotoffload: address the scheduler P2 review findings
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Four P2s, all confirmed against the code.

The upload semaphore was allocated inside SyncOnce, so an
operator-forced scan overlapping the Run loop's pass got its own full
allowance — two uploads under a configured limit of one. It now lives
on the Scheduler.

An absent persisted snapshot surfaced as a publish failure. A young or
lightly-used group simply has not snapshotted yet, so that emitted a
warning and a failure metric every interval until Raft eventually
produced one. It is now a "no_persisted_snapshot" skip.

Jitter randomised only the delay between whole-process scans, so
groups in one process still started together. Scheduled scans now
stagger group starts across the jitter window. SyncOnce deliberately
does not stagger: it is the "scan now" entry point, and delaying an
explicit request by up to a jitter window would be worse than the
burst it avoids.

Manifest reuse compared BinaryVersion, so a process restarting on a
new binary conflicted with the manifest the previous binary had
committed for the same index, and every scan failed until Raft made a
new snapshot. BinaryVersion records which binary published the
artifact rather than anything about the snapshot, so reuse now ignores
it and the committed manifest keeps the original publisher's version
as the audit record.

The two jitter draws are factored into one helper so the weak-RNG
exemption is stated once rather than duplicated.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 internal/snapshotoffload/publish.go        |  13 +++
 internal/snapshotoffload/scheduler.go      |  69 ++++++++++--
 internal/snapshotoffload/scheduler_test.go | 118 +++++++++++++++++++++
 3 files changed, 194 insertions(+), 6 deletions(-)

diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go
index a6d174b51..fec40dd99 100644
--- a/internal/snapshotoffload/publish.go
+++ b/internal/snapshotoffload/publish.go
@@ -271,9 +271,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)
 }
 
diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
index be811c108..85b16b1ef 100644
--- a/internal/snapshotoffload/scheduler.go
+++ b/internal/snapshotoffload/scheduler.go
@@ -36,6 +36,11 @@ type Scheduler struct {
 	// store, not from this map.
 	mu        sync.Mutex
 	published map[uint64]uint64
+	// uploads bounds concurrent uploads across every scan on this
+	// scheduler, including an operator-forced SyncOnce that overlaps
+	// the Run loop's pass. Allocating it per scan would give each its
+	// own full allowance.
+	uploads chan struct{}
 }
 
 // OffloadGroup is one local Raft group the scheduler may publish for.
@@ -152,6 +157,12 @@ func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluste
 	if err := s.validate(); err != nil {
 		return nil, err
 	}
+	// One limiter for the scheduler, not one per scan: an
+	// operator-forced SyncOnce can overlap the pass running from Run,
+	// and a per-scan semaphore would grant each its own full
+	// allowance — two concurrent uploads under a configured limit of
+	// one.
+	s.uploads = make(chan struct{}, s.concurrency)
 	return s, nil
 }
 
@@ -198,23 +209,40 @@ func (s *Scheduler) Run(ctx context.Context) error {
 		case <-ctx.Done():
 			return nil
 		case <-timer.C:
-			s.SyncOnce(ctx)
+			s.scan(ctx, true)
 			timer.Reset(s.nextDelay())
 		}
 	}
 }
 
 func (s *Scheduler) nextDelay() time.Duration {
+	return s.interval + s.jitterSlice()
+}
+
+// jitterSlice returns a uniform duration in [0, jitter), or zero when
+// jitter is disabled. Both the inter-scan delay and the per-group
+// stagger draw from it, so the weak-RNG exemption is stated once:
+// this is load spreading, never a security decision.
+func (s *Scheduler) jitterSlice() time.Duration {
 	if s.jitter <= 0 {
-		return s.interval
+		return 0
 	}
-	return s.interval + time.Duration(rand.Int64N(int64(s.jitter))) //nolint:gosec // scheduling jitter, not a security decision.
+	return time.Duration(rand.Int64N(int64(s.jitter))) //nolint:gosec // scheduling jitter, not a security decision.
 }
 
 // SyncOnce runs one scan across every local group, bounded by the upload
 // concurrency limit. Exported so tests and operators can force a pass.
 func (s *Scheduler) SyncOnce(ctx context.Context) {
-	sem := make(chan struct{}, s.concurrency)
+	// No stagger: SyncOnce is the "scan now" entry point (operator
+	// action, tests), and delaying it by up to a jitter window would
+	// make an explicit request take minutes to start.
+	s.scan(ctx, false)
+}
+
+// scan runs one pass. stagger spreads group starts across the jitter
+// window so a multi-group process does not begin every upload on the
+// same tick; it is used only by the Run loop.
+func (s *Scheduler) scan(ctx context.Context, stagger bool) {
 	var wg sync.WaitGroup
 	for _, group := range s.groups {
 		if ctx.Err() != nil {
@@ -223,18 +251,38 @@ func (s *Scheduler) SyncOnce(ctx context.Context) {
 		wg.Add(1)
 		go func(g OffloadGroup) {
 			defer wg.Done()
+			if stagger && !s.sleepStagger(ctx) {
+				return
+			}
 			select {
-			case sem <- struct{}{}:
+			case s.uploads <- struct{}{}:
 			case <-ctx.Done():
 				return
 			}
-			defer func() { <-sem }()
+			defer func() { <-s.uploads }()
 			s.publishGroup(ctx, g)
 		}(group)
 	}
 	wg.Wait()
 }
 
+// sleepStagger waits a random slice of the jitter window. It reports
+// false when ctx ended first, so the caller abandons the group.
+func (s *Scheduler) sleepStagger(ctx context.Context) bool {
+	slice := s.jitterSlice()
+	if slice <= 0 {
+		return true
+	}
+	timer := time.NewTimer(slice)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
 func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 	// Cheap pre-check first: a follower must not even open the
 	// snapshot. NewScheduler rejects a nil callback, so this is
@@ -272,6 +320,15 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 			s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "already_published")
 			return
 		}
+		if errors.Is(err, ErrObjectNotFound) {
+			// A young or lightly-used group has not persisted its
+			// first snapshot yet. That is a normal scan outcome, not
+			// an outage: reporting it as a failure would emit a
+			// warning and a failure metric every interval until Raft
+			// eventually snapshots.
+			s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "no_persisted_snapshot")
+			return
+		}
 		s.observer.ObserveSnapshotOffloadFailed(group.GroupID, err)
 		s.logger.WarnContext(ctx, "snapshot offload publish failed",
 			slog.Uint64("group_id", group.GroupID), slog.String("error", err.Error()))
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
index 715bbfe6a..e4fce6464 100644
--- a/internal/snapshotoffload/scheduler_test.go
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -291,3 +291,121 @@ func TestSchedulerSkipsRepublishingAnUnchangedSnapshot(t *testing.T) {
 	require.Len(t, published, 1, "an unchanged snapshot must be published exactly once")
 	require.Equal(t, []string{"already_published", "already_published"}, skipped)
 }
+
+// TestSchedulerSharesTheUploadLimitAcrossConcurrentScans pins that the
+// limiter belongs to the scheduler, not to one scan. An operator-forced
+// SyncOnce can overlap the Run loop's pass, and a per-scan semaphore
+// would hand each its own full allowance — two uploads under a
+// configured limit of one.
+func TestSchedulerSharesTheUploadLimitAcrossConcurrentScans(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	var inFlight, peak atomic.Int64
+
+	groups := make([]OffloadGroup, 0, 4)
+	for i := range 4 {
+		groupID := uint64(i) + 1 //nolint:gosec // loop index over a 4-element fixture.
+		dir := seedSchedulerGroup(t, root, "shared"+string(rune('a'+i)))
+		groups = append(groups, OffloadGroup{
+			GroupID: groupID,
+			DataDir: dir,
+			IsLeader: func() bool {
+				cur := inFlight.Add(1)
+				for {
+					old := peak.Load()
+					if cur <= old || peak.CompareAndSwap(old, cur) {
+						break
+					}
+				}
+				time.Sleep(2 * time.Millisecond)
+				inFlight.Add(-1)
+				return true
+			},
+			VerifyLeader: func(context.Context) error { return nil },
+		})
+	}
+
+	s := newTestScheduler(t, store, groups)
+
+	// Two overlapping scans on the same scheduler.
+	var wg sync.WaitGroup
+	for range 2 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.SyncOnce(context.Background())
+		}()
+	}
+	wg.Wait()
+
+	require.Equal(t, int64(1), peak.Load(),
+		"overlapping scans must share the configured upload limit")
+}
+
+// TestSchedulerTreatsAbsentPersistedSnapshotAsASkip covers a young or
+// lightly-used group: Raft has not produced a snapshot yet, which is a
+// normal scan outcome. Reporting it as a failure would emit a warning
+// and a failure metric every interval until Raft eventually snapshots.
+func TestSchedulerTreatsAbsentPersistedSnapshotAsASkip(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+
+	// A data dir with no persisted snapshot at all.
+	empty := filepath.Join(root, "empty-group")
+	require.NoError(t, os.MkdirAll(empty, 0o755))
+
+	s := newTestScheduler(t, store, []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      empty,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}, WithSchedulerObserver(obs))
+	s.SyncOnce(context.Background())
+
+	published, skipped, failed := obs.snapshot()
+	require.Empty(t, published)
+	require.Empty(t, failed, "a group with no snapshot yet is not an outage")
+	require.Equal(t, []string{"no_persisted_snapshot"}, skipped)
+}
+
+// TestPublishReusesACommittedManifestAcrossABinaryUpgrade pins the
+// upgrade retry path: a process that restarts on a new binary and
+// republishes an index it has not published locally must reuse the
+// committed manifest instead of conflicting with it. Otherwise every
+// scan fails until Raft happens to produce a new snapshot.
+func TestPublishReusesACommittedManifestAcrossABinaryUpgrade(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "upgrade")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+
+	first, err := PublishPersistedSnapshot(context.Background(), PublishOptions{
+		Store:         store,
+		DataDir:       dataDir,
+		Prefix:        "cluster-a",
+		GroupID:       7,
+		SourceCluster: "cluster-a",
+		BinaryVersion: "v1.0.0",
+	})
+	require.NoError(t, err)
+
+	// Same snapshot, newer binary.
+	second, err := PublishPersistedSnapshot(context.Background(), PublishOptions{
+		Store:         store,
+		DataDir:       dataDir,
+		Prefix:        "cluster-a",
+		GroupID:       7,
+		SourceCluster: "cluster-a",
+		BinaryVersion: "v2.0.0",
+	})
+	require.NoError(t, err, "an upgraded binary must reuse the committed manifest")
+	require.Equal(t, first.SnapshotIndex, second.SnapshotIndex)
+	require.Equal(t, "v1.0.0", second.BinaryVersion,
+		"the committed manifest keeps the publishing binary's version as the audit record")
+}

From 410fa93b0e39be198ef6b86994eee2b9e3b3b252 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Mon, 7 Sep 2026 22:13:59 +0900
Subject: [PATCH 4/7] snapshotoffload: address the second round of scheduler
 P2s
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Four findings, all confirmed.

The "no snapshot yet" skip matched ErrObjectNotFound, which also covers
an object disappearing from the store mid-publish — a real outage that
would have been silently reported as a routine skip. openPublishExport
now returns its own ErrNoPersistedSnapshot and the scheduler matches
only that.

A group with an empty or whitespace-only DataDir was accepted at
construction and then failed validatePublishOptions on every attempt,
turning a static misconfiguration into a recurring failure metric
instead of a startup error. A whitespace-only source cluster passed the
bare != "" check but buildManifest trimmed it away, so artifacts would
publish without the source-cluster identity the scheduler requires.
Both are now rejected at construction.

The upload semaphore bounds aggregate work, not work per group: with
concurrency above one, two overlapping scans could each take a slot for
the SAME group, read the same high-water mark before either recorded a
publish, and both spool and upload the same snapshot. Groups are now
single-flighted.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 internal/snapshotoffload/manifest.go       |   8 ++
 internal/snapshotoffload/publish.go        |   2 +-
 internal/snapshotoffload/scheduler.go      |  50 ++++++++-
 internal/snapshotoffload/scheduler_test.go | 119 +++++++++++++++++++++
 4 files changed, 177 insertions(+), 2 deletions(-)

diff --git a/internal/snapshotoffload/manifest.go b/internal/snapshotoffload/manifest.go
index 5b6264064..ca30a7686 100644
--- a/internal/snapshotoffload/manifest.go
+++ b/internal/snapshotoffload/manifest.go
@@ -23,6 +23,14 @@ var (
 	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
diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go
index fec40dd99..65bf614f3 100644
--- a/internal/snapshotoffload/publish.go
+++ b/internal/snapshotoffload/publish.go
@@ -115,7 +115,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
 }
diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
index 85b16b1ef..aaceef96f 100644
--- a/internal/snapshotoffload/scheduler.go
+++ b/internal/snapshotoffload/scheduler.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"log/slog"
 	"math/rand/v2"
+	"strings"
 	"sync"
 	"time"
 
@@ -41,6 +42,14 @@ type Scheduler struct {
 	// the Run loop's pass. Allocating it per scan would give each its
 	// own full allowance.
 	uploads chan struct{}
+	// inFlight holds the groups currently being published. The
+	// semaphore bounds AGGREGATE work, not work per group: with
+	// concurrency above one, two overlapping scans can each take a
+	// slot for the SAME group, read the same high-water mark before
+	// either records a publish, and both spool and upload the same
+	// multi-gigabyte snapshot. Single-flighting per group is what
+	// makes a group's publish idempotent under overlap.
+	inFlight map[uint64]struct{}
 }
 
 // OffloadGroup is one local Raft group the scheduler may publish for.
@@ -150,6 +159,7 @@ func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluste
 		logger:      slog.Default().With(slog.String("component", "snapshot-offload")),
 		now:         time.Now,
 		published:   make(map[uint64]uint64),
+		inFlight:    make(map[uint64]struct{}),
 	}
 	for _, opt := range opts {
 		opt(s)
@@ -167,6 +177,11 @@ func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluste
 }
 
 func (s *Scheduler) validate() error {
+	// Trim before checking: a whitespace-only name passes a bare !=
+	// "" test but buildManifest trims it to empty, so the scheduler
+	// would publish artifacts without the source-cluster identity it
+	// requires.
+	s.sourceName = strings.TrimSpace(s.sourceName)
 	switch {
 	case s.store == nil:
 		return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires an object store")
@@ -180,6 +195,12 @@ func (s *Scheduler) validate() error {
 	// instead, where the operator sees it.
 	for _, group := range s.groups {
 		switch {
+		case strings.TrimSpace(group.DataDir) == "":
+			// Otherwise every publish fails validatePublishOptions at
+			// runtime, turning a static misconfiguration into a
+			// recurring failure metric instead of a startup error.
+			return errors.Wrapf(ErrInvalidOptions,
+				"snapshot offload group %d requires a data dir", group.GroupID)
 		case group.IsLeader == nil:
 			return errors.Wrapf(ErrInvalidOptions,
 				"snapshot offload group %d requires an IsLeader callback", group.GroupID)
@@ -260,6 +281,11 @@ func (s *Scheduler) scan(ctx context.Context, stagger bool) {
 				return
 			}
 			defer func() { <-s.uploads }()
+			if !s.beginGroup(g.GroupID) {
+				s.observer.ObserveSnapshotOffloadSkipped(g.GroupID, "already_in_flight")
+				return
+			}
+			defer s.endGroup(g.GroupID)
 			s.publishGroup(ctx, g)
 		}(group)
 	}
@@ -320,12 +346,16 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 			s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "already_published")
 			return
 		}
-		if errors.Is(err, ErrObjectNotFound) {
+		if errors.Is(err, ErrNoPersistedSnapshot) {
 			// A young or lightly-used group has not persisted its
 			// first snapshot yet. That is a normal scan outcome, not
 			// an outage: reporting it as a failure would emit a
 			// warning and a failure metric every interval until Raft
 			// eventually snapshots.
+			//
+			// Matched on its own sentinel, NOT on ErrObjectNotFound:
+			// an object disappearing from the store mid-publish is a
+			// genuine failure and must stay one.
 			s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "no_persisted_snapshot")
 			return
 		}
@@ -339,6 +369,24 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 		group.GroupID, manifest.SnapshotIndex, manifest.Payload.Bytes, s.now().Sub(started))
 }
 
+// beginGroup claims a group for publishing, reporting false when
+// another scan already holds it.
+func (s *Scheduler) beginGroup(groupID uint64) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, busy := s.inFlight[groupID]; busy {
+		return false
+	}
+	s.inFlight[groupID] = struct{}{}
+	return true
+}
+
+func (s *Scheduler) endGroup(groupID uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.inFlight, groupID)
+}
+
 // publishedIndex returns this process's high-water mark for a group.
 // It is intentionally in-memory only: a restart re-publishes once,
 // which the object store's content addressing makes cheap and which
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
index e4fce6464..9a1bed7a8 100644
--- a/internal/snapshotoffload/scheduler_test.go
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -2,6 +2,7 @@ package snapshotoffload
 
 import (
 	"context"
+	"io"
 	"os"
 	"path/filepath"
 	"sync"
@@ -409,3 +410,121 @@ func TestPublishReusesACommittedManifestAcrossABinaryUpgrade(t *testing.T) {
 	require.Equal(t, "v1.0.0", second.BinaryVersion,
 		"the committed manifest keeps the publishing binary's version as the audit record")
 }
+
+// TestSchedulerRejectsInvalidGroupAndClusterConfiguration keeps static
+// misconfiguration a startup error instead of a recurring per-interval
+// failure metric.
+func TestSchedulerRejectsInvalidGroupAndClusterConfiguration(t *testing.T) {
+	t.Parallel()
+
+	store := newTestLocalStore(t, t.TempDir())
+	valid := OffloadGroup{
+		GroupID:      7,
+		DataDir:      t.TempDir(),
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}
+
+	noDataDir := valid
+	noDataDir.DataDir = "   "
+	_, err := NewScheduler(store, []OffloadGroup{noDataDir}, "p", "cluster-a", "v")
+	require.ErrorIs(t, err, ErrInvalidOptions)
+	require.ErrorContains(t, err, "data dir")
+
+	// A whitespace-only cluster name passes a bare != "" test but
+	// buildManifest trims it away, so artifacts would be published
+	// without the source-cluster identity the scheduler requires.
+	_, err = NewScheduler(store, []OffloadGroup{valid}, "p", "   ", "v")
+	require.ErrorIs(t, err, ErrInvalidOptions)
+	require.ErrorContains(t, err, "source cluster")
+}
+
+// TestSchedulerReportsRemoteObjectLossAsAFailure separates the two
+// not-found cases. A group with no local snapshot is a skip; an object
+// disappearing from the store mid-publish is a real failure, and
+// collapsing them would silence the second.
+func TestSchedulerReportsRemoteObjectLossAsAFailure(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "remoteloss")
+	obs := &recordingObserver{}
+	store := &objectLosingStore{ObjectStore: newTestLocalStore(t, filepath.Join(root, "objects"))}
+
+	s := newTestScheduler(t, store, []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}, WithSchedulerObserver(obs))
+	s.SyncOnce(context.Background())
+
+	published, skipped, failed := obs.snapshot()
+	require.Empty(t, published)
+	require.NotEmpty(t, failed, "a vanished remote object is an outage, not a quiet skip")
+	require.NotContains(t, skipped, "no_persisted_snapshot")
+}
+
+// objectLosingStore makes every object read report not-found, standing
+// in for an object deleted between the head and the get.
+type objectLosingStore struct {
+	ObjectStore
+}
+
+func (s *objectLosingStore) PutObject(
+	ctx context.Context, key string, body io.Reader, opts PutOptions,
+) (ObjectInfo, error) {
+	return ObjectInfo{}, errors.Wrapf(ErrObjectNotFound, "object %s vanished", key)
+}
+
+// TestSchedulerSingleFlightsAGroupAcrossOverlappingScans pins that the
+// aggregate upload limiter is not enough: with concurrency above one,
+// two overlapping scans could each take a slot for the SAME group,
+// read the same high-water mark, and both spool and upload the same
+// snapshot.
+func TestSchedulerSingleFlightsAGroupAcrossOverlappingScans(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "singleflight")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+
+	var concurrentEntries, peak atomic.Int64
+	groups := []OffloadGroup{{
+		GroupID: 7,
+		DataDir: dataDir,
+		IsLeader: func() bool {
+			cur := concurrentEntries.Add(1)
+			for {
+				old := peak.Load()
+				if cur <= old || peak.CompareAndSwap(old, cur) {
+					break
+				}
+			}
+			time.Sleep(5 * time.Millisecond)
+			concurrentEntries.Add(-1)
+			return true
+		},
+		VerifyLeader: func(context.Context) error { return nil },
+	}}
+
+	s := newTestScheduler(t, store, groups,
+		WithSchedulerObserver(obs), WithSchedulerConcurrency(4))
+
+	var wg sync.WaitGroup
+	for range 4 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.SyncOnce(context.Background())
+		}()
+	}
+	wg.Wait()
+
+	require.Equal(t, int64(1), peak.Load(),
+		"one group must never be published by two scans at once")
+	published, _, failed := obs.snapshot()
+	require.Empty(t, failed)
+	require.Len(t, published, 1, "the same snapshot must be uploaded once, not once per scan")
+}

From 05a34e554a584e04f1bf815ac829d6872f75b8c3 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Tue, 8 Sep 2026 16:00:21 +0900
Subject: [PATCH 5/7] snapshotoffload: bound the leadership recheck and keep
 validate pure
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three P2s, all confirmed.

The pre-commit leadership recheck ran under whatever context the caller
supplied. The callback contract does not require callers to wrap their
engine method, and a raw etcd Engine.VerifyLeader issues a ReadIndex
that waits out its context during quorum loss — handed the long-lived
Run context, a scan would block until shutdown and no later snapshot
would ever be scheduled. Each attempt now gets its own 5s deadline,
matching the coordinator's own ReadIndex wrappers.

validate() trimmed sourceName in place, but Run calls validate too, so
an operator SyncOnce launched right after Run had publishGroup reading
sourceName while validate wrote it. The trim now happens once in
NewScheduler, before the scheduler escapes, and validate is pure.

Splitting ErrNoPersistedSnapshot out of ErrObjectNotFound silently
moved the standalone publish command's missing-snapshot exit code from
2 (data error) to 1 (invocation error). The sentinel is now classified
as a data error, restoring the CLI contract automation depends on.

Two of these initially had tests that did not pin them: asserting the
post-construction sourceName proves nothing because it is trimmed
either way, and there was no exit-code test at all. Both now fail when
their fix is reverted.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 cmd/elastickv-snapshot-offload/main.go      |   5 +
 cmd/elastickv-snapshot-offload/main_test.go |  46 +++++++++
 internal/snapshotoffload/scheduler.go       |  38 +++++--
 internal/snapshotoffload/scheduler_test.go  | 106 ++++++++++++++++++++
 4 files changed, 189 insertions(+), 6 deletions(-)

diff --git a/cmd/elastickv-snapshot-offload/main.go b/cmd/elastickv-snapshot-offload/main.go
index 78c4f9fbf..efdd1640d 100644
--- a/cmd/elastickv-snapshot-offload/main.go
+++ b/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
diff --git a/cmd/elastickv-snapshot-offload/main_test.go b/cmd/elastickv-snapshot-offload/main_test.go
index 008e2f5f9..a9dedf22b 100644
--- a/cmd/elastickv-snapshot-offload/main_test.go
+++ b/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))
+		})
+	}
+}
diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
index aaceef96f..3db245099 100644
--- a/internal/snapshotoffload/scheduler.go
+++ b/internal/snapshotoffload/scheduler.go
@@ -164,6 +164,11 @@ func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluste
 	for _, opt := range opts {
 		opt(s)
 	}
+	// Trim once here, before the scheduler escapes: a whitespace-only
+	// name passes a bare != "" test but buildManifest trims it to
+	// empty, so the scheduler would publish artifacts without the
+	// source-cluster identity it requires.
+	s.sourceName = strings.TrimSpace(s.sourceName)
 	if err := s.validate(); err != nil {
 		return nil, err
 	}
@@ -177,11 +182,10 @@ func NewScheduler(store ObjectStore, groups []OffloadGroup, prefix, sourceCluste
 }
 
 func (s *Scheduler) validate() error {
-	// Trim before checking: a whitespace-only name passes a bare !=
-	// "" test but buildManifest trims it to empty, so the scheduler
-	// would publish artifacts without the source-cluster identity it
-	// requires.
-	s.sourceName = strings.TrimSpace(s.sourceName)
+	// validate is pure: Run calls it too, and mutating shared
+	// configuration there would race a concurrent operator SyncOnce
+	// reading sourceName to build PublishOptions. The trim happens
+	// once in NewScheduler, before the scheduler is published.
 	switch {
 	case s.store == nil:
 		return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires an object store")
@@ -332,7 +336,7 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 		SourceCluster: s.sourceName,
 		BinaryVersion: s.binVersion,
 		SpoolDir:      s.spoolDir,
-		VerifyLeader:  group.VerifyLeader,
+		VerifyLeader:  s.boundedVerifyLeader(group.VerifyLeader),
 		// Suppress the whole spool when this node has already
 		// published this index. Without it an unchanged snapshot is
 		// fully re-read and re-hashed on every tick.
@@ -369,6 +373,28 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) {
 		group.GroupID, manifest.SnapshotIndex, manifest.Payload.Bytes, s.now().Sub(started))
 }
 
+// verifyLeaderTimeout bounds one pre-commit leadership recheck. It
+// matches the deadline the coordinator's own ReadIndex wrappers use.
+const verifyLeaderTimeout = 5 * time.Second
+
+// boundedVerifyLeader gives each leadership recheck its own deadline.
+//
+// The callback contract does not require callers to wrap their engine
+// method, and a raw etcd Engine.VerifyLeader issues a ReadIndex that,
+// during quorum loss, waits until its context expires. Handed the
+// long-lived Run context that expires only at shutdown, a scan would
+// block forever and no later snapshot would ever be scheduled.
+func (s *Scheduler) boundedVerifyLeader(verify func(context.Context) error) func(context.Context) error {
+	if verify == nil {
+		return nil
+	}
+	return func(ctx context.Context) error {
+		bounded, cancel := context.WithTimeout(ctx, verifyLeaderTimeout)
+		defer cancel()
+		return verify(bounded)
+	}
+}
+
 // beginGroup claims a group for publishing, reporting false when
 // another scan already holds it.
 func (s *Scheduler) beginGroup(groupID uint64) bool {
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
index 9a1bed7a8..87ae3a271 100644
--- a/internal/snapshotoffload/scheduler_test.go
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -528,3 +528,109 @@ func TestSchedulerSingleFlightsAGroupAcrossOverlappingScans(t *testing.T) {
 	require.Empty(t, failed)
 	require.Len(t, published, 1, "the same snapshot must be uploaded once, not once per scan")
 }
+
+// TestSchedulerBoundsTheLeadershipRecheck pins that the pre-commit
+// leadership recheck gets its own deadline.
+//
+// The callback contract does not require callers to wrap their engine
+// method, and a raw etcd Engine.VerifyLeader issues a ReadIndex that
+// waits out its context during quorum loss. Handed the long-lived Run
+// context — which expires only at shutdown — a scan would block
+// forever and no later snapshot would ever be scheduled.
+func TestSchedulerBoundsTheLeadershipRecheck(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "bounded")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	obs := &recordingObserver{}
+
+	gotDeadline := make(chan bool, 1)
+	s := newTestScheduler(t, store, []OffloadGroup{{
+		GroupID:  7,
+		DataDir:  dataDir,
+		IsLeader: func() bool { return true },
+		VerifyLeader: func(ctx context.Context) error {
+			_, ok := ctx.Deadline()
+			select {
+			case gotDeadline <- ok:
+			default:
+			}
+			return nil
+		},
+	}}, WithSchedulerObserver(obs))
+
+	// A context with no deadline of its own, like the Run context.
+	s.SyncOnce(context.Background())
+
+	select {
+	case ok := <-gotDeadline:
+		require.True(t, ok,
+			"the leadership recheck must run under its own deadline, not the caller's open-ended context")
+	default:
+		t.Fatal("VerifyLeader was never invoked")
+	}
+}
+
+// TestSchedulerValidateDoesNotMutateSharedConfiguration guards the
+// data race: Run calls validate too, and an operator SyncOnce launched
+// right after Run reads sourceName to build PublishOptions.
+func TestSchedulerValidateDoesNotMutateSharedConfiguration(t *testing.T) {
+	t.Parallel()
+
+	store := newTestLocalStore(t, t.TempDir())
+	s, err := NewScheduler(store, nil, "p", "  cluster-a  ", "v")
+	require.NoError(t, err)
+	require.Equal(t, "cluster-a", s.sourceName, "the trim must happen once, at construction")
+
+	// Plant an untrimmed value and re-validate. Asserting that the
+	// post-construction value is already trimmed proves nothing —
+	// it is trimmed either way. What must hold is that validate,
+	// which Run also calls while a concurrent SyncOnce reads
+	// sourceName, performs no write at all.
+	s.sourceName = "  padded  "
+	require.NoError(t, s.validate())
+	require.Equal(t, "  padded  ", s.sourceName,
+		"validate must not write shared configuration; Run calls it while SyncOnce reads")
+}
+
+// TestSchedulerRunAndSyncOnceAreRaceFree is the guard for mutating
+// shared configuration during validation. Run calls validate too, and
+// launching SyncOnce right after Run — the natural way to avoid
+// waiting out the first interval — has publishGroup reading
+// sourceName while validate would be writing it.
+//
+// Run validates once at startup, so the overlap window is narrow and
+// this test is a smoke check rather than a deterministic reproduction;
+// TestSchedulerValidateDoesNotMutateSharedConfiguration pins the
+// property itself.
+func TestSchedulerRunAndSyncOnceAreRaceFree(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	dataDir := seedSchedulerGroup(t, root, "racefree")
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+
+	s := newTestScheduler(t, store, []OffloadGroup{{
+		GroupID:      7,
+		DataDir:      dataDir,
+		IsLeader:     func() bool { return true },
+		VerifyLeader: func(context.Context) error { return nil },
+	}}, WithSchedulerInterval(time.Millisecond), WithSchedulerJitter(0))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_ = s.Run(ctx)
+	}()
+	// Overlap an operator-forced scan with the Run loop's validation.
+	for range 20 {
+		s.SyncOnce(ctx)
+	}
+	cancel()
+	wg.Wait()
+}

From e1c52ae82daa3864d0a5f290e6ae2a0f96bc1e93 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Tue, 8 Sep 2026 16:39:34 +0900
Subject: [PATCH 6/7] snapshotoffload: tighten the commit fence and bound scan
 goroutines
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two P1s.

The §4 leadership recheck ran before putManifest, which then spends an
unbounded remote read in verifyExistingManifest before creating the
object. A node demoted during that read still committed a manifest —
the one thing the scheduler exists to prevent, and the 5s deadline on
the verifier did not help because it bounds only the verifier. The
recheck now sits immediately before createManifestObject, after the
absence probe.

Each scan spawned one goroutine per group before the upload semaphore
applied, and a staggered scan armed a timer inside each. A process
hosting many groups would burst O(group-count) stacks and timers every
interval even at the default concurrency of one. Groups now dispatch
through a bounded worker pool. The semaphore stays: the pool bounds one
scan's goroutines, the semaphore bounds uploads across concurrent scans.

The goroutine test initially proved nothing — it sampled
runtime.NumGoroutine() after SyncOnce returned, by which point the
surplus goroutines parked on the semaphore had all exited. It now
samples during the scan and fails when the pool is removed.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 internal/snapshotoffload/offload_test.go   | 64 +++++++++++++++++++-
 internal/snapshotoffload/publish.go        | 26 +++++---
 internal/snapshotoffload/scheduler.go      | 69 ++++++++++++++++------
 internal/snapshotoffload/scheduler_test.go | 62 +++++++++++++++++++
 4 files changed, 194 insertions(+), 27 deletions(-)

diff --git a/internal/snapshotoffload/offload_test.go b/internal/snapshotoffload/offload_test.go
index 24a114fce..a8942cadd 100644
--- a/internal/snapshotoffload/offload_test.go
+++ b/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")
+}
diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go
index 65bf614f3..d53914c0f 100644
--- a/internal/snapshotoffload/publish.go
+++ b/internal/snapshotoffload/publish.go
@@ -98,12 +98,7 @@ func commitManifest(
 	if err := validateManifest(*manifest); err != nil {
 		return nil, err
 	}
-	if opts.VerifyLeader != nil {
-		if err := opts.VerifyLeader(ctx); err != nil {
-			return nil, errors.Wrap(err, "snapshot offload: leadership lost before manifest commit")
-		}
-	}
-	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
@@ -153,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
@@ -165,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
 	}
diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
index 3db245099..82393d872 100644
--- a/internal/snapshotoffload/scheduler.go
+++ b/internal/snapshotoffload/scheduler.go
@@ -268,34 +268,65 @@ func (s *Scheduler) SyncOnce(ctx context.Context) {
 // window so a multi-group process does not begin every upload on the
 // same tick; it is used only by the Run loop.
 func (s *Scheduler) scan(ctx context.Context, stagger bool) {
+	// A bounded worker pool, not one goroutine per group. A process
+	// hosting many groups would otherwise stack an O(group-count)
+	// burst of goroutines — and, on a staggered scan, one timer each —
+	// every interval, before the upload semaphore ever applies.
+	work := make(chan OffloadGroup)
+	workers := min(s.concurrency, len(s.groups))
+
 	var wg sync.WaitGroup
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for g := range work {
+				if ctx.Err() != nil {
+					return
+				}
+				if stagger && !s.sleepStagger(ctx) {
+					return
+				}
+				s.publishGroupBounded(ctx, g)
+			}
+		}()
+	}
+
 	for _, group := range s.groups {
 		if ctx.Err() != nil {
 			break
 		}
-		wg.Add(1)
-		go func(g OffloadGroup) {
-			defer wg.Done()
-			if stagger && !s.sleepStagger(ctx) {
-				return
-			}
-			select {
-			case s.uploads <- struct{}{}:
-			case <-ctx.Done():
-				return
-			}
-			defer func() { <-s.uploads }()
-			if !s.beginGroup(g.GroupID) {
-				s.observer.ObserveSnapshotOffloadSkipped(g.GroupID, "already_in_flight")
-				return
-			}
-			defer s.endGroup(g.GroupID)
-			s.publishGroup(ctx, g)
-		}(group)
+		select {
+		case work <- group:
+		case <-ctx.Done():
+		}
 	}
+	close(work)
 	wg.Wait()
 }
 
+// publishGroupBounded takes the process-wide upload slot and the
+// per-group single-flight claim, then publishes.
+//
+// The semaphore is still needed alongside the worker pool: the pool
+// bounds one scan's goroutines, while the semaphore bounds uploads
+// across concurrent scans (an operator SyncOnce overlapping Run).
+func (s *Scheduler) publishGroupBounded(ctx context.Context, group OffloadGroup) {
+	select {
+	case s.uploads <- struct{}{}:
+	case <-ctx.Done():
+		return
+	}
+	defer func() { <-s.uploads }()
+
+	if !s.beginGroup(group.GroupID) {
+		s.observer.ObserveSnapshotOffloadSkipped(group.GroupID, "already_in_flight")
+		return
+	}
+	defer s.endGroup(group.GroupID)
+	s.publishGroup(ctx, group)
+}
+
 // sleepStagger waits a random slice of the jitter window. It reports
 // false when ctx ended first, so the caller abandons the group.
 func (s *Scheduler) sleepStagger(ctx context.Context) bool {
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
index 87ae3a271..703840bf3 100644
--- a/internal/snapshotoffload/scheduler_test.go
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -2,9 +2,11 @@ package snapshotoffload
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -634,3 +636,63 @@ func TestSchedulerRunAndSyncOnceAreRaceFree(t *testing.T) {
 	cancel()
 	wg.Wait()
 }
+
+// TestSchedulerBoundsScanGoroutines pins that a scan does not stack one
+// goroutine per group. A process hosting many groups would otherwise
+// burst O(group-count) stacks — and, on a staggered scan, one timer
+// each — every interval, before the upload semaphore ever applies.
+func TestSchedulerBoundsScanGoroutines(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+
+	const groupCount = 40
+	var concurrent, peak, peakGoroutines atomic.Int64
+	groups := make([]OffloadGroup, 0, groupCount)
+	for i := range groupCount {
+		groupID := uint64(i) + 1 //nolint:gosec // loop index over a fixed-size fixture.
+		dir := seedSchedulerGroup(t, root, fmt.Sprintf("bounded-%d", i))
+		groups = append(groups, OffloadGroup{
+			GroupID: groupID,
+			DataDir: dir,
+			IsLeader: func() bool {
+				cur := concurrent.Add(1)
+				for {
+					old := peak.Load()
+					if cur <= old || peak.CompareAndSwap(old, cur) {
+						break
+					}
+				}
+				// Sample goroutines DURING the scan. With one
+				// goroutine per group the surplus sit parked on the
+				// upload semaphore and are invisible once SyncOnce
+				// has returned.
+				live := int64(runtime.NumGoroutine())
+				for {
+					old := peakGoroutines.Load()
+					if live <= old || peakGoroutines.CompareAndSwap(old, live) {
+						break
+					}
+				}
+				time.Sleep(time.Millisecond)
+				concurrent.Add(-1)
+				return true
+			},
+			VerifyLeader: func(context.Context) error { return nil },
+		})
+	}
+
+	const concurrency = 3
+	s := newTestScheduler(t, store, groups, WithSchedulerConcurrency(concurrency))
+
+	before := int64(runtime.NumGoroutine())
+	s.SyncOnce(context.Background())
+
+	require.LessOrEqual(t, peak.Load(), int64(concurrency),
+		"in-flight group work must stay within the configured concurrency")
+	// A per-group goroutine scan would park groupCount-concurrency
+	// goroutines on the semaphore; a pool adds only `workers`.
+	require.Less(t, peakGoroutines.Load(), before+int64(groupCount)/2,
+		"a scan must not stack one goroutine per group")
+}

From 6a3853eb4925c2caa829e140772a37a91f71594e Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Wed, 9 Sep 2026 13:14:35 +0900
Subject: [PATCH 7/7] snapshotoffload: keep every staggered start inside one
 jitter window
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The bounded worker pool moved the stagger sleep inside the worker loop,
so each worker slept a fresh jitter slice before every group it took
and the delays accumulated. With the default single worker and a 3m45s
jitter, 100 groups would push the last upload hours out — and Run does
not arm the next interval until the scan returns, so later groups could
go unvisited indefinitely. That was a regression introduced by the pool,
not a pre-existing bug.

Group start times are now absolute offsets from one scan start, carried
alongside the group. A worker already past a group's start time
proceeds immediately instead of sleeping again, so the whole scan stays
within a single jitter window regardless of group count.

The test measures a staggered scan against an unstaggered baseline of
the same fixture rather than against a fixed bound: the first version
compared to groupCount*jitter/2, which is exactly the accumulating
mean, and passed with the fix reverted.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 internal/snapshotoffload/scheduler.go      | 46 +++++++++++++++----
 internal/snapshotoffload/scheduler_test.go | 53 ++++++++++++++++++++++
 2 files changed, 89 insertions(+), 10 deletions(-)

diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go
index 82393d872..2db05eee1 100644
--- a/internal/snapshotoffload/scheduler.go
+++ b/internal/snapshotoffload/scheduler.go
@@ -272,7 +272,7 @@ func (s *Scheduler) scan(ctx context.Context, stagger bool) {
 	// hosting many groups would otherwise stack an O(group-count)
 	// burst of goroutines — and, on a staggered scan, one timer each —
 	// every interval, before the upload semaphore ever applies.
-	work := make(chan OffloadGroup)
+	work := make(chan staggeredGroup)
 	workers := min(s.concurrency, len(s.groups))
 
 	var wg sync.WaitGroup
@@ -284,20 +284,31 @@ func (s *Scheduler) scan(ctx context.Context, stagger bool) {
 				if ctx.Err() != nil {
 					return
 				}
-				if stagger && !s.sleepStagger(ctx) {
+				if !s.waitForStart(ctx, g.startAt) {
 					return
 				}
-				s.publishGroupBounded(ctx, g)
+				s.publishGroupBounded(ctx, g.group)
 			}
 		}()
 	}
 
+	// Every start time is an offset from ONE scan start, not a fresh
+	// sleep per group. Sleeping a full jitter slice before each group
+	// makes the delays accumulate: with the default single worker and
+	// a 3m45s jitter, 100 groups would add hours before the last
+	// upload, and Run does not arm the next interval until the scan
+	// returns — so later groups could go unvisited indefinitely.
+	scanStart := s.now()
 	for _, group := range s.groups {
 		if ctx.Err() != nil {
 			break
 		}
+		entry := staggeredGroup{group: group}
+		if stagger {
+			entry.startAt = scanStart.Add(s.jitterSlice())
+		}
 		select {
-		case work <- group:
+		case work <- entry:
 		case <-ctx.Done():
 		}
 	}
@@ -327,14 +338,29 @@ func (s *Scheduler) publishGroupBounded(ctx context.Context, group OffloadGroup)
 	s.publishGroup(ctx, group)
 }
 
-// sleepStagger waits a random slice of the jitter window. It reports
-// false when ctx ended first, so the caller abandons the group.
-func (s *Scheduler) sleepStagger(ctx context.Context) bool {
-	slice := s.jitterSlice()
-	if slice <= 0 {
+// staggeredGroup pairs a group with the absolute instant its work may
+// begin. Carrying the instant rather than a duration is what keeps
+// every start inside a single jitter window: a worker that is already
+// past a group's start time proceeds immediately instead of sleeping
+// again.
+type staggeredGroup struct {
+	group   OffloadGroup
+	startAt time.Time
+}
+
+// waitForStart blocks until startAt. A zero startAt, or one already in
+// the past because earlier groups took longer than the offset, returns
+// immediately. It reports false when ctx ended first, so the caller
+// abandons the group.
+func (s *Scheduler) waitForStart(ctx context.Context, startAt time.Time) bool {
+	if startAt.IsZero() {
+		return true
+	}
+	delay := startAt.Sub(s.now())
+	if delay <= 0 {
 		return true
 	}
-	timer := time.NewTimer(slice)
+	timer := time.NewTimer(delay)
 	defer timer.Stop()
 	select {
 	case <-timer.C:
diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go
index 703840bf3..b788a65ef 100644
--- a/internal/snapshotoffload/scheduler_test.go
+++ b/internal/snapshotoffload/scheduler_test.go
@@ -696,3 +696,56 @@ func TestSchedulerBoundsScanGoroutines(t *testing.T) {
 	require.Less(t, peakGoroutines.Load(), before+int64(groupCount)/2,
 		"a scan must not stack one goroutine per group")
 }
+
+// TestSchedulerStaggerDoesNotAccumulateAcrossGroups pins that every
+// group start lands inside ONE jitter window.
+//
+// Sleeping a fresh jitter slice before each group makes the delays
+// compound: with the default single worker and a 3m45s jitter, 100
+// groups would push the last upload hours out, and Run does not arm
+// the next interval until the scan returns — so later groups could go
+// unvisited indefinitely.
+func TestSchedulerStaggerDoesNotAccumulateAcrossGroups(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+
+	const groupCount = 12
+	groups := make([]OffloadGroup, 0, groupCount)
+	for i := range groupCount {
+		groupID := uint64(i) + 1 //nolint:gosec // loop index over a fixed-size fixture.
+		dir := seedSchedulerGroup(t, root, fmt.Sprintf("stagger-%d", i))
+		groups = append(groups, OffloadGroup{
+			GroupID:      groupID,
+			DataDir:      dir,
+			IsLeader:     func() bool { return true },
+			VerifyLeader: func(context.Context) error { return nil },
+		})
+	}
+
+	const jitter = 300 * time.Millisecond
+	s := newTestScheduler(t, store, groups, WithSchedulerJitter(jitter))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Baseline: the same scan with no stagger, so the comparison is
+	// against this fixture's real publish cost rather than a guess.
+	baselineStart := time.Now()
+	s.scan(ctx, false)
+	baseline := time.Since(baselineStart)
+
+	// A second scan republishes nothing (the high-water mark short-
+	// circuits it), so this measures scheduling overhead almost alone.
+	staggeredStart := time.Now()
+	s.scan(ctx, true)
+	staggered := time.Since(staggeredStart)
+
+	// One shared window adds at most ~jitter over the baseline.
+	// Sleeping a fresh slice per group would add groupCount*jitter/2
+	// ≈ 1.8s here; allow 3x jitter of slack for scheduling noise and
+	// the bound still separates the two by a wide margin.
+	require.Less(t, staggered, baseline+3*jitter,
+		"stagger must offset group starts from one scan start, not sleep a fresh slice per group")
+}