snapshotoffload: add the M3 two-phase retention/GC

docs/design/2026_07_19_partial_physical_snapshot_object_offload.md

@@ -1,6 +1,6 @@
 # Physical Snapshot Object Offload
 
-Status: Partial — M0/M1/M2 implemented; M3 pending
+Status: Partial — M0/M1/M2 implemented; M3 retention/GC implemented, remaining M3 items pending
 Author: bootjp
 Date: 2026-07-19
 Updated: 2026-07-23
@@ -48,7 +48,7 @@ The M1 object-store-neutral substrate now adds:
 - `cmd/elastickv-snapshot-offload publish` and `restore` for local and
   S3-backed operator workflows.
 
-The runtime scheduler and retention/GC remain pending.
+The runtime scheduler remains pending. Retention/GC is implemented per §5; the remaining M3 items (restore drills, corruption tests, multi-node acceptance, operator documentation) are pending.
 
 ## 2. Safety boundary
 
@@ -121,6 +121,37 @@ window. GC runs in two phases:
 2. after a grace period, rebuild the live payload SHA set from all remaining
    manifests and delete only payload objects with no live reference.
 
+Payload reclamation is **two-pass mark-and-sweep**. A pass that finds a
+payload unreferenced and past its grace *marks* it; only a later pass
+that finds the same object unchanged, with the mark aged past
+`MinMarkAge`, deletes it.
+
+The second pass is required because a publisher reusing a
+content-addressed payload refreshes it by rewriting **identical
+bytes**, and no conditional-delete primitive on a general-purpose S3
+bucket detects that: `If-Match` compares a content-derived ETag, which
+identical bytes leave unchanged, and `IfMatchLastModifiedTime` /
+`IfMatchSize` are directory-buckets only. A single-pass GC could
+therefore delete a payload between the publisher refreshing it and its
+manifest committing. Spanning two passes means any publish shorter than
+the inter-pass interval is observed — through the refreshed mtime or
+the newly committed manifest — before the sweep. `MinMarkAge` must
+therefore exceed the longest plausible publish.
+
+The mark state is in-memory and per-process. Losing it on restart
+delays reclamation by one pass and never advances it. Marks for objects
+absent from a pass's (complete) listing are pruned, so external
+reclamation cannot leak them.
+
+**Accepted residual.** A refresh that begins *inside* the sweep pass —
+after the revalidation and head, before the delete — is still not
+observed, because no general-purpose-bucket precondition can detect a
+content-preserving rewrite. Closing that last window needs a claim or
+lease protocol and a new key prefix; two-pass mark-and-sweep was chosen
+over that on the grounds that a publish completing entirely within the
+gap between two adjacent object-store calls is not a realistic
+scenario, while the layout change is a permanent cost.
+
 Malformed manifests fail closed: they are reported and excluded from both
 automatic manifest deletion and payload reclamation. Listing failure,
 pagination failure, or an incomplete group scan performs no deletes. This
@@ -152,6 +183,16 @@ credentials provider, schedule, retention count/window, upload concurrency,
 and server-side encryption mode. Static secrets must use file or environment
 providers and must not appear in process arguments or manifests.
 
+**Versioned buckets.** Retention deletes by key, not by version. On a
+bucket with S3 versioning enabled a keyed delete only writes a delete
+marker, so the bytes persist as a noncurrent version that later
+listings cannot see: GC reports successful reclamation while storage
+grows without bound. A versioned backup bucket therefore requires a
+noncurrent-version expiration lifecycle rule. Whether to instead
+enumerate and delete versions directly, or to refuse versioned buckets
+at startup, is an open operational decision tracked with the remaining
+M3 work.
+
 Storage-envelope encryption protects values but not all physical keys and
 metadata. The external bucket therefore requires private ACLs, TLS, and
 server-side encryption (SSE-S3 or SSE-KMS). Anonymous reads and writes are a
@@ -165,7 +206,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 | 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 |
+| M3 | Retention/GC, restore drills, corruption tests, multi-node acceptance, operational documentation | Partially implemented: the §5 two-phase retention/GC (`retention.go`) with `RetentionStore` list/delete on both the local and S3 stores. Restore corruption drills are implemented (`restore_corruption_test.go`: truncated, over-length, missing and tampered-descriptor payloads, plus a positive restore-into-fresh-dir drill). Multi-node acceptance, operational documentation, and the §7 versioned-bucket decision remain pending. |
 
 The filename and header remain `partial` until M1-M3 complete the central
 object-offload subsystem. At that point the completion PR must use `git mv` to

internal/snapshotoffload/manifest.go

@@ -23,6 +23,12 @@ var (
 	ErrObjectConflict = errors.New("snapshot offload: object conflict")
 	ErrObjectNotFound = errors.New("snapshot offload: object not found")
 
+	// ErrObjectModified is returned by DeleteObjectIfUnmodified when
+	// the object changed after the caller validated it. For retention
+	// this is not a failure: it means a concurrent publish claimed the
+	// payload, so the correct response is to leave it alone.
+	ErrObjectModified = errors.New("snapshot offload: object modified since validation")
+
 	// 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

internal/snapshotoffload/publish.go

@@ -359,23 +359,62 @@ func spoolExport(ctx context.Context, export *etcdraftengine.PersistedSnapshotEx
 }
 
 func putPayload(ctx context.Context, store ObjectStore, key string, file *os.File, size int64, sha string) error {
-	if exists, err := verifyExistingStoreObject(ctx, store, key, size, sha); err != nil {
-		return errors.Wrap(err, "verify existing snapshot payload")
-	} else if exists {
-		return nil
-	}
-	if _, err := file.Seek(0, io.SeekStart); err != nil {
-		return errors.WithStack(err)
-	}
-	info, err := store.PutObject(ctx, key, file, PutOptions{
+	opts := PutOptions{
 		Size:        size,
 		SHA256:      sha,
 		ContentType: "application/octet-stream",
-	})
+	}
+	exists, err := verifyExistingStoreObject(ctx, store, key, size, sha)
+	if err != nil {
+		return errors.Wrap(err, "verify existing snapshot payload")
+	}
+	if exists {
+		return refreshExistingPayload(ctx, store, key, file, opts)
+	}
+	if err := seekPayloadFile(file); err != nil {
+		return err
+	}
+	info, err := store.PutObject(ctx, key, file, opts)
 	if err != nil {
 		return errors.Wrap(err, "put snapshot payload")
 	}
-	if info.Size != size || (info.SHA256 != "" && info.SHA256 != sha) {
+	return validatePayloadObjectInfo(key, info, opts)
+}
+
+// refreshExistingPayload restarts a reused payload's retention grace by
+// rewriting it.
+//
+// A store that cannot refresh is a hard error, not a silent skip. The
+// §5 two-pass sweep detects a reuse precisely BECAUSE the refresh moves
+// the object's mtime; if nothing moves, retention sees an untouched
+// object, sweeps it, and the publisher commits a manifest naming bytes
+// that no longer exist. Failing here costs one publish; skipping
+// quietly costs the backup.
+func refreshExistingPayload(ctx context.Context, store ObjectStore, key string, file *os.File, opts PutOptions) error {
+	refresher, ok := store.(ObjectRefresher)
+	if !ok {
+		return errors.Wrapf(ErrInvalidOptions,
+			"object store cannot refresh existing payload %s; reuse would leave it eligible for reclamation", key)
+	}
+	if err := seekPayloadFile(file); err != nil {
+		return err
+	}
+	info, err := refresher.RefreshObject(ctx, key, file, opts)
+	if err != nil {
+		return errors.Wrap(err, "refresh existing snapshot payload")
+	}
+	return validatePayloadObjectInfo(key, info, opts)
+}
+
+func seekPayloadFile(file *os.File) error {
+	if _, err := file.Seek(0, io.SeekStart); err != nil {
+		return errors.WithStack(err)
+	}
+	return nil
+}
+
+func validatePayloadObjectInfo(key string, info ObjectInfo, opts PutOptions) error {
+	if info.Size != opts.Size || (info.SHA256 != "" && info.SHA256 != opts.SHA256) {
 		return errors.Wrapf(ErrIntegrity, "payload object %s remote integrity mismatch", key)
 	}
 	return nil

internal/snapshotoffload/restore_corruption_test.go

@@ -0,0 +1,188 @@
+package snapshotoffload
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+// The M3 corruption drills. Each plants one specific defect in a
+// published artifact and asserts restore fails closed AND leaves the
+// destination data dir absent — a half-restored dir is worse than no
+// restore, because the node would come up serving truncated state.
+
+// publishForRestoreDrill publishes a snapshot and returns the store and
+// its manifest.
+func publishForRestoreDrill(t *testing.T, root string, payload []byte) (*LocalStore, Manifest) {
+	t.Helper()
+	sourceDataDir := seedPhysicalSnapshot(t, root, payload, 21, 5, singlePeer())
+	store := newTestLocalStore(t, filepath.Join(root, "objects"))
+	manifest, err := PublishPersistedSnapshot(context.Background(), PublishOptions{
+		Store:         store,
+		DataDir:       sourceDataDir,
+		Prefix:        "cluster-a",
+		GroupID:       1,
+		SourceCluster: "cluster-a",
+	})
+	require.NoError(t, err)
+	return store, *manifest
+}
+
+// requireRestoreFailsClosed runs a restore and asserts it failed and
+// left no data dir behind.
+func requireRestoreFailsClosed(t *testing.T, store *LocalStore, manifestKey, dataDir string) error {
+	t.Helper()
+	_, err := RestorePhysicalSnapshot(context.Background(), RestoreOptions{
+		Store:       store,
+		ManifestKey: manifestKey,
+		DataDir:     dataDir,
+		Peers:       singlePeer(),
+	})
+	require.Error(t, err)
+	_, statErr := os.Stat(dataDir)
+	require.True(t, os.IsNotExist(statErr),
+		"a failed restore must leave the destination absent, not half-written")
+	return err
+}
+
+// TestRestoreRejectsTruncatedPayload covers a partial upload or a
+// truncating filesystem: the bytes hash differently AND are short. The
+// length check must fire before any content is trusted.
+func TestRestoreRejectsTruncatedPayload(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	payload := []byte("EKVTHLC1a-payload-long-enough-to-truncate-meaningfully")
+	store, manifest := publishForRestoreDrill(t, root, payload)
+
+	payloadPath, err := store.pathForKey(manifest.Payload.Key)
+	require.NoError(t, err)
+	require.NoError(t, os.WriteFile(payloadPath, payload[:len(payload)/2], 0o600))
+
+	err = requireRestoreFailsClosed(t, store, manifest.ManifestKey, filepath.Join(root, "restored"))
+	require.True(t, errors.Is(err, ErrIntegrity))
+}
+
+// TestRestoreRejectsPayloadGrownBeyondItsDeclaredLength is the
+// complement: extra bytes appended to the object.
+func TestRestoreRejectsPayloadGrownBeyondItsDeclaredLength(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	payload := []byte("EKVTHLC1a-payload-to-be-extended")
+	store, manifest := publishForRestoreDrill(t, root, payload)
+
+	payloadPath, err := store.pathForKey(manifest.Payload.Key)
+	require.NoError(t, err)
+	require.NoError(t, os.WriteFile(payloadPath, append(payload, []byte("extra")...), 0o600))
+
+	err = requireRestoreFailsClosed(t, store, manifest.ManifestKey, filepath.Join(root, "restored"))
+	require.True(t, errors.Is(err, ErrIntegrity))
+}
+
+// TestRestoreRejectsAManifestWhosePayloadIsMissing is the dangling
+// reference case — precisely the state a retention bug would leave
+// behind if it reclaimed a payload a committed manifest still names.
+// Restore must fail cleanly rather than produce an empty data dir.
+func TestRestoreRejectsAManifestWhosePayloadIsMissing(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	payload := []byte("EKVTHLC1payload-that-gets-reclaimed")
+	store, manifest := publishForRestoreDrill(t, root, payload)
+
+	require.NoError(t, store.DeleteObject(context.Background(), manifest.Payload.Key))
+
+	err := requireRestoreFailsClosed(t, store, manifest.ManifestKey, filepath.Join(root, "restored"))
+	require.True(t, errors.Is(err, ErrObjectNotFound),
+		"a manifest naming an absent payload must report the payload as missing")
+}
+
+// TestRestoreRejectsAManifestWithATamperedPayloadDescriptor covers a
+// manifest edited to name a different payload SHA, in both shapes an
+// attacker can produce.
+//
+// Leaving the self-hash stale is caught by the manifest's own
+// integrity check. Refreshing the self-hash defeats that check but is
+// still caught downstream, because the payload OBJECT KEY encodes the
+// content hash: a descriptor claiming a different SHA points at a key
+// whose bytes cannot hash to it. Both must fail closed.
+func TestRestoreRejectsAManifestWithATamperedPayloadDescriptor(t *testing.T) {
+	t.Parallel()
+
+	falseSHA := hexSHA256Bytes([]byte("a completely different payload"))
+
+	tests := []struct {
+		name            string
+		refreshSelfHash bool
+	}{
+		{name: "stale self hash", refreshSelfHash: false},
+		{name: "self hash refreshed to match the edit", refreshSelfHash: true},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			root := t.TempDir()
+			payload := []byte("EKVTHLC1payload-with-an-edited-descriptor")
+			store, manifest := publishForRestoreDrill(t, root, payload)
+
+			tampered := manifest
+			tampered.Payload.SHA256 = falseSHA
+			encoded, _, err := tampered.MarshalCanonical()
+			require.NoError(t, err)
+			if !tc.refreshSelfHash {
+				encoded = bytes.Replace(encoded,
+					[]byte(tampered.ManifestSHA256), []byte(manifest.ManifestSHA256), 1)
+			}
+			manifestPath, err := store.pathForKey(manifest.ManifestKey)
+			require.NoError(t, err)
+			require.NoError(t, os.WriteFile(manifestPath, encoded, 0o600))
+
+			err = requireRestoreFailsClosed(t, store, manifest.ManifestKey,
+				filepath.Join(root, "restored"))
+			require.True(t, errors.Is(err, ErrIntegrity),
+				"an edited payload descriptor must fail the integrity contract, got %v", err)
+		})
+	}
+}
+
+// TestRestoreDrillSucceedsIntoAFreshDirectory is the positive drill:
+// the artifact published above restores cleanly into an absent data
+// dir, which is the operation the corruption cases must not
+// half-perform.
+func TestRestoreDrillSucceedsIntoAFreshDirectory(t *testing.T) {
+	t.Parallel()
+
+	root := t.TempDir()
+	payload := []byte("EKVTHLC1a-healthy-payload-for-the-drill")
+	store, manifest := publishForRestoreDrill(t, root, payload)
+
+	restoreDataDir := filepath.Join(root, "restored")
+	result, err := RestorePhysicalSnapshot(context.Background(), RestoreOptions{
+		Store:       store,
+		ManifestKey: manifest.ManifestKey,
+		DataDir:     restoreDataDir,
+		Peers:       singlePeer(),
+	})
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.DirExists(t, restoreDataDir)
+
+	// A second restore into the now-populated dir must refuse rather
+	// than overwrite: the preflight is what protects an operator who
+	// re-runs the drill against a live node's data dir.
+	_, err = RestorePhysicalSnapshot(context.Background(), RestoreOptions{
+		Store:       store,
+		ManifestKey: manifest.ManifestKey,
+		DataDir:     restoreDataDir,
+		Peers:       singlePeer(),
+	})
+	require.Error(t, err, "restore must refuse a destination that already exists")
+}