store/s3: sparse-aware, chunked, content-addressed checkpoint transfer

docs/sandboxd-api.md

@@ -750,7 +750,9 @@ concurrent-heal cap is already full — retryable, and the response carries a
 The peer-transfer route behind the probe and heal above — internal, not
 part of the public API; an SDK caller has no reason to call it directly.
 
-- `GET` streams the whole record — guest memory, disk, and meta — as a tar.
+- `GET` streams the whole record — guest memory, disk, and meta — as a tar
+  in which every export file travels as its size plus its non-zero 16 MiB chunks,
+  each with its SHA-256, so holes cost nothing on the wire.
   Operator-token only: the stream carries no tenant scoping, and its only
   real caller is a peer's heal pull, which presents the fleet `api_token`.
   401 missing or unrecognized token, 403 a valid tenant token (authenticated

sandboxd/go.mod

@@ -5,7 +5,6 @@ go 1.27.1
 require (
 	github.com/aws/aws-sdk-go-v2 v1.47.3
 	github.com/aws/aws-sdk-go-v2/config v1.33.9
-	github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.4.13
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.114.0
 	github.com/aws/smithy-go v1.28.5
 	github.com/cocoonstack/sandbox/protocol/wire v0.0.0-00010101000000-000000000000

sandboxd/go.sum

@@ -8,8 +8,6 @@ github.com/aws/aws-sdk-go-v2/credentials v1.20.9 h1:SNO55PnWhaDafsOcv16lCA0kdoFM
 github.com/aws/aws-sdk-go-v2/credentials v1.20.9/go.mod h1:+2reOTRtdD3VSBqpRDNwrZIYy8yMCYa7oK3WKTJ/I1I=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.3 h1:gr3b87OLB29jymWPzp/CAyXwQRJVE+sC4Wm0qBIKHsk=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.3/go.mod h1:BRjf4L/Yx2hY54o+ZibPbLJyMMeSwqYQXwcJfqW66a8=
-github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.4.13 h1:wO7TVbywHwdpHLUiX6DnmP2RDYOACVeJCb6zMfSFViU=
-github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.4.13/go.mod h1:Zc9r0r7wMid/NkbsLrkGxe5vZufWyP0CiC2dDXZ8ldk=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.6 h1:CqL8NN3p4I0JTwwV5MEoatIl4Y2amZ2X9dRD+0hbRL0=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.6/go.mod h1:exdJ32hMqXl0Wyo6NwqEC5yIhtua3+PDaKE8P29m4qE=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.6 h1:FpBq24TG6WVSObTNhQOv7Q0088rounUUlwn3zlNItkM=

sandboxd/pool/archive_test.go

@@ -51,6 +51,23 @@ func TestArchivePublishWindowPinsCheckpoint(t *testing.T) {
 	}
 }
 
+func TestArchivePublishesUnderTheSandboxLineage(t *testing.T) {
+	eng := newFakeEngine()
+	m := newTestManager(t, eng, archivePool(3600))
+	sb := mustClaim(t, m, testKey)
+	if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil {
+		t.Fatalf("hibernate: %v", err)
+	}
+	rec := &lineageStore{Store: m.ckpts}
+	m.ckpts = rec
+	if err := m.archive(t.Context(), sb); err != nil {
+		t.Fatalf("archive: %v", err)
+	}
+	if rec.lineage != sb.ID {
+		t.Errorf("archive lineage = %q, want the sandbox id %q", rec.lineage, sb.ID)
+	}
+}
+
 func TestArchiveWakeRehibernateWindowAborts(t *testing.T) {
 	eng := newFakeEngine()
 	m := newTestManager(t, eng, archivePool(3600))
@@ -155,7 +172,7 @@ func TestReconcileReclaimsOrphanArchiveCk(t *testing.T) {
 		if err := os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(meta), 0o600); err != nil {
 			t.Fatalf("meta: %v", err)
 		}
-		if err := m.ckpts.Publish(t.Context(), staging, id); err != nil {
+		if err := m.ckpts.Publish(t.Context(), staging, id, ""); err != nil {
 			t.Fatalf("publish: %v", err)
 		}
 	}
@@ -920,15 +937,25 @@ type stallingStore struct {
 	release   chan struct{}
 }
 
-func (s *stallingStore) Publish(ctx context.Context, staging, id string) error {
-	if err := s.Store.Publish(ctx, staging, id); err != nil {
+func (s *stallingStore) Publish(ctx context.Context, staging, id, lineage string) error {
+	if err := s.Store.Publish(ctx, staging, id, lineage); err != nil {
 		return err
 	}
 	s.published <- id
 	<-s.release
 	return nil
 }
 
+type lineageStore struct {
+	store.Store
+	lineage string
+}
+
+func (s *lineageStore) Publish(ctx context.Context, staging, id, lineage string) error {
+	s.lineage = lineage
+	return s.Store.Publish(ctx, staging, id, lineage)
+}
+
 type failingDeleteStore struct {
 	store.Store
 	attempts chan string

sandboxd/pool/checkpoint.go

@@ -189,7 +189,7 @@ func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID
 	if err := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil {
 		return types.Checkpoint{}, "", fmt.Errorf("write checkpoint meta: %w", err)
 	}
-	if err := m.ckpts.Publish(ctx, staging, ckpt.ID); err != nil {
+	if err := m.ckpts.Publish(ctx, staging, ckpt.ID, sb.ID); err != nil {
 		return types.Checkpoint{}, "", fmt.Errorf("commit checkpoint: %w", err)
 	}
 	return ckpt, srcSnap, nil
@@ -278,7 +278,7 @@ func (m *Manager) runHeal(ctx context.Context, ckptID string) (types.Checkpoint,
 	if ckpt, err := m.loadCheckpoint(ctx, ckptID); err == nil {
 		return ckpt, nil // published by another path while this one staged
 	}
-	if err := m.ckpts.Publish(ctx, staging, ckptID); err != nil {
+	if err := m.ckpts.Publish(ctx, staging, ckptID, ""); err != nil {
 		return types.Checkpoint{}, fmt.Errorf("publish healed checkpoint: %w", err)
 	}
 	return m.loadCheckpoint(ctx, ckptID)

sandboxd/pool/checkpoint_test.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -363,6 +364,29 @@ func TestRejectedCheckpointOpsLeakNoRecLock(t *testing.T) {
 	}
 }
 
+func TestSharedStoreSweepsOncePerTick(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		shared    bool
+		wantCkpts int32
+		wantTpls  int32
+	}{
+		{"per-node store sweeps both views", false, 1, 1},
+		{"shared store sweeps its one prefix once", true, 1, 0},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey})
+			ckpts := &sweepCountingStore{Store: m.ckpts}
+			tpls := &sweepCountingStore{Store: m.tpls}
+			m.ckpts, m.tpls, m.ckptsShared = ckpts, tpls, tc.shared
+			m.sweepStoreGenerations(t.Context())
+			if ckpts.sweeps.Load() != tc.wantCkpts || tpls.sweeps.Load() != tc.wantTpls {
+				t.Errorf("sweeps = checkpoints %d, templates %d, want %d, %d", ckpts.sweeps.Load(), tpls.sweeps.Load(), tc.wantCkpts, tc.wantTpls)
+			}
+		})
+	}
+}
+
 type stallingMetasStore struct {
 	store.Store
 	entered chan struct{}
@@ -378,17 +402,27 @@ func (s *stallingMetasStore) Metas(ctx context.Context) ([]store.Record, error)
 	return s.Store.Metas(ctx)
 }
 
+type sweepCountingStore struct {
+	store.Store
+	sweeps atomic.Int32
+}
+
+func (s *sweepCountingStore) SweepGenerations(ctx context.Context) error {
+	s.sweeps.Add(1)
+	return s.Store.SweepGenerations(ctx)
+}
+
 type stallingSweepStore struct {
 	store.Store
 	entered chan struct{}
 	release chan struct{}
 }
 
-func (s *stallingSweepStore) SweepGenerations() error {
+func (s *stallingSweepStore) SweepGenerations(ctx context.Context) error {
 	select {
 	case s.entered <- struct{}{}:
 	default:
 	}
 	<-s.release
-	return s.Store.SweepGenerations()
+	return s.Store.SweepGenerations(ctx)
 }

sandboxd/pool/expiry_test.go

@@ -312,10 +312,10 @@ type failingPublishStore struct {
 	publishes atomic.Int32
 }
 
-func (s *failingPublishStore) Publish(ctx context.Context, staging, id string) error {
+func (s *failingPublishStore) Publish(ctx context.Context, staging, id, lineage string) error {
 	s.publishes.Add(1)
 	if s.fail.Load() {
 		return errors.New("store unavailable")
 	}
-	return s.Store.Publish(ctx, staging, id)
+	return s.Store.Publish(ctx, staging, id, lineage)
 }

sandboxd/pool/failover_test.go

@@ -1914,7 +1914,7 @@ func seedArchiveCk(t *testing.T, m *Manager, id, sandboxID string) {
 	if err := os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(meta), 0o600); err != nil {
 		t.Fatalf("meta: %v", err)
 	}
-	if err := m.ckpts.Publish(t.Context(), staging, id); err != nil {
+	if err := m.ckpts.Publish(t.Context(), staging, id, ""); err != nil {
 		t.Fatalf("publish: %v", err)
 	}
 }
@@ -1982,9 +1982,9 @@ type publishCountingStore struct {
 	n atomic.Int32
 }
 
-func (s *publishCountingStore) Publish(ctx context.Context, staging, id string) error {
+func (s *publishCountingStore) Publish(ctx context.Context, staging, id, lineage string) error {
 	s.n.Add(1)
-	return s.Store.Publish(ctx, staging, id)
+	return s.Store.Publish(ctx, staging, id, lineage)
 }
 
 type unreadableStore struct {

sandboxd/pool/pool.go

@@ -676,10 +676,13 @@ func (m *Manager) sweepStoreGenerations(ctx context.Context) {
 	}
 	defer m.genSweeping.Store(false)
 	logger := log.WithFunc("pool.sweepStoreGenerations")
-	if err := m.ckpts.SweepGenerations(); err != nil {
+	if err := m.ckpts.SweepGenerations(ctx); err != nil {
 		logger.Error(ctx, err, "sweep checkpoint generations")
 	}
-	if err := m.tpls.SweepGenerations(); err != nil {
+	if m.ckptsShared {
+		return
+	}
+	if err := m.tpls.SweepGenerations(ctx); err != nil {
 		logger.Error(ctx, err, "sweep template generations")
 	}
 }

sandboxd/pool/promote_test.go

@@ -441,7 +441,7 @@ func TestTemplatesListWhatTheNodeHoldsAcrossARestart(t *testing.T) {
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+oldID+`","created_at":"2026-01-01T00:00:00Z"}`), 0o600); err != nil {
 		t.Fatalf("write keyless meta: %v", err)
 	}
-	if _, err = m.tpls.PublishDigested(t.Context(), staging, oldID); err != nil {
+	if _, err = m.tpls.PublishDigested(t.Context(), staging, oldID, ""); err != nil {
 		t.Fatalf("publish keyless record: %v", err)
 	}
 

sandboxd/pool/template.go

@@ -303,7 +303,7 @@ func (m *Manager) commitTemplate(ctx context.Context, staging string, rec templa
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil {
 		return "", err
 	}
-	digest, err := m.tpls.PublishDigested(ctx, staging, id)
+	digest, err := m.tpls.PublishDigested(ctx, staging, id, "")
 	if err != nil {
 		return "", fmt.Errorf("publish template: %w", err)
 	}

sandboxd/store/dir/dir.go

@@ -56,11 +56,11 @@ func (d *Store) Stage(id string) (string, error) {
 	return os.MkdirTemp(d.root, id+"-*.tmp")
 }
 
-func (d *Store) Publish(ctx context.Context, staging, id string) error {
+func (d *Store) Publish(ctx context.Context, staging, id, _ string) error {
 	return d.publish(ctx, staging, id, "")
 }
 
-func (d *Store) PublishDigested(ctx context.Context, staging, id string) (string, error) {
+func (d *Store) PublishDigested(ctx context.Context, staging, id, _ string) (string, error) {
 	digest, err := hashExport(ctx, filepath.Join(staging, store.ExportDir))
 	if err != nil {
 		return "", fmt.Errorf("digest export: %w", err)
@@ -155,10 +155,14 @@ func (d *Store) SweepStaging() error {
 	if err := utils.RemoveDirEntries(d.root, func(name string) bool { return strings.HasSuffix(name, ".tmp") }); err != nil {
 		return err
 	}
-	return d.SweepGenerations()
+	return d.sweepRecords()
 }
 
-func (d *Store) SweepGenerations() error {
+func (d *Store) SweepGenerations(context.Context) error {
+	return d.sweepRecords()
+}
+
+func (d *Store) sweepRecords() error {
 	entries, err := os.ReadDir(d.root)
 	if err != nil {
 		return err

sandboxd/store/dir/dir_test.go

@@ -134,7 +134,7 @@ func TestPublishDigestedChunkBoundaries(t *testing.T) {
 			if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(metaJSON(tt.name)), 0o600); err != nil {
 				t.Fatalf("write meta: %v", err)
 			}
-			digest, err := st.PublishDigested(t.Context(), staging, id)
+			digest, err := st.PublishDigested(t.Context(), staging, id, "")
 			if err != nil {
 				t.Fatalf("PublishDigested: %v", err)
 			}
@@ -216,7 +216,7 @@ func TestPublishDigestedFailureAndFreshRetry(t *testing.T) {
 	if err = os.Mkdir(blocker, 0o750); err != nil {
 		t.Fatalf("block digest commit: %v", err)
 	}
-	if digest, publishErr := st.PublishDigested(t.Context(), failed, id); publishErr == nil || digest != "" {
+	if digest, publishErr := st.PublishDigested(t.Context(), failed, id, ""); publishErr == nil || digest != "" {
 		t.Fatalf("PublishDigested with blocked sidecar = %q, %v, want empty/error", digest, publishErr)
 	}
 	dir, meta, digest, err := st.Fetch(t.Context(), id)
@@ -274,7 +274,7 @@ func TestSweepGenerationsPairsCurrentAndSupersededDigests(t *testing.T) {
 		backdate(t, path)
 	}
 
-	if err = st.SweepGenerations(); err != nil {
+	if err = st.SweepGenerations(t.Context()); err != nil {
 		t.Fatalf("SweepGenerations: %v", err)
 	}
 	for _, path := range []string{firstGen, firstSidecar} {
@@ -311,7 +311,7 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) {
 		t.Fatalf("stage retry: %v", err)
 	}
 	seedRecord(t, retry, "same")
-	if publishErr := st.Publish(t.Context(), retry, id); publishErr == nil {
+	if publishErr := st.Publish(t.Context(), retry, id, ""); publishErr == nil {
 		t.Fatal("expired installed generation was revived")
 	}
 	if removeErr := os.RemoveAll(retry); removeErr != nil {
@@ -320,7 +320,7 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) {
 	if _, readErr := st.ReadMeta(t.Context(), id); !errors.Is(readErr, store.ErrNotFound) {
 		t.Fatalf("expired retry committed meta: %v", readErr)
 	}
-	if sweepErr := st.SweepGenerations(); sweepErr != nil {
+	if sweepErr := st.SweepGenerations(t.Context()); sweepErr != nil {
 		t.Fatalf("sweep expired generation: %v", sweepErr)
 	}
 	if _, statErr := os.Stat(gen); !errors.Is(statErr, fs.ErrNotExist) {
@@ -454,7 +454,7 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) {
 	if mkdirErr := os.MkdirAll(commitBlocker, 0o750); mkdirErr != nil {
 		t.Fatalf("block meta commit: %v", mkdirErr)
 	}
-	if publishErr := writer.Publish(t.Context(), staging, publishID); publishErr == nil {
+	if publishErr := writer.Publish(t.Context(), staging, publishID, ""); publishErr == nil {
 		t.Fatal("publish with blocked meta commit succeeded")
 	}
 	fresh := filepath.Join(root, publishID, store.ExportGen(meta))
@@ -513,7 +513,7 @@ func TestSweepGenerationsToleratesConcurrentDelete(t *testing.T) {
 			if err := os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(metaJSON(id)), 0o600); err != nil {
 				return err
 			}
-			if err := st.Publish(ctx, staging, id); err != nil {
+			if err := st.Publish(ctx, staging, id, ""); err != nil {
 				return fmt.Errorf("publish: %w", err)
 			}
 			if err := st.Delete(ctx, id); err != nil {
@@ -529,7 +529,7 @@ func TestSweepGenerationsToleratesConcurrentDelete(t *testing.T) {
 				return nil
 			default:
 			}
-			if err := st.SweepGenerations(); err != nil {
+			if err := st.SweepGenerations(t.Context()); err != nil {
 				return fmt.Errorf("sweep: %w", err)
 			}
 		}
@@ -646,7 +646,7 @@ func mustPublish(t testing.TB, st *Store, id, metaID string) {
 		t.Fatalf("stage %s: %v", metaID, err)
 	}
 	seedRecord(t, staging, metaID)
-	if err := st.Publish(t.Context(), staging, id); err != nil {
+	if err := st.Publish(t.Context(), staging, id, ""); err != nil {
 		t.Fatalf("publish %s: %v", metaID, err)
 	}
 }
@@ -658,7 +658,7 @@ func mustPublishDigested(t *testing.T, st *Store, id, metaID string) string {
 		t.Fatalf("stage %s: %v", metaID, err)
 	}
 	seedRecord(t, staging, metaID)
-	digest, err := st.PublishDigested(t.Context(), staging, id)
+	digest, err := st.PublishDigested(t.Context(), staging, id, "")
 	if err != nil {
 		t.Fatalf("publish digested %s: %v", metaID, err)
 	}