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/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/manifest.go b/internal/snapshotoffload/manifest.go index f750ce0ba..ca30a7686 100644 --- a/internal/snapshotoffload/manifest.go +++ b/internal/snapshotoffload/manifest.go @@ -22,6 +22,20 @@ var ( ErrIntegrity = errors.New("snapshot offload: integrity check failed") ErrObjectConflict = errors.New("snapshot offload: object conflict") ErrObjectNotFound = errors.New("snapshot offload: object not found") + + // ErrNoPersistedSnapshot reports that the LOCAL data dir has no + // persisted snapshot yet. It is deliberately distinct from + // ErrObjectNotFound: a young group that has not snapshotted is a + // normal scan outcome, whereas an object vanishing from the store + // mid-publish is a real failure, and collapsing the two would + // silence the second. + ErrNoPersistedSnapshot = errors.New("snapshot offload: no persisted snapshot available") + + // ErrSnapshotNotNewer reports that the persisted snapshot is not + // newer than the caller's high-water mark, so nothing was + // published. It is a normal outcome for a scheduler tick over an + // unchanged snapshot, not a failure. + ErrSnapshotNotNewer = errors.New("snapshot offload: persisted snapshot is not newer than the last published index") ) type Manifest struct { 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 734359d20..d53914c0f 100644 --- a/internal/snapshotoffload/publish.go +++ b/internal/snapshotoffload/publish.go @@ -25,6 +25,24 @@ type PublishOptions struct { BinaryVersion string CreatedAt time.Time SpoolDir string + // VerifyLeader, when set, is re-checked immediately before the manifest is + // committed. §4 requires leadership to hold at that instant, not merely when + // the snapshot was opened: spooling a multi-gigabyte payload takes long + // enough to lose an election. Failing here can leave an unreferenced + // content-addressed payload, which GC reclaims, but never a committed + // manifest naming a snapshot this node no longer had the right to publish. + VerifyLeader func(context.Context) error + // SkipIfNotNewerThan suppresses the publish when the persisted + // snapshot's index is not greater than this value. Zero disables + // the check. + // + // The comparison happens after the export is opened but BEFORE + // the payload is spooled, which is the whole point: a scheduler + // that ticks every 15 minutes over an unchanged snapshot would + // otherwise re-read, re-hash and re-fsync a multi-gigabyte + // payload every tick just to discover the object store already + // has it. + SkipIfNotNewerThan uint64 } func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manifest, error) { @@ -38,6 +56,10 @@ func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manife defer func() { _ = export.Close() }() metadata := export.Metadata() + if opts.SkipIfNotNewerThan > 0 && metadata.Index <= opts.SkipIfNotNewerThan { + return nil, errors.Wrapf(ErrSnapshotNotNewer, + "persisted snapshot index %d is not newer than %d", metadata.Index, opts.SkipIfNotNewerThan) + } payloadFile, payloadSHA, payloadBytes, err := spoolExport(ctx, export, publishSpoolDir(opts)) if err != nil { return nil, err @@ -56,6 +78,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,7 +98,7 @@ func PublishPersistedSnapshot(ctx context.Context, opts PublishOptions) (*Manife if err := validateManifest(*manifest); err != nil { return nil, err } - if err := putManifest(ctx, opts.Store, manifest, opts.CreatedAt.IsZero()); err != nil { + if err := putManifest(ctx, opts.Store, manifest, opts.CreatedAt.IsZero(), opts.VerifyLeader); err != nil { return nil, err } return manifest, nil @@ -75,7 +110,7 @@ func openPublishExport(dataDir string) (*etcdraftengine.PersistedSnapshotExport, return nil, errors.Wrap(err, "open persisted snapshot export") } if !ok { - return nil, errors.Wrap(ErrObjectNotFound, "no persisted snapshot available") + return nil, errors.WithStack(ErrNoPersistedSnapshot) } return export, nil } @@ -113,7 +148,13 @@ func buildManifest( }, nil } -func putManifest(ctx context.Context, store ObjectStore, manifest *Manifest, reuseExistingCreatedAt bool) error { +func putManifest( + ctx context.Context, + store ObjectStore, + manifest *Manifest, + reuseExistingCreatedAt bool, + verifyLeader func(context.Context) error, +) error { data, manifestSHA, err := manifest.MarshalCanonical() if err != nil { return err @@ -125,6 +166,17 @@ func putManifest(ctx context.Context, store ObjectStore, manifest *Manifest, reu } else if exists { return nil } + // §4: leadership must hold at the instant the manifest is created, + // not merely before the absence probe above. That probe is a remote + // read whose latency is unbounded by anything the caller controls, + // so checking before it leaves a window in which a demoted node + // still commits a manifest — precisely the guarantee this + // scheduler exists to provide. + if verifyLeader != nil { + if err := verifyLeader(ctx); err != nil { + return errors.Wrap(err, "snapshot offload: leadership lost before manifest commit") + } + } if err := createManifestObject(ctx, store, manifest, data, size, objectSHA, reuseExistingCreatedAt); err != nil { return err } @@ -231,9 +283,22 @@ func manifestMatchesCandidate(existing Manifest, candidate Manifest, reuseExisti return reflect.DeepEqual(existing, candidate) } +// sameManifestExceptCreation compares a retry's candidate against the +// committed manifest, ignoring the fields that legitimately differ +// between two publishes of the SAME snapshot. +// +// BinaryVersion is one of them. It records which binary published the +// artifact, not anything about the snapshot itself, so after an +// upgrade a process that republishes an index it has not yet published +// locally would otherwise conflict with the manifest the previous +// binary committed — and keep failing every scan until Raft happens to +// produce a new snapshot. The committed manifest keeps the original +// publisher's version, which is the correct audit record for the +// bytes that actually exist. func sameManifestExceptCreation(existing Manifest, candidate Manifest) bool { candidate.CreatedAt = existing.CreatedAt candidate.ManifestSHA256 = existing.ManifestSHA256 + candidate.BinaryVersion = existing.BinaryVersion return reflect.DeepEqual(existing, candidate) } diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go new file mode 100644 index 000000000..2db05eee1 --- /dev/null +++ b/internal/snapshotoffload/scheduler.go @@ -0,0 +1,498 @@ +package snapshotoffload + +import ( + "context" + "log/slog" + "math/rand/v2" + "strings" + "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 + // 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{} + // 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. +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. +// 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, + 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), + inFlight: make(map[uint64]struct{}), + } + 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 + } + // 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 +} + +func (s *Scheduler) validate() error { + // 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") + 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 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) + case group.VerifyLeader == nil: + return errors.Wrapf(ErrInvalidOptions, + "snapshot offload group %d requires a VerifyLeader callback", group.GroupID) + } + } + 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.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 0 + } + 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) { + // 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) { + // 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 staggeredGroup) + 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 !s.waitForStart(ctx, g.startAt) { + return + } + 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 <- entry: + 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) +} + +// 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(delay) + 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 + // 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 + } + 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: 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. + 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 + } + 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 + } + 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)) +} + +// 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 { + 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 +// 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() + 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..b788a65ef --- /dev/null +++ b/internal/snapshotoffload/scheduler_test.go @@ -0,0 +1,751 @@ +package snapshotoffload + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + "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 := 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()) + + 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 := newTestScheduler(t, store, []OffloadGroup{{ + GroupID: 7, + DataDir: dataDir, + IsLeader: func() bool { return true }, + VerifyLeader: func(context.Context) error { return lost }, + }}, 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 }, + VerifyLeader: func(context.Context) error { return nil }, + }} + + 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 := newTestScheduler(t, store, groups, 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 + }, + VerifyLeader: func(context.Context) error { return nil }, + }) + } + + s := newTestScheduler(t, store, groups) + 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 := 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() + require.Empty(t, published) + require.Empty(t, failed, "cancellation is shutdown, not a publish failure") +} + +func TestSchedulerRunRequiresStoreAndSourceCluster(t *testing.T) { + t.Parallel() + + _, err := NewScheduler(nil, nil, "p", "c", "v") + require.ErrorIs(t, err, ErrInvalidOptions) + store := newTestLocalStore(t, t.TempDir()) + _, 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) +} + +// 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") +} + +// 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") +} + +// 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() +} + +// 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") +} + +// 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") +}