diff --git a/docs/deploy.md b/docs/deploy.md index 5ffe8288..ff2f1d30 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -100,7 +100,7 @@ sandboxd reads one JSON file (`-config`, default | `preview_secret` | — | cluster-shared HMAC secret signing preview tokens (all nodes share one) | | `preview_advertise` | = `preview_listen` | the browser-facing preview base URL; nodes behind one TLS proxy may share it, while signed tokens route internally through each owner's `advertise_addr` | | `checkpoint_dir` | `/checkpoints` | where checkpoints and promoted templates live. Point it at a shared FUSE mount (JuiceFS over object storage, NFS) and every node sharing the mount can branch every checkpoint — records are generation-addressed with `meta.json` as the atomic commit pointer, so no cross-node locking is required of the filesystem. A re-publish retains its superseded export generation for at least ~1h and until a following hourly sweep so an in-flight clone that resolved the old metadata can finish; budget the current generation plus every generation retained across that grace-and-sweep window. An explicit delete can make a concurrent clone fail visibly. One contract on any shared root (mount or bucket): a template key has a single writer — promotes go to the sandbox's owner node, and operators must not race promotes of one name from different nodes (checkpoint ids are node-generated and never collide) | -| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. Re-publish retains prior export generations until Delete so an in-flight fetch that selected old metadata can finish; budget storage for those generations. An explicit S3 Delete can still make a concurrent fetch that has not finished materializing fail visibly. A crash between upload and the meta.json commit marker leaves orphan objects invisible to listings — add an S3 lifecycle rule to reclaim them. Absent = the dir backend at `checkpoint_dir` | +| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true, "sparse": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). `sparse` packs only allocated file extents into 64 MiB data objects and reconstructs holes on fetch; default false for rolling-upgrade compatibility. Deploy a sparse-capable sandboxd to every node before enabling it cluster-wide. Credentials come from the standard AWS chain (env/IAM role), never this file. Re-publish retains prior export generations until Delete so an in-flight fetch that selected old metadata can finish; budget storage for those generations. An explicit S3 Delete can still make a concurrent fetch that has not finished materializing fail visibly. A crash between upload and the meta.json commit marker leaves orphan objects invisible to listings — add an S3 lifecycle rule to reclaim them. Absent = the dir backend at `checkpoint_dir` | | `checkpoint_ttl_hours` | 0 (keep forever) | ages out checkpoints older than this; the sweep runs hourly and at startup. Explicit deletes never wait for it. Must be nonzero and match fleet-wide when `checkpoint_peer_heal` is on — it is the expiry eligibility point for a healed replica a delete broadcast missed, after which its next successful hourly sweep removes it; persistent sweep failure extends retention until one succeeds, so it is not a hard ceiling | | `checkpoint_peer_heal` | false | on a cluster, lets a node pull a checkpoint it lacks from a peer — found via a live probe, not gossip — rather than failing the branch; see [placement lifecycle](cluster.md#checkpoints-on-a-cluster). Three requirements, all enforced at config load: a nonempty `api_token` (the blob transfer between peers authenticates with it; without one the raw record stream would be open), `mesh.cluster_key` set (the pull presents the fleet `api_token` to an address learned from the peer probe, so the gossip layer carrying that address must itself be authenticated), and `checkpoint_ttl_hours` nonzero (a replica a delete broadcast missed becomes eligible for expiry after it, and its next successful hourly sweep removes it — so it is the finite eligibility point, not an exact ceiling). A shared checkpoint store (`checkpoint_store` kind `s3`) ignores this setting — every node already resolves every checkpoint directly, so there is nothing to heal | | `warm_max` (pool entry) | 0 (static) | turns on the demand-adaptive watermark for that pool: the warm target rises from `warm` toward `warm_max` while claims arrive faster than the measured provision lead covers, and decays back over ~a minute of silence | @@ -276,7 +276,7 @@ here validates on load: "checkpoint_store": { "kind": "s3", - "s3": {"bucket": "sandbox-ckpt", "prefix": "ck/", "region": "us-east-1"} + "s3": {"bucket": "sandbox-ckpt", "prefix": "ck/", "region": "us-east-1", "sparse": true} }, "checkpoint_ttl_hours": 168, diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 3df01115..c1b947f4 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -7,10 +7,12 @@ package s3 import ( "bytes" "context" + "encoding/json" "errors" "fmt" "io" "os" + pathpkg "path" "path/filepath" "regexp" "slices" @@ -41,6 +43,7 @@ type Config struct { Endpoint string `json:"endpoint,omitempty"` Region string `json:"region,omitempty"` ForcePathStyle bool `json:"force_path_style,omitempty"` + Sparse bool `json:"sparse,omitempty"` } type publishFile struct { @@ -49,6 +52,13 @@ type publishFile struct { digestPath string } +type publishChunk struct { + path string + key string + size int64 + extents []sparseExtent +} + type transferManager interface { UploadObject(context.Context, *transfermanager.UploadObjectInput, ...func(*transfermanager.Options)) (*transfermanager.UploadObjectOutput, error) DownloadObject(context.Context, *transfermanager.DownloadObjectInput, ...func(*transfermanager.Options)) (*transfermanager.DownloadObjectOutput, error) @@ -64,6 +74,7 @@ type Store struct { prefix string staging string idRe *regexp.Regexp + sparse bool fetches singleflight.Group } @@ -94,7 +105,10 @@ func New(ctx context.Context, cfg Config, stagingRoot string, idRe *regexp.Regex o.PartSizeBytes = uploadPartSizeBytes o.Concurrency = 8 }) - return &Store{client: client, tm: tm, bucket: cfg.Bucket, prefix: cfg.Prefix, staging: stagingRoot, idRe: idRe}, nil + return &Store{ + client: client, tm: tm, bucket: cfg.Bucket, prefix: cfg.Prefix, + staging: stagingRoot, idRe: idRe, sparse: cfg.Sparse, + }, nil } func (s *Store) Stage(id string) (string, error) { @@ -216,7 +230,8 @@ func (s *Store) publish(ctx context.Context, staging, id string, digested bool) if err != nil { return "", fmt.Errorf("staging has no %s: %w", store.MetaFile, err) } - files, err := s.publishFiles(staging, id, store.ExportGen(metaRaw), digested) + gen := store.ExportGen(metaRaw) + files, chunks, sparse, err := s.publishFiles(staging, id, gen, digested) if err != nil { return "", err } @@ -235,9 +250,27 @@ func (s *Store) publish(ctx context.Context, staging, id string, digested bool) return digestErr }) } + for _, chunk := range chunks { + g.Go(func() error { + return s.uploadExtents(gctx, chunk.key, chunk.path, chunk.size, chunk.extents) + }) + } if err = g.Wait(); err != nil { return "", err } + if len(sparse.Files) > 0 { + raw, marshalErr := json.Marshal(sparse) + if marshalErr != nil { + return "", fmt.Errorf("encode sparse manifest: %w", marshalErr) + } + if len(raw) > sparseManifestMaxBytes { + return "", fmt.Errorf("sparse manifest exceeds %d bytes", sparseManifestMaxBytes) + } + manifestKey := s.key(id, gen+"/"+sparseManifestObject) + if err = s.uploadReader(ctx, manifestKey, bytes.NewReader(raw), int64(len(raw)), nil); err != nil { + return "", err + } + } digest := "" var metadata map[string]string if digested { @@ -257,12 +290,24 @@ func (s *Store) publish(ctx context.Context, staging, id string, digested bool) return digest, nil } -func (s *Store) publishFiles(staging, id, gen string, digested bool) ([]publishFile, error) { - root := staging +func (s *Store) publishFiles( + staging, id, gen string, + digested bool, +) ([]publishFile, []publishChunk, sparseManifest, error) { if digested { - root = filepath.Join(staging, store.ExportDir) + files, err := s.publishDigestFiles(filepath.Join(staging, store.ExportDir), id, gen) + return files, nil, sparseManifest{Version: sparseManifestVersion}, err } + return s.publishCheckpointFiles(filepath.Join(staging, store.ExportDir), id, gen, s.sparse) +} + +func (s *Store) publishCheckpointFiles( + root, id, gen string, + sparseEnabled bool, +) ([]publishFile, []publishChunk, sparseManifest, error) { var files []publishFile + var chunks []publishChunk + manifest := sparseManifest{Version: sparseManifestVersion} err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { if err != nil || entry.IsDir() { return err @@ -271,20 +316,47 @@ func (s *Store) publishFiles(staging, id, gen string, digested bool) ([]publishF if err != nil { return err } - if !digested { - if rel == store.MetaFile { + exportRel := filepath.ToSlash(rel) + if exportRel == sparseObjectPrefix || strings.HasPrefix(exportRel, sparseObjectPrefix+"/") { + return fmt.Errorf("export path %s uses reserved sparse namespace", exportRel) + } + if sparseEnabled { + sparseFile, sparseChunks, ok, sparseErr := planSparseFile(path, exportRel) + if sparseErr != nil { + return sparseErr + } + if ok { + manifest.Files = append(manifest.Files, sparseFile) + for _, chunk := range sparseChunks { + chunks = append(chunks, publishChunk{ + path: path, key: s.key(id, gen+"/"+chunk.Object), + size: chunk.Size, extents: chunk.Extents, + }) + } return nil } - files = append(files, publishFile{ - path: path, - key: s.key(id, gen+strings.TrimPrefix(rel, store.ExportDir)), - }) - return nil + } + files = append(files, publishFile{path: path, key: s.key(id, gen+"/"+exportRel)}) + return nil + }) + slices.SortFunc(manifest.Files, func(a, b sparseFile) int { return strings.Compare(a.Path, b.Path) }) + return files, chunks, manifest, err +} + +func (s *Store) publishDigestFiles(root, id, gen string) ([]publishFile, error) { + var files []publishFile + err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + if err != nil || entry.IsDir() { + return err } info, err := entry.Info() if err != nil { return err } + rel, err := filepath.Rel(root, path) + if err != nil { + return err + } rel = filepath.ToSlash(rel) if !info.Mode().IsRegular() { return fmt.Errorf("digest export %s is not a regular file", rel) @@ -324,16 +396,46 @@ func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string if len(keys) == 0 { return fmt.Errorf("record %s has no export", id) } + manifest, hasSparse, err := s.readSparseManifest(ctx, exportPrefix, keys) + if err != nil { + return err + } + exportRoot := filepath.Join(local, store.ExportDir) + if err := materializeSparseFiles(exportRoot, manifest.Files); err != nil { + return err + } g, gctx := errgroup.WithContext(ctx) g.SetLimit(4) for _, key := range keys { + if hasSparse && (key == exportPrefix+sparseManifestObject || strings.HasPrefix(key, exportPrefix+sparseObjectPrefix+"/")) { + continue + } + destination, pathErr := exportPath(exportRoot, strings.TrimPrefix(key, exportPrefix)) + if pathErr != nil { + return pathErr + } g.Go(func() error { - return s.download(gctx, key, filepath.Join(local, store.ExportDir, strings.TrimPrefix(key, exportPrefix))) + return s.download(gctx, key, destination) }) } + for _, file := range manifest.Files { + destination, pathErr := exportPath(exportRoot, file.Path) + if pathErr != nil { + return pathErr + } + for _, chunk := range file.Chunks { + key := exportPrefix + chunk.Object + g.Go(func() error { + return s.downloadExtents(gctx, key, destination, chunk) + }) + } + } if err := g.Wait(); err != nil { return err } + if err := chmodSparseFiles(exportRoot, manifest.Files); err != nil { + return err + } if err := os.WriteFile(filepath.Join(local, store.MetaFile), meta, 0o600); err != nil { return err } @@ -363,6 +465,24 @@ func (s *Store) upload(ctx context.Context, key, path string) error { return s.uploadReader(ctx, key, f, info.Size(), nil) } +func (s *Store) uploadExtents( + ctx context.Context, + key, path string, + size int64, + extents []sparseExtent, +) error { + f, err := os.Open(path) //nolint:gosec // path walked from our own staging dir + if err != nil { + return err + } + defer func() { _ = f.Close() }() + readers := make([]io.Reader, len(extents)) + for i, extent := range extents { + readers[i] = io.NewSectionReader(f, extent.Offset, extent.Size) + } + return s.uploadReader(ctx, key, io.MultiReader(readers...), size, nil) +} + func (s *Store) uploadDigested(ctx context.Context, key, path, digestPath string) (store.DigestFile, error) { f, err := os.Open(path) //nolint:gosec // path walked from our own staging dir if err != nil { @@ -413,6 +533,65 @@ func (s *Store) download(ctx context.Context, key, path string) error { return nil } +func (s *Store) downloadExtents(ctx context.Context, key, path string, chunk sparseChunk) error { + f, err := os.OpenFile(path, os.O_WRONLY, 0) //nolint:gosec // validated path under our temp dir + if err != nil { + return err + } + defer func() { _ = f.Close() }() + writer := newExtentWriterAt(f, chunk.Extents) + if _, err := s.tm.DownloadObject(ctx, &transfermanager.DownloadObjectInput{ + Bucket: &s.bucket, Key: &key, WriterAt: writer, + }); err != nil { + return fmt.Errorf("download sparse chunk %s: %w", key, err) + } + return nil +} + +func (s *Store) readSparseManifest( + ctx context.Context, + exportPrefix string, + keys []string, +) (sparseManifest, bool, error) { + manifestKey := exportPrefix + sparseManifestObject + if !slices.Contains(keys, manifestKey) { + return sparseManifest{}, false, nil + } + out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.bucket, Key: &manifestKey}) + if err != nil { + return sparseManifest{}, false, fmt.Errorf("read sparse manifest: %w", err) + } + defer func() { _ = out.Body.Close() }() + raw, err := io.ReadAll(io.LimitReader(out.Body, sparseManifestMaxBytes+1)) + if err != nil { + return sparseManifest{}, false, fmt.Errorf("read sparse manifest: %w", err) + } + if len(raw) > sparseManifestMaxBytes { + return sparseManifest{}, false, fmt.Errorf("sparse manifest exceeds %d bytes", sparseManifestMaxBytes) + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + var manifest sparseManifest + if err := decoder.Decode(&manifest); err != nil { + return sparseManifest{}, false, fmt.Errorf("decode sparse manifest: %w", err) + } + if decoder.Decode(&struct{}{}) != io.EOF { + return sparseManifest{}, false, fmt.Errorf("decode sparse manifest: trailing data") + } + if err := validateSparseManifest(manifest, exportPrefix, keys); err != nil { + return sparseManifest{}, false, err + } + return manifest, true, nil +} + +func exportPath(root, name string) (string, error) { + clean := pathpkg.Clean(name) + if clean == "." || clean == ".." || pathpkg.IsAbs(clean) || strings.HasPrefix(clean, "../") { + return "", fmt.Errorf("invalid export path %q", name) + } + return filepath.Join(root, filepath.FromSlash(clean)), nil +} + func (s *Store) deleteKeys(ctx context.Context, keys []string) error { // DeleteObjects caps one batch at 1000 keys. for chunk := range slices.Chunk(keys, 1000) { diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index f80f3fc4..e0519514 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -19,7 +19,9 @@ import ( "sync" "testing" + "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/store/storetest" @@ -138,6 +140,256 @@ func TestFetchLegacyExportLayout(t *testing.T) { } } +func TestSparseCheckpointUploadsOnlyAllocatedExtents(t *testing.T) { + const ( + id = "ck_00000000000000cc" + logicalSize = int64(128 << 20) + ) + meta := []byte(`{"id":"` + id + `"}`) + fake := &fakeS3{objects: map[string][]byte{}} + st := newTestStore(t, fake) + st.sparse = true + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage: %v", err) + } + export := filepath.Join(staging, store.ExportDir) + if mkdirErr := os.MkdirAll(export, 0o750); mkdirErr != nil { + t.Fatalf("mkdir export: %v", mkdirErr) + } + disk := filepath.Join(export, "disk.raw") + f, err := os.OpenFile(disk, os.O_CREATE|os.O_RDWR, 0o640) + if err != nil { + t.Fatalf("create sparse disk: %v", err) + } + if truncateErr := f.Truncate(logicalSize); truncateErr != nil { + t.Fatalf("truncate sparse disk: %v", truncateErr) + } + markers := map[int64][]byte{ + 4 << 10: []byte("first-extent"), + logicalSize - (8 << 10): []byte("last-extent"), + } + for offset, marker := range markers { + if _, writeErr := f.WriteAt(marker, offset); writeErr != nil { + t.Fatalf("write sparse marker: %v", writeErr) + } + } + if closeErr := f.Close(); closeErr != nil { + t.Fatalf("close sparse disk: %v", closeErr) + } + if writeErr := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); writeErr != nil { + t.Fatalf("write meta: %v", writeErr) + } + + planned, _, sparse, err := planSparseFile(disk, "disk.raw") + if err != nil { + t.Fatalf("plan sparse file: %v", err) + } + if !sparse { + t.Fatal("test filesystem did not expose sparse extents") + } + if planned.Size != logicalSize || len(planned.Chunks) == 0 { + t.Fatalf("sparse plan = %+v", planned) + } + if publishErr := st.Publish(t.Context(), staging, id); publishErr != nil { + t.Fatalf("Publish: %v", publishErr) + } + + exportPrefix := "ck/" + id + "/" + store.ExportGen(meta) + "/" + fake.mu.Lock() + objects := maps.Clone(fake.objects) + fake.mu.Unlock() + if _, ok := objects[exportPrefix+"disk.raw"]; ok { + t.Fatal("sparse disk was uploaded as one logical-size object") + } + if _, ok := objects[exportPrefix+sparseManifestObject]; !ok { + t.Fatal("sparse manifest was not uploaded") + } + var uploaded int64 + for key, body := range objects { + if strings.HasPrefix(key, exportPrefix+sparseObjectPrefix+"/chunks/") { + uploaded += int64(len(body)) + } + } + if uploaded == 0 || uploaded >= logicalSize/8 { + t.Fatalf("uploaded sparse data = %d, logical size = %d", uploaded, logicalSize) + } + + dir, _, _, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch: %v", err) + } + defer release() + restored := filepath.Join(dir, "disk.raw") + info, err := os.Stat(restored) + if err != nil { + t.Fatalf("stat restored disk: %v", err) + } + if info.Size() != logicalSize || info.Mode().Perm() != 0o640 { + t.Fatalf("restored disk size/mode = %d/%#o, want %d/%#o", info.Size(), info.Mode().Perm(), logicalSize, 0o640) + } + restoredFile, err := os.Open(restored) //nolint:gosec // test path + if err != nil { + t.Fatalf("open restored disk: %v", err) + } + defer func() { _ = restoredFile.Close() }() + for offset, marker := range markers { + got := make([]byte, len(marker)) + if _, err := restoredFile.ReadAt(got, offset); err != nil { + t.Fatalf("read marker at %d: %v", offset, err) + } + if !bytes.Equal(got, marker) { + t.Errorf("marker at %d = %q, want %q", offset, got, marker) + } + } + hole := make([]byte, 4096) + if _, err := restoredFile.ReadAt(hole, logicalSize/2); err != nil { + t.Fatalf("read restored hole: %v", err) + } + if !bytes.Equal(hole, make([]byte, len(hole))) { + t.Fatal("restored hole contains non-zero data") + } +} + +func TestSparseCheckpointEncodingIsOptIn(t *testing.T) { + const id = "ck_00000000000000cf" + meta := []byte(`{"id":"` + id + `"}`) + fake := &fakeS3{objects: map[string][]byte{}} + st := newTestStore(t, fake) + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage: %v", err) + } + export := filepath.Join(staging, store.ExportDir) + if mkdirErr := os.MkdirAll(export, 0o750); mkdirErr != nil { + t.Fatalf("mkdir export: %v", mkdirErr) + } + f, err := os.Create(filepath.Join(export, "disk.raw")) //nolint:gosec // test path + if err != nil { + t.Fatalf("create sparse disk: %v", err) + } + if truncateErr := f.Truncate(1 << 20); truncateErr != nil { + t.Fatalf("truncate sparse disk: %v", truncateErr) + } + if closeErr := f.Close(); closeErr != nil { + t.Fatalf("close sparse disk: %v", closeErr) + } + if writeErr := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); writeErr != nil { + t.Fatalf("write meta: %v", writeErr) + } + if publishErr := st.Publish(t.Context(), staging, id); publishErr != nil { + t.Fatalf("Publish: %v", publishErr) + } + key := "ck/" + id + "/" + store.ExportGen(meta) + "/disk.raw" + fake.mu.Lock() + got := len(fake.objects[key]) + fake.mu.Unlock() + if got != 1<<20 { + t.Fatalf("default dense object size = %d, want %d", got, 1<<20) + } +} + +func TestSparseManifestRejectsMissingChunk(t *testing.T) { + manifest := sparseManifest{ + Version: sparseManifestVersion, + Files: []sparseFile{{ + Path: "disk.raw", Size: 4096, + Chunks: []sparseChunk{{ + Size: 4096, Object: sparseObjectPrefix + "/chunks/file/0000", + Extents: []sparseExtent{{Offset: 0, Size: 4096}}, + }}, + }}, + } + err := validateSparseManifest(manifest, "ck/id/export/", []string{"ck/id/export/" + sparseManifestObject}) + if err == nil || !strings.Contains(err.Error(), "missing sparse chunk") { + t.Fatalf("validate missing chunk = %v", err) + } +} + +func TestAllHoleCheckpointNeedsNoDataObject(t *testing.T) { + const ( + id = "ck_00000000000000cd" + logicalSize = int64(32 << 20) + ) + meta := []byte(`{"id":"` + id + `"}`) + fake := &fakeS3{objects: map[string][]byte{}} + st := newTestStore(t, fake) + st.sparse = true + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage: %v", err) + } + export := filepath.Join(staging, store.ExportDir) + if mkdirErr := os.MkdirAll(export, 0o750); mkdirErr != nil { + t.Fatalf("mkdir export: %v", mkdirErr) + } + disk := filepath.Join(export, "empty.raw") + f, err := os.Create(disk) //nolint:gosec // test path + if err != nil { + t.Fatalf("create sparse disk: %v", err) + } + if truncateErr := f.Truncate(logicalSize); truncateErr != nil { + t.Fatalf("truncate sparse disk: %v", truncateErr) + } + if closeErr := f.Close(); closeErr != nil { + t.Fatalf("close sparse disk: %v", closeErr) + } + if writeErr := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); writeErr != nil { + t.Fatalf("write meta: %v", writeErr) + } + if publishErr := st.Publish(t.Context(), staging, id); publishErr != nil { + t.Fatalf("Publish: %v", publishErr) + } + + exportPrefix := "ck/" + id + "/" + store.ExportGen(meta) + "/" + fake.mu.Lock() + for key := range fake.objects { + if strings.HasPrefix(key, exportPrefix+sparseObjectPrefix+"/chunks/") { + fake.mu.Unlock() + t.Fatalf("all-hole disk uploaded data object %s", key) + } + } + fake.mu.Unlock() + dir, _, _, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch: %v", err) + } + defer release() + info, err := os.Stat(filepath.Join(dir, "empty.raw")) + if err != nil { + t.Fatalf("stat restored all-hole disk: %v", err) + } + if info.Size() != logicalSize { + t.Fatalf("restored all-hole disk size = %d, want %d", info.Size(), logicalSize) + } +} + +func TestExtentWriterMapsPackedObject(t *testing.T) { + path := filepath.Join(t.TempDir(), "disk.raw") + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + t.Fatalf("create target: %v", err) + } + defer func() { _ = f.Close() }() + if truncateErr := f.Truncate(32); truncateErr != nil { + t.Fatalf("truncate target: %v", truncateErr) + } + writer := newExtentWriterAt(f, []sparseExtent{{Offset: 3, Size: 4}, {Offset: 20, Size: 5}}) + if _, writeErr := writer.WriteAt([]byte("cdefghi"), 2); writeErr != nil { + t.Fatalf("cross-extent WriteAt: %v", writeErr) + } + if _, writeErr := writer.WriteAt([]byte("ab"), 0); writeErr != nil { + t.Fatalf("leading WriteAt: %v", writeErr) + } + got, err := os.ReadFile(path) //nolint:gosec // test path + if err != nil { + t.Fatalf("read target: %v", err) + } + if string(got[3:7]) != "abcd" || string(got[20:25]) != "efghi" { + t.Fatalf("mapped extents = %q/%q, want abcd/efghi", got[3:7], got[20:25]) + } +} + func TestDigestReaderIgnoresReadBoundaries(t *testing.T) { data := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize)) data = append(data, 'y') @@ -364,11 +616,87 @@ func TestS3BackendContractRealEndpoint(t *testing.T) { Endpoint: endpoint, Region: "us-east-1", ForcePathStyle: true, + Sparse: true, }, t.TempDir(), store.CheckpointIDRe) if err != nil { t.Fatalf("New: %v", err) } storetest.RunContract(t, st) + t.Run("sparse multipart", func(t *testing.T) { runSparseRealEndpoint(t, st) }) +} + +func runSparseRealEndpoint(t *testing.T, st *Store) { + t.Helper() + const ( + id = "ck_00000000000000ce" + logicalSize = int64(96 << 20) + dataOffset = int64(8 << 20) + ) + meta := []byte(`{"id":"` + id + `"}`) + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage: %v", err) + } + export := filepath.Join(staging, store.ExportDir) + if mkdirErr := os.MkdirAll(export, 0o750); mkdirErr != nil { + t.Fatalf("mkdir export: %v", mkdirErr) + } + disk := filepath.Join(export, "disk.raw") + f, err := os.OpenFile(disk, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + t.Fatalf("create sparse disk: %v", err) + } + if truncateErr := f.Truncate(logicalSize); truncateErr != nil { + t.Fatalf("truncate sparse disk: %v", truncateErr) + } + data := bytes.Repeat([]byte{0x5a}, 20<<20) // exceeds multipart's 16 MiB part size. + if _, writeErr := f.WriteAt(data, dataOffset); writeErr != nil { + t.Fatalf("write sparse extent: %v", writeErr) + } + if closeErr := f.Close(); closeErr != nil { + t.Fatalf("close sparse disk: %v", closeErr) + } + if writeErr := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); writeErr != nil { + t.Fatalf("write meta: %v", writeErr) + } + if publishErr := st.Publish(t.Context(), staging, id); publishErr != nil { + t.Fatalf("Publish: %v", publishErr) + } + t.Cleanup(func() { _ = st.Delete(context.Background(), id) }) + + exportPrefix := st.key(id, store.ExportGen(meta)) + "/" + keys, err := st.list(t.Context(), exportPrefix+sparseObjectPrefix+"/chunks/") + if err != nil { + t.Fatalf("list sparse chunks: %v", err) + } + var uploaded int64 + for _, key := range keys { + head, headErr := st.client.HeadObject(t.Context(), &awss3.HeadObjectInput{Bucket: &st.bucket, Key: &key}) + if headErr != nil { + t.Fatalf("head sparse chunk: %v", headErr) + } + uploaded += aws.ToInt64(head.ContentLength) + } + if uploaded < int64(len(data)) || uploaded >= logicalSize/2 { + t.Fatalf("uploaded sparse bytes = %d, data/logical = %d/%d", uploaded, len(data), logicalSize) + } + dir, _, _, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch: %v", err) + } + defer release() + restored, err := os.Open(filepath.Join(dir, "disk.raw")) //nolint:gosec // test path + if err != nil { + t.Fatalf("open restored disk: %v", err) + } + defer func() { _ = restored.Close() }() + probe := make([]byte, 4096) + if _, err := restored.ReadAt(probe, dataOffset+(4<<20)); err != nil { + t.Fatalf("read restored extent: %v", err) + } + if !bytes.Equal(probe, bytes.Repeat([]byte{0x5a}, len(probe))) { + t.Fatal("restored multipart extent differs") + } } func publishRecord(t *testing.T, st *Store, id string, meta []byte, content string) { diff --git a/sandboxd/store/s3/sparse.go b/sandboxd/store/s3/sparse.go new file mode 100644 index 00000000..c247dc1c --- /dev/null +++ b/sandboxd/store/s3/sparse.go @@ -0,0 +1,282 @@ +package s3 + +import ( + "crypto/sha256" + "errors" + "fmt" + "io" + "os" + pathpkg "path" + "path/filepath" + "sort" + "strings" + + "golang.org/x/sys/unix" +) + +const ( + sparseManifestVersion = 1 + sparseChunkSize = 64 << 20 + sparseObjectPrefix = ".sandboxd-sparse-v1" + sparseManifestObject = sparseObjectPrefix + "/manifest.json" + sparseManifestMaxBytes = 16 << 20 +) + +type sparseManifest struct { + Version int `json:"version"` + Files []sparseFile `json:"files"` +} + +type sparseFile struct { + Path string `json:"path"` + Size int64 `json:"size"` + Mode uint32 `json:"mode"` + Chunks []sparseChunk `json:"chunks"` +} + +type sparseChunk struct { + Size int64 `json:"size"` + Object string `json:"object"` + Extents []sparseExtent `json:"extents"` +} + +type sparseExtent struct { + Offset int64 `json:"offset"` + Size int64 `json:"size"` +} + +type extentWriterAt struct { + file *os.File + extents []sparseExtent + dataOffsets []int64 + size int64 +} + +func newExtentWriterAt(file *os.File, extents []sparseExtent) *extentWriterAt { + w := &extentWriterAt{file: file, extents: extents, dataOffsets: make([]int64, len(extents))} + for i, extent := range extents { + w.dataOffsets[i] = w.size + w.size += extent.Size + } + return w +} + +func (w *extentWriterAt) WriteAt(p []byte, offset int64) (int, error) { + if offset < 0 || offset > w.size || int64(len(p)) > w.size-offset { + return 0, io.ErrShortWrite + } + written := 0 + for len(p) > 0 { + i := sort.Search(len(w.extents), func(i int) bool { + return w.dataOffsets[i]+w.extents[i].Size > offset + }) + if i == len(w.extents) { + return written, io.ErrShortWrite + } + within := offset - w.dataOffsets[i] + n := min(int64(len(p)), w.extents[i].Size-within) + count, err := w.file.WriteAt(p[:n], w.extents[i].Offset+within) + written += count + offset += int64(count) + p = p[count:] + if err != nil { + return written, err + } + if int64(count) != n { + return written, io.ErrShortWrite + } + } + return written, nil +} + +func planSparseFile(path, exportPath string) (sparseFile, []sparseChunk, bool, error) { + f, err := os.Open(path) //nolint:gosec // path walked from our own staging dir + if err != nil { + return sparseFile{}, nil, false, err + } + defer func() { _ = f.Close() }() + info, err := f.Stat() + if err != nil { + return sparseFile{}, nil, false, err + } + if !info.Mode().IsRegular() || info.Size() == 0 { + return sparseFile{}, nil, false, nil + } + extents, supported, err := sparseExtents(f, info.Size()) + if err != nil { + return sparseFile{}, nil, false, fmt.Errorf("scan sparse file %s: %w", exportPath, err) + } + if !supported { + return sparseFile{}, nil, false, nil + } + var dataSize int64 + for _, extent := range extents { + dataSize += extent.size + } + if dataSize >= info.Size() { + return sparseFile{}, nil, false, nil + } + fileID := fmt.Sprintf("%x", sha256.Sum256([]byte(exportPath))) + planned := sparseFile{Path: exportPath, Size: info.Size(), Mode: uint32(info.Mode().Perm())} + planned.Chunks = packExtents(fileID, extents) + return planned, planned.Chunks, true, nil +} + +func packExtents(fileID string, extents []fileExtent) []sparseChunk { + var chunks []sparseChunk + for _, extent := range extents { + for offset := extent.offset; offset < extent.offset+extent.size; { + if len(chunks) == 0 || chunks[len(chunks)-1].Size == sparseChunkSize { + chunks = append(chunks, sparseChunk{ + Object: fmt.Sprintf("%s/chunks/%s/%08x", sparseObjectPrefix, fileID, len(chunks)), + }) + } + chunk := &chunks[len(chunks)-1] + size := min(sparseChunkSize-chunk.Size, extent.offset+extent.size-offset) + chunk.Extents = append(chunk.Extents, sparseExtent{Offset: offset, Size: size}) + chunk.Size += size + offset += size + } + } + return chunks +} + +type fileExtent struct { + offset int64 + size int64 +} + +func sparseExtents(file *os.File, size int64) ([]fileExtent, bool, error) { + var extents []fileExtent + +scan: + for offset := int64(0); offset < size; { + data, err := unix.Seek(int(file.Fd()), offset, unix.SEEK_DATA) + switch { + case errors.Is(err, unix.ENXIO): + break scan + case sparseSeekUnsupported(err): + return nil, false, nil + case err != nil: + return nil, true, err + } + if data < offset { + return nil, false, nil + } + if data >= size { + break + } + hole, err := unix.Seek(int(file.Fd()), data, unix.SEEK_HOLE) + switch { + case errors.Is(err, unix.ENXIO): + hole = size + case sparseSeekUnsupported(err): + return nil, false, nil + case err != nil: + return nil, true, err + } + hole = min(hole, size) + if hole <= data { + return nil, false, nil + } + extents = append(extents, fileExtent{offset: data, size: hole - data}) + offset = hole + } + return extents, true, nil +} + +func sparseSeekUnsupported(err error) bool { + return errors.Is(err, unix.EINVAL) || errors.Is(err, unix.ENOTSUP) || errors.Is(err, unix.ENOSYS) +} + +func validateSparseManifest(manifest sparseManifest, exportPrefix string, keys []string) error { + if manifest.Version != sparseManifestVersion { + return fmt.Errorf("unsupported sparse manifest version %d", manifest.Version) + } + keySet := make(map[string]struct{}, len(keys)) + for _, key := range keys { + keySet[key] = struct{}{} + } + paths := make(map[string]struct{}, len(manifest.Files)) + objects := make(map[string]struct{}) + for _, file := range manifest.Files { + clean := pathpkg.Clean(file.Path) + if clean != file.Path || clean == "." || pathpkg.IsAbs(clean) || strings.HasPrefix(clean, "../") { + return fmt.Errorf("invalid sparse file path %q", file.Path) + } + if _, ok := paths[file.Path]; ok { + return fmt.Errorf("duplicate sparse file path %q", file.Path) + } + paths[file.Path] = struct{}{} + if file.Size < 0 { + return fmt.Errorf("negative sparse file size for %q", file.Path) + } + if _, ok := keySet[exportPrefix+file.Path]; ok { + return fmt.Errorf("sparse file %q also has a dense object", file.Path) + } + var previousEnd int64 + for _, chunk := range file.Chunks { + cleanObject := pathpkg.Clean(chunk.Object) + if cleanObject != chunk.Object || !strings.HasPrefix(cleanObject, sparseObjectPrefix+"/chunks/") { + return fmt.Errorf("invalid sparse chunk object %q", chunk.Object) + } + if _, ok := objects[chunk.Object]; ok { + return fmt.Errorf("duplicate sparse chunk object %q", chunk.Object) + } + objects[chunk.Object] = struct{}{} + if _, ok := keySet[exportPrefix+chunk.Object]; !ok { + return fmt.Errorf("missing sparse chunk object %q", chunk.Object) + } + var chunkSize int64 + for _, extent := range chunk.Extents { + if extent.Offset < previousEnd || extent.Size <= 0 || extent.Offset > file.Size || extent.Size > file.Size-extent.Offset { + return fmt.Errorf("invalid sparse extent for %q at %d", file.Path, extent.Offset) + } + chunkSize += extent.Size + previousEnd = extent.Offset + extent.Size + } + if chunk.Size <= 0 || chunk.Size != chunkSize || chunk.Size > sparseChunkSize { + return fmt.Errorf("invalid sparse chunk size for %q: %d", file.Path, chunk.Size) + } + } + } + return nil +} + +func materializeSparseFiles(root string, files []sparseFile) error { + for _, file := range files { + path, err := exportPath(root, file.Path) + if err != nil { + return err + } + if mkdirErr := os.MkdirAll(filepath.Dir(path), 0o750); mkdirErr != nil { + return mkdirErr + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) //nolint:gosec // validated path + if err != nil { + return err + } + truncateErr := f.Truncate(file.Size) + closeErr := f.Close() + if truncateErr != nil { + return truncateErr + } + if closeErr != nil { + return closeErr + } + } + return nil +} + +func chmodSparseFiles(root string, files []sparseFile) error { + for _, file := range files { + path, err := exportPath(root, file.Path) + if err != nil { + return err + } + if err := os.Chmod(path, os.FileMode(file.Mode)&os.ModePerm); err != nil { + return err + } + } + return nil +}