From 03006f48cc7700b073cec94b5764e85f124767de Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 11 Aug 2026 23:18:16 +0800 Subject: [PATCH 1/7] store: compute template digests while publishing --- sandboxd/pool/archive.go | 2 +- sandboxd/pool/archive_test.go | 2 +- sandboxd/pool/checkpoint.go | 4 +- sandboxd/pool/promote_test.go | 77 +-- sandboxd/pool/template.go | 99 +--- sandboxd/store/digest.go | 65 +++ sandboxd/store/digest_test.go | 88 ++++ sandboxd/store/dir/dir.go | 334 +++++++++--- sandboxd/store/dir/dir_test.go | 289 ++++++++++- sandboxd/store/s3/digest.go | 62 +++ sandboxd/store/s3/s3.go | 229 ++++++-- sandboxd/store/s3/s3_test.go | 717 ++++++++++++++++++-------- sandboxd/store/store.go | 9 +- sandboxd/store/storetest/storetest.go | 90 +++- 14 files changed, 1552 insertions(+), 515 deletions(-) create mode 100644 sandboxd/store/digest.go create mode 100644 sandboxd/store/digest_test.go create mode 100644 sandboxd/store/s3/digest.go diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 686f2680..168f7953 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -173,7 +173,7 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string, m.recDone(ck) } }() - dir, _, release, err := m.ckpts.Fetch(ctx, ck) + dir, _, _, release, err := m.ckpts.Fetch(ctx, ck) if errors.Is(err, store.ErrNotFound) { return "", ErrUnknownSandbox // record disagrees with the store } diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index 8e3074db..1c4d5f98 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -977,7 +977,7 @@ func mustArchive(t *testing.T, m *Manager, sb *types.Sandbox) { func ckExists(t *testing.T, m *Manager, ck string) bool { t.Helper() - _, _, release, err := m.ckpts.Fetch(t.Context(), ck) + _, _, _, release, err := m.ckpts.Fetch(t.Context(), ck) //nolint:dogsled // existence only needs Fetch success if err != nil { return false } diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index af149785..053be4ed 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -177,7 +177,7 @@ func (m *Manager) FetchCheckpoint(ctx context.Context, ckptID string) (string, [ } l := m.recLock(ckptID) l.RLock() - dir, meta, release, err := m.ckpts.Fetch(ctx, ckptID) + dir, meta, _, release, err := m.ckpts.Fetch(ctx, ckptID) if err != nil { l.RUnlock() m.recDone(ckptID) @@ -233,7 +233,7 @@ func (m *Manager) claimLoaded(ctx context.Context, ckpt types.Checkpoint, ttl ti l := m.recLock(ckpt.ID) l.RLock() defer func() { l.RUnlock(); m.recDone(ckpt.ID) }() - dir, _, release, err := m.ckpts.Fetch(ctx, ckpt.ID) + dir, _, _, release, err := m.ckpts.Fetch(ctx, ckpt.ID) if errors.Is(err, store.ErrNotFound) { return nil, ErrUnknownCheckpoint // deleted between the pre-check and the lock } diff --git a/sandboxd/pool/promote_test.go b/sandboxd/pool/promote_test.go index 9363dc07..d0fad7ec 100644 --- a/sandboxd/pool/promote_test.go +++ b/sandboxd/pool/promote_test.go @@ -1,7 +1,7 @@ package pool import ( - "encoding/json" + "bytes" "errors" "os" "path/filepath" @@ -33,7 +33,7 @@ func TestPromoteThenClaimClonesFromTemplate(t *testing.T) { if gotDigest == "" { t.Fatal("Promote returned an empty content digest") } - golden, _, release, err := m.tpls.Fetch(t.Context(), store.TemplateID(key.Hash())) + golden, _, _, release, err := m.tpls.Fetch(t.Context(), store.TemplateID(key.Hash())) if err != nil { t.Fatalf("template export missing: %v", err) } @@ -52,16 +52,12 @@ func TestPromoteThenClaimClonesFromTemplate(t *testing.T) { if child.TemplateDigest != gotDigest { t.Errorf("claim template digest %q, want promoted content digest %q", child.TemplateDigest, gotDigest) } - raw, err := m.tpls.ReadMeta(t.Context(), store.TemplateID(key.Hash())) + meta, err := m.tpls.ReadMeta(t.Context(), store.TemplateID(key.Hash())) if err != nil { t.Fatalf("read template metadata: %v", err) } - var rec templateRecord - if err := json.Unmarshal(raw, &rec); err != nil { - t.Fatalf("decode template metadata: %v", err) - } - if rec.ContentDigest != gotDigest { - t.Errorf("persisted content digest %q, want %q", rec.ContentDigest, gotDigest) + if bytes.Contains(meta, []byte(`"content_digest"`)) { + t.Errorf("template metadata still contains the digest: %s", meta) } } @@ -100,69 +96,6 @@ func TestRepromoteContentDigestTracksExportBytes(t *testing.T) { } } -func TestExportContentDigestCanonicalTree(t *testing.T) { - a, b := t.TempDir(), t.TempDir() - for _, root := range []string{a, b} { - if err := os.Mkdir(filepath.Join(root, "nested"), 0o750); err != nil { - t.Fatalf("mkdir tree: %v", err) - } - } - if err := os.Mkdir(filepath.Join(b, "ignored-empty-dir"), 0o750); err != nil { - t.Fatalf("mkdir empty dir: %v", err) - } - if err := os.WriteFile(filepath.Join(a, "z.bin"), []byte("z"), 0o600); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(a, "nested", "a.bin"), []byte("a"), 0o600); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(b, "nested", "a.bin"), []byte("a"), 0o644); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(b, "z.bin"), []byte("z"), 0o644); err != nil { - t.Fatal(err) - } - if err := os.Chtimes(filepath.Join(b, "z.bin"), time.Unix(1, 0), time.Unix(2, 0)); err != nil { - t.Fatal(err) - } - - da, err := exportContentDigest(a) - if err != nil { - t.Fatalf("digest tree a: %v", err) - } - db, err := exportContentDigest(b) - if err != nil { - t.Fatalf("digest tree b: %v", err) - } - if da != db { - t.Errorf("creation order/mode/mtime changed digest: %q != %q", da, db) - } - const want = "sha256:dfcaba733ae86b82c0760b4d2bb7828a595475e098e99a9769090aeda1390c0a" - if da != want { - t.Errorf("canonical digest %q, want %q", da, want) - } - if writeErr := os.WriteFile(filepath.Join(b, "z.bin"), []byte("changed"), 0o644); writeErr != nil { - t.Fatal(writeErr) - } - changed, err := exportContentDigest(b) - if err != nil { - t.Fatalf("digest changed tree: %v", err) - } - if changed == da { - t.Errorf("content change kept digest %q", changed) - } -} - -func TestExportContentDigestRejectsNonRegularEntry(t *testing.T) { - root := t.TempDir() - if err := os.Symlink("target", filepath.Join(root, "link")); err != nil { - t.Fatal(err) - } - if _, err := exportContentDigest(root); err == nil { - t.Fatal("digest accepted a symbolic link") - } -} - func TestPromoteHibernatedUsesWakeImage(t *testing.T) { eng := newFakeEngine() m := newTestManager(t, eng) diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index 4f262005..35b70527 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -2,18 +2,12 @@ package pool import ( "context" - "crypto/sha256" - "encoding/binary" - "encoding/hex" "encoding/json" "errors" "fmt" - "io" - "io/fs" "os" "path/filepath" "slices" - "strings" "sync" "time" @@ -26,10 +20,9 @@ import ( // from the requested key, never the reverse. Tenant attributes the promote // and scopes deletion; empty means the operator (root). type templateRecord struct { - ID string `json:"id"` - Tenant string `json:"tenant,omitempty"` - ContentDigest string `json:"content_digest,omitempty"` - CreatedAt time.Time `json:"created_at"` + ID string `json:"id"` + Tenant string `json:"tenant,omitempty"` + CreatedAt time.Time `json:"created_at"` } // Promote publishes a claimed sandbox as a template under (template, parent @@ -278,7 +271,7 @@ func (m *Manager) resolveGolden(ctx context.Context, key types.PoolKey) (string, id := store.TemplateID(key.Hash()) l := m.recLock(id) l.RLock() - dir, meta, release, err := m.tpls.Fetch(ctx, id) + dir, meta, digest, release, err := m.tpls.Fetch(ctx, id) if err != nil { l.RUnlock() m.recDone(id) @@ -294,7 +287,7 @@ func (m *Manager) resolveGolden(ctx context.Context, key types.PoolKey) (string, m.recDone(id) return "", "", func() {}, fmt.Errorf("decode template metadata: %w", err) } - return dir, rec.ContentDigest, func() { release(); l.RUnlock(); m.recDone(id) }, nil + return dir, digest, func() { release(); l.RUnlock(); m.recDone(id) }, nil } // publishTemplate exports snap into the store under the key's template id. @@ -311,27 +304,22 @@ func (m *Manager) publishTemplate(ctx context.Context, snap string, key types.Po return m.commitTemplate(ctx, staging, id, tenant) } -// commitTemplate hashes the staged export once, then re-checks the owner and -// swaps meta+generation in, all serialized under the template lock. func (m *Manager) commitTemplate(ctx context.Context, staging, id, tenant string) (string, error) { - digest, err := exportContentDigest(filepath.Join(staging, store.ExportDir)) - if err != nil { - return "", fmt.Errorf("digest template export: %w", err) - } l := m.recLock(id) l.Lock() defer func() { l.Unlock(); m.recDone(id) }() if ownerErr := m.checkTemplateOwner(ctx, id, tenant); ownerErr != nil { return "", ownerErr } - meta, err := json.Marshal(templateRecord{ID: id, Tenant: tenant, ContentDigest: digest, CreatedAt: time.Now()}) + meta, err := json.Marshal(templateRecord{ID: id, Tenant: tenant, CreatedAt: time.Now()}) if err != nil { return "", err } - if err := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil { + if err = os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil { return "", err } - if err := m.tpls.Publish(ctx, staging, id); err != nil { + digest, err := m.tpls.PublishDigested(ctx, staging, id) + if err != nil { return "", fmt.Errorf("publish template: %w", err) } m.tplMu.Lock() @@ -339,72 +327,3 @@ func (m *Manager) commitTemplate(ctx context.Context, staging, id, tenant string m.tplMu.Unlock() return digest, nil } - -// exportContentDigest identifies a snapshot export by a versioned canonical -// regular-file stream: slash-relative path, type, length, and content, globally -// sorted by path. Directories, ownership metadata, modes, and mtimes are excluded. -func exportContentDigest(root string) (string, error) { - type digestEntry struct { - path string - rel string - size int64 - } - var entries []digestEntry - err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr - } - if path == root { - return nil - } - rel, err := filepath.Rel(root, path) - if err != nil { - return err - } - rel = filepath.ToSlash(rel) - info, err := entry.Info() - if err != nil { - return err - } - if entry.IsDir() { - return nil - } - if !info.Mode().IsRegular() { - return fmt.Errorf("unsupported export entry %s (%s)", rel, info.Mode().Type()) - } - entries = append(entries, digestEntry{path: path, rel: rel, size: info.Size()}) - return nil - }) - if err != nil { - return "", err - } - slices.SortFunc(entries, func(a, b digestEntry) int { return strings.Compare(a.rel, b.rel) }) - - h := sha256.New() - _, _ = io.WriteString(h, "sandbox-template-export-v1\x00") - for _, entry := range entries { - var frame [9]byte - frame[0] = 'f' - binary.BigEndian.PutUint64(frame[1:], uint64(len(entry.rel))) - _, _ = h.Write(frame[:]) - _, _ = io.WriteString(h, entry.rel) - binary.BigEndian.PutUint64(frame[:8], uint64(entry.size)) //nolint:gosec // regular file sizes are non-negative - _, _ = h.Write(frame[:8]) - f, err := os.Open(entry.path) //nolint:gosec // path is walked from private export staging - if err != nil { - return "", err - } - n, copyErr := io.Copy(h, f) - closeErr := f.Close() - if copyErr != nil { - return "", copyErr - } - if closeErr != nil { - return "", closeErr - } - if n != entry.size { - return "", fmt.Errorf("export entry %s changed size while hashing", entry.rel) - } - } - return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil -} diff --git a/sandboxd/store/digest.go b/sandboxd/store/digest.go new file mode 100644 index 00000000..15244417 --- /dev/null +++ b/sandboxd/store/digest.go @@ -0,0 +1,65 @@ +package store + +import ( + "cmp" + "crypto/sha256" + "encoding/binary" + "encoding/hex" + "fmt" + "slices" +) + +const ( + // DigestChunkSize is the fixed byte width of every non-final digest chunk. + DigestChunkSize int64 = 16 << 20 + + digestDomain = "sandbox-template-export-v2\x00" +) + +// DigestFile is one regular-file entry in the canonical export manifest. +type DigestFile struct { + Path string // slash-relative path; length and sorting use its raw bytes + Size int64 + Chunks [][sha256.Size]byte // indexed by ascending file offset +} + +// AssembleDigest returns the canonical v2 digest of indexed file chunks. +func AssembleDigest(files []DigestFile) (string, error) { + files = slices.Clone(files) + slices.SortFunc(files, func(a, b DigestFile) int { return cmp.Compare(a.Path, b.Path) }) + + h := sha256.New() + _, _ = h.Write([]byte(digestDomain)) + for i, file := range files { + if file.Path == "" { + return "", fmt.Errorf("digest file path is empty") + } + if i > 0 && files[i-1].Path == file.Path { + return "", fmt.Errorf("duplicate digest path %q", file.Path) + } + if file.Size < 0 { + return "", fmt.Errorf("digest file %q has negative size %d", file.Path, file.Size) + } + var wantChunks int64 + if file.Size > 0 { + wantChunks = (file.Size-1)/DigestChunkSize + 1 + } + if int64(len(file.Chunks)) != wantChunks { + return "", fmt.Errorf("digest file %q has %d chunks, want %d for size %d", file.Path, len(file.Chunks), wantChunks, file.Size) + } + + var frame [8]byte + _, _ = h.Write([]byte{'f'}) + binary.BigEndian.PutUint64(frame[:], uint64(len(file.Path))) + _, _ = h.Write(frame[:]) + _, _ = h.Write([]byte(file.Path)) + binary.BigEndian.PutUint64(frame[:], uint64(file.Size)) + _, _ = h.Write(frame[:]) + binary.BigEndian.PutUint64(frame[:], uint64(len(file.Chunks))) + _, _ = h.Write(frame[:]) + for _, chunk := range file.Chunks { + _, _ = h.Write(chunk[:]) + } + } + return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil +} diff --git a/sandboxd/store/digest_test.go b/sandboxd/store/digest_test.go new file mode 100644 index 00000000..1ab596a4 --- /dev/null +++ b/sandboxd/store/digest_test.go @@ -0,0 +1,88 @@ +package store + +import ( + "bytes" + "crypto/sha256" + "testing" +) + +func TestAssembleDigestVectors(t *testing.T) { + a := sha256.Sum256([]byte("a")) + z := sha256.Sum256([]byte("z")) + tests := []struct { + name string + files []DigestFile + want string + }{ + { + name: "sorted tree", + files: []DigestFile{ + {Path: "z.bin", Size: 1, Chunks: [][sha256.Size]byte{z}}, + {Path: "nested/a.bin", Size: 1, Chunks: [][sha256.Size]byte{a}}, + }, + want: "sha256:ad65b315de70e494c767969e65fc5de65c73846c4ebcdc7051abe08b056637ad", + }, + { + name: "empty file", + files: []DigestFile{{Path: "empty.bin"}}, + want: "sha256:e3a1004a091d5fbaf95fb995f6999355a060a04949e1c3c9d8e886abcd933251", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := AssembleDigest(tt.files) + if err != nil { + t.Fatalf("AssembleDigest: %v", err) + } + if got != tt.want { + t.Errorf("digest %q, want %q", got, tt.want) + } + }) + } +} + +func TestAssembleDigestUsesChunkOffsetOrder(t *testing.T) { + first := sha256.Sum256(bytes.Repeat([]byte("a"), int(DigestChunkSize))) + second := sha256.Sum256([]byte("b")) + completed := []struct { + index int + sum [sha256.Size]byte + }{ + {index: 1, sum: second}, + {index: 0, sum: first}, + } + chunks := make([][sha256.Size]byte, len(completed)) + for _, result := range completed { + chunks[result.index] = result.sum + } + + got, err := AssembleDigest([]DigestFile{{Path: "big.bin", Size: DigestChunkSize + 1, Chunks: chunks}}) + if err != nil { + t.Fatalf("AssembleDigest: %v", err) + } + const want = "sha256:d6aba4c9ea1dfbace830a37704633339e4e485654542569c878e70b7cf088bff" + if got != want { + t.Errorf("digest %q, want %q", got, want) + } +} + +func TestAssembleDigestRejectsInvalidEntries(t *testing.T) { + chunk := sha256.Sum256([]byte("x")) + tests := []struct { + name string + files []DigestFile + }{ + {name: "empty path", files: []DigestFile{{}}}, + {name: "duplicate path", files: []DigestFile{{Path: "x"}, {Path: "x"}}}, + {name: "negative size", files: []DigestFile{{Path: "x", Size: -1}}}, + {name: "missing chunk", files: []DigestFile{{Path: "x", Size: 1}}}, + {name: "extra chunk", files: []DigestFile{{Path: "x", Chunks: [][sha256.Size]byte{chunk}}}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := AssembleDigest(tt.files); err == nil { + t.Fatal("AssembleDigest accepted an invalid entry") + } + }) + } +} diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index 0358c6f5..e3de6957 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -6,6 +6,7 @@ package dir import ( "bytes" "context" + "crypto/sha256" "errors" "fmt" "io" @@ -13,21 +14,37 @@ import ( "os" "path/filepath" "regexp" + "runtime" "slices" "strings" "time" + "golang.org/x/sync/errgroup" + "github.com/cocoonstack/sandbox/sandboxd/store" ) const ( - oldSuffix = ".old" // pre-generation crash artifacts, reclaimed at Delete + oldSuffix = ".old" // pre-generation crash artifacts, reclaimed at Delete + digestPrefix = "digest-" + digestWorkerLimit = 8 + digestReadBufferSize = 128 << 10 // generationGrace bounds the shared-mount race where another node is // still cloning a generation it resolved just before a re-publish. generationGrace = time.Hour ) +type digestSource struct { + path string + digest store.DigestFile +} + +type digestJob struct { + file int + chunk int +} + var _ store.Store = (*Store)(nil) // Store keeps records as //{meta.json,export-/...}: the @@ -53,80 +70,46 @@ func (d *Store) Stage(id string) (string, error) { // Publish makes the generation fresh before its install, refreshes the // outgoing generation at supersession, and commits by renaming meta.json. -func (d *Store) Publish(_ context.Context, staging, id string) error { - metaRaw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // our own staging dir +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) { + digest, err := hashExport(ctx, filepath.Join(staging, store.ExportDir)) if err != nil { - return fmt.Errorf("staging has no %s: %w", store.MetaFile, err) - } - final := filepath.Join(d.root, id) - if err := os.MkdirAll(final, 0o750); err != nil { - return err + return "", fmt.Errorf("digest export: %w", err) } - genDir := filepath.Join(final, store.ExportGen(metaRaw)) - now := time.Now() - genInfo, statErr := os.Stat(genDir) //nolint:gosec // G703: id pinned by the instance idRe before any call - switch { - case errors.Is(statErr, fs.ErrNotExist): - stagedExport := filepath.Join(staging, store.ExportDir) - // Make the generation fresh before a peer can observe it without committed meta. - if err := os.Chtimes(stagedExport, now, now); err != nil { //nolint:gosec // our own staging dir - return err - } - if err := os.Rename(stagedExport, genDir); err != nil { //nolint:gosec // G703: our own staging dir - return err - } - case statErr != nil: - return statErr - default: - currentMeta, readErr := os.ReadFile(filepath.Join(final, store.MetaFile)) //nolint:gosec // fixed path under our root - if readErr == nil && bytes.Equal(currentMeta, metaRaw) { - _ = d.sweepGenerations(id) - return os.RemoveAll(staging) - } - if readErr != nil && !errors.Is(readErr, fs.ErrNotExist) { - return readErr - } - // A sweep may already have selected an expired path for removal. - if time.Since(genInfo.ModTime()) >= generationGrace { - return fmt.Errorf("generation %s expired before commit; retry after sweep", filepath.Base(genDir)) - } + if err := d.publish(ctx, staging, id, digest); err != nil { + return "", fmt.Errorf("publish digested record: %w", err) } - // Age the previous current generation from supersession, not publication. - if err := touchCurrentGeneration(final, now); err != nil { - return err - } - tmp := filepath.Join(final, store.MetaFile+".tmp") - if err := os.WriteFile(tmp, metaRaw, 0o600); err != nil { //nolint:gosec // fixed path under our root - return err - } - if err := os.Rename(tmp, filepath.Join(final, store.MetaFile)); err != nil { - return err - } - _ = d.sweepGenerations(id) // later publishes and startups retry a failed sweep - return os.RemoveAll(staging) + return digest, nil } // Fetch resolves the committed meta and returns its immutable generation // directory; release is a no-op. Records published before per-generation // dirs fall back to the flat export layout. -func (d *Store) Fetch(ctx context.Context, id string) (string, []byte, func(), error) { +func (d *Store) Fetch(ctx context.Context, id string) (string, []byte, string, func(), error) { meta, err := d.ReadMeta(ctx, id) if err != nil { - return "", nil, nil, err + return "", nil, "", nil, err } dir := filepath.Join(d.root, id, store.ExportGen(meta)) if _, statErr := os.Stat(dir); errors.Is(statErr, fs.ErrNotExist) { dir = filepath.Join(d.root, id, store.ExportDir) switch _, legacyErr := os.Stat(dir); { case errors.Is(legacyErr, fs.ErrNotExist): - return "", nil, nil, store.ErrNotFound + return "", nil, "", nil, store.ErrNotFound case legacyErr != nil: - return "", nil, nil, legacyErr + return "", nil, "", nil, legacyErr } } else if statErr != nil { - return "", nil, nil, statErr + return "", nil, "", nil, statErr + } + digest, err := os.ReadFile(filepath.Join(d.root, id, digestName(meta))) //nolint:gosec // id pinned by the instance idRe + if err != nil && !errors.Is(err, fs.ErrNotExist) { + return "", nil, "", nil, fmt.Errorf("read digest: %w", err) } - return dir, meta, func() {}, nil + return dir, meta, string(digest), func() {}, nil } func (d *Store) ReadMeta(_ context.Context, id string) ([]byte, error) { @@ -218,6 +201,67 @@ func (d *Store) SweepGenerations() error { return nil } +func (d *Store) publish(_ context.Context, staging, id, digest string) error { + metaRaw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // our own staging dir + if err != nil { + return fmt.Errorf("staging has no %s: %w", store.MetaFile, err) + } + final := filepath.Join(d.root, id) + if err := os.MkdirAll(final, 0o750); err != nil { + return err + } + genDir := filepath.Join(final, store.ExportGen(metaRaw)) + now := time.Now() + committed := false + genInfo, statErr := os.Stat(genDir) //nolint:gosec // G703: id pinned by the instance idRe before any call + switch { + case errors.Is(statErr, fs.ErrNotExist): + stagedExport := filepath.Join(staging, store.ExportDir) + // Make the generation fresh before a peer can observe it without committed meta. + if err := os.Chtimes(stagedExport, now, now); err != nil { //nolint:gosec // our own staging dir + return err + } + if err := os.Rename(stagedExport, genDir); err != nil { //nolint:gosec // G703: our own staging dir + return err + } + case statErr != nil: + return statErr + default: + currentMeta, readErr := os.ReadFile(filepath.Join(final, store.MetaFile)) //nolint:gosec // fixed path under our root + if readErr == nil && bytes.Equal(currentMeta, metaRaw) { + committed = true + break + } + if readErr != nil && !errors.Is(readErr, fs.ErrNotExist) { + return readErr + } + // A sweep may already have selected an expired path for removal. + if time.Since(genInfo.ModTime()) >= generationGrace { + return fmt.Errorf("generation %s expired before commit; retry after sweep", filepath.Base(genDir)) + } + } + if err := writeGenerationDigest(final, metaRaw, digest); err != nil { + return err + } + if committed { + _ = d.sweepGenerations(id) + return os.RemoveAll(staging) + } + // Age the previous current generation from supersession, not publication. + if err := touchCurrentGeneration(final, now); err != nil { + return err + } + tmp := filepath.Join(final, store.MetaFile+".tmp") + if err := os.WriteFile(tmp, metaRaw, 0o600); err != nil { //nolint:gosec // fixed path under our root + return err + } + if err := os.Rename(tmp, filepath.Join(final, store.MetaFile)); err != nil { + return err + } + _ = d.sweepGenerations(id) // later publishes and startups retry a failed sweep + return os.RemoveAll(staging) +} + // sweepGenerations reclaims only entries whose own install or supersession // time is outside the grace. Without committed meta it never removes the // record directory, so a peer publishing into the same path remains safe. @@ -256,10 +300,11 @@ func (d *Store) sweepGenerations(id string) (err error) { return err } current := store.ExportGen(meta) + currentDigest := digestName(meta) hasCurrent := slices.ContainsFunc(entries, func(e fs.DirEntry) bool { return e.Name() == current }) for _, e := range entries { name := e.Name() - if name == store.MetaFile || name == current { + if name == store.MetaFile || name == current || name == currentDigest { continue } if name == store.ExportDir && !hasCurrent { @@ -281,15 +326,21 @@ func touchCurrentGeneration(final string, now time.Time) error { return err } current := filepath.Join(final, store.ExportGen(meta)) - err = os.Chtimes(current, now, now) //nolint:gosec // current is a hash-derived name under our root - if !errors.Is(err, fs.ErrNotExist) { - return err + exportErr := os.Chtimes(current, now, now) //nolint:gosec // current is a hash-derived name under our root + if errors.Is(exportErr, fs.ErrNotExist) { + exportErr = os.Chtimes(filepath.Join(final, store.ExportDir), now, now) } - err = os.Chtimes(filepath.Join(final, store.ExportDir), now, now) - if errors.Is(err, fs.ErrNotExist) { + if errors.Is(exportErr, fs.ErrNotExist) { + return nil + } + if exportErr != nil { + return exportErr + } + digestErr := os.Chtimes(filepath.Join(final, digestName(meta)), now, now) //nolint:gosec // hash-derived path under our root + if errors.Is(digestErr, fs.ErrNotExist) { return nil } - return err + return digestErr } func removeAgedEntry(final string, entry fs.DirEntry) error { @@ -306,3 +357,162 @@ func removeAgedEntry(final string, entry fs.DirEntry) error { } return os.RemoveAll(filepath.Join(final, entry.Name())) } + +func writeGenerationDigest(final string, meta []byte, digest string) error { + path := filepath.Join(final, digestName(meta)) + tmp := path + ".tmp" + if digest == "" { + return errors.Join(os.RemoveAll(path), os.RemoveAll(tmp)) //nolint:gosec // hash-derived paths under our root + } + if err := os.WriteFile(tmp, []byte(digest), 0o600); err != nil { //nolint:gosec // hash-derived path under our root + return fmt.Errorf("write digest: %w", err) + } + if err := os.Rename(tmp, path); err != nil { //nolint:gosec // hash-derived paths under our root + return fmt.Errorf("commit digest: %w", err) + } + return nil +} + +func digestName(meta []byte) string { + return digestPrefix + store.ExportGenHash(meta) +} + +func hashExport(ctx context.Context, root string) (string, error) { + sources, err := collectDigestSources(root) + if err != nil { + return "", err + } + if err := ctx.Err(); err != nil { + return "", err + } + totalChunks := 0 + for i := range sources { + totalChunks += len(sources[i].digest.Chunks) + } + if totalChunks > 0 { + workers := min(totalChunks, runtime.GOMAXPROCS(0), digestWorkerLimit) + if err := hashChunks(ctx, sources, workers, hashChunk); err != nil { + return "", err + } + } + files := make([]store.DigestFile, len(sources)) + for i := range sources { + info, err := os.Lstat(sources[i].path) + if err != nil { + return "", fmt.Errorf("stat export entry %s: %w", sources[i].digest.Path, err) + } + if !info.Mode().IsRegular() || info.Size() != sources[i].digest.Size { + return "", fmt.Errorf("export entry %s changed while hashing", sources[i].digest.Path) + } + files[i] = sources[i].digest + } + return store.AssembleDigest(files) +} + +func hashChunks( + ctx context.Context, + sources []digestSource, + workers int, + hashFn func(*digestSource, int, []byte) error, +) error { + jobs := make(chan digestJob) + group, groupCtx := errgroup.WithContext(ctx) + for range workers { + group.Go(func() error { + buffer := make([]byte, digestReadBufferSize) + for { + select { + case <-groupCtx.Done(): + return groupCtx.Err() + case job, ok := <-jobs: + if !ok { + return nil + } + if err := hashFn(&sources[job.file], job.chunk, buffer); err != nil { + return err + } + } + } + }) + } + group.Go(func() error { + defer close(jobs) + for i := range sources { + for chunk := range sources[i].digest.Chunks { + select { + case <-groupCtx.Done(): + return groupCtx.Err() + case jobs <- digestJob{file: i, chunk: chunk}: + } + } + } + return nil + }) + return group.Wait() +} + +func collectDigestSources(root string) ([]digestSource, error) { + info, err := os.Lstat(root) + if err != nil { + return nil, err + } + if !info.IsDir() { + return nil, fmt.Errorf("export root is not a directory") + } + var sources []digestSource + err = filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + if path == root || entry.IsDir() { + return nil + } + rel, relErr := filepath.Rel(root, path) + if relErr != nil { + return relErr + } + rel = filepath.ToSlash(rel) + entryInfo, infoErr := entry.Info() + if infoErr != nil { + return infoErr + } + if !entryInfo.Mode().IsRegular() { + return fmt.Errorf("unsupported export entry %s (%s)", rel, entryInfo.Mode().Type()) + } + chunks := entryInfo.Size() / store.DigestChunkSize + if entryInfo.Size()%store.DigestChunkSize != 0 { + chunks++ + } + sources = append(sources, digestSource{ + path: path, + digest: store.DigestFile{ + Path: rel, Size: entryInfo.Size(), Chunks: make([][sha256.Size]byte, int(chunks)), + }, + }) + return nil + }) + return sources, err +} + +func hashChunk(source *digestSource, chunk int, buffer []byte) error { + f, err := os.Open(source.path) //nolint:gosec // path walked from private export staging + if err != nil { + return fmt.Errorf("open export entry %s: %w", source.digest.Path, err) + } + offset := int64(chunk) * store.DigestChunkSize + size := min(store.DigestChunkSize, source.digest.Size-offset) + hash := sha256.New() + n, copyErr := io.CopyBuffer(hash, io.NewSectionReader(f, offset, size), buffer) + closeErr := f.Close() + if copyErr != nil { + return fmt.Errorf("hash export entry %s chunk %d: %w", source.digest.Path, chunk, copyErr) + } + if closeErr != nil { + return fmt.Errorf("close export entry %s: %w", source.digest.Path, closeErr) + } + if n != size { + return fmt.Errorf("export entry %s changed size while hashing", source.digest.Path) + } + copy(source.digest.Chunks[chunk][:], hash.Sum(nil)) + return nil +} diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index 387cd470..0ecbf2f0 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -1,6 +1,8 @@ package dir import ( + "bytes" + "crypto/sha256" "errors" "fmt" "os" @@ -36,24 +38,33 @@ func TestFetchPinsGenerationAcrossStoreInstances(t *testing.T) { if err != nil { t.Fatalf("new writer: %v", err) } - mustPublish(t, writer, id, "first") - backdate(t, filepath.Join(root, id, store.ExportGen([]byte(metaJSON("first"))))) + firstMeta := []byte(metaJSON("first")) + firstDigest := mustPublishDigested(t, writer, id, "first") + firstDigestPath := filepath.Join(root, id, digestName(firstMeta)) + backdate(t, filepath.Join(root, id, store.ExportGen(firstMeta))) + backdate(t, firstDigestPath) backdate(t, filepath.Join(root, id, store.MetaFile)) - dir, meta, release, err := reader.Fetch(t.Context(), id) + dir, meta, digest, release, err := reader.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch first: %v", err) } if string(meta) != metaJSON("first") { t.Fatalf("fetch meta %q, want first generation", meta) } - mustPublish(t, writer, id, "second") + if digest != firstDigest { + t.Fatalf("fetch digest %q, want first generation %q", digest, firstDigest) + } + secondDigest := mustPublishDigested(t, writer, id, "second") if _, statErr := os.Stat(filepath.Join(dir, "disk.img")); statErr != nil { t.Fatalf("first generation disturbed by re-publish: %v", statErr) } + if _, statErr := os.Stat(firstDigestPath); statErr != nil { + t.Fatalf("first generation digest disturbed by re-publish: %v", statErr) + } release() - _, meta, release, err = reader.Fetch(t.Context(), id) + _, meta, digest, release, err = reader.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch second: %v", err) } @@ -61,6 +72,232 @@ func TestFetchPinsGenerationAcrossStoreInstances(t *testing.T) { if string(meta) != metaJSON("second") { t.Fatalf("fetch meta %q, want second generation", meta) } + if digest != secondDigest || digest == firstDigest { + t.Fatalf("fetch digest %q, want second generation %q", digest, secondDigest) + } +} + +func TestPlainPublishReplacesDigestWithEmpty(t *testing.T) { + const id = "ck_00000000000000aa" + root := t.TempDir() + st, err := New(root, store.CheckpointIDRe) + if err != nil { + t.Fatalf("New: %v", err) + } + firstMeta := []byte(metaJSON("digested")) + firstDigest := mustPublishDigested(t, st, id, "digested") + firstDigestPath := filepath.Join(root, id, digestName(firstMeta)) + mustPublish(t, st, id, "plain") + + _, meta, digest, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch: %v", err) + } + defer release() + if string(meta) != metaJSON("plain") || digest != "" { + t.Fatalf("plain replacement meta/digest = %q/%q, want plain/empty", meta, digest) + } + if firstDigest == "" { + t.Fatal("digested generation returned an empty digest") + } + if _, statErr := os.Stat(firstDigestPath); statErr != nil { + t.Fatalf("superseded digest removed inside the grace: %v", statErr) + } +} + +func TestPublishDigestedChunkBoundaries(t *testing.T) { + const id = "ck_00000000000000aa" + content := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize)+1) + content[len(content)-1] = 'y' + tests := []struct { + name string + size int64 + want string + }{ + {name: "below", size: store.DigestChunkSize - 1, want: "sha256:9463427f8626e0b65863b401d333d32ad8e0b11efb34db7f502798aea753fc7e"}, + {name: "exact", size: store.DigestChunkSize, want: "sha256:20ef154fac2fa87246572b15e008adfef78fc47ce93e479c367112ea750006c6"}, + {name: "above", size: store.DigestChunkSize + 1, want: "sha256:da8dd289e7d66408bd7d202642f180f2c22bc508cf0fe1d6e9f0e0b7967f6a5e"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st, err := New(t.TempDir(), store.CheckpointIDRe) + if err != nil { + t.Fatalf("New: %v", err) + } + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage: %v", err) + } + export := filepath.Join(staging, store.ExportDir) + if err = os.MkdirAll(export, 0o750); err != nil { + t.Fatalf("mkdir export: %v", err) + } + if err = os.WriteFile(filepath.Join(export, "boundary.bin"), content[:int(tt.size)], 0o600); err != nil { + t.Fatalf("write export: %v", err) + } + 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) + if err != nil { + t.Fatalf("PublishDigested: %v", err) + } + if digest != tt.want { + t.Errorf("digest = %q, want %q", digest, tt.want) + } + _, _, fetched, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch: %v", err) + } + defer release() + if fetched != tt.want { + t.Errorf("fetched digest = %q, want %q", fetched, tt.want) + } + }) + } +} + +func TestDigestWorkersStoreChunksByOffset(t *testing.T) { + data := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize)) + data = append(data, 'y') + path := filepath.Join(t.TempDir(), "boundary.bin") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatalf("write export: %v", err) + } + sources := []digestSource{{ + path: path, + digest: store.DigestFile{ + Path: "boundary.bin", Size: int64(len(data)), Chunks: make([][sha256.Size]byte, 2), + }, + }} + secondDone := make(chan struct{}) + completed := make(chan int, 2) + hashFn := func(source *digestSource, chunk int, buffer []byte) error { + if chunk == 0 { + <-secondDone + } + err := hashChunk(source, chunk, buffer) + completed <- chunk + if chunk == 1 { + close(secondDone) + } + return err + } + if err := hashChunks(t.Context(), sources, 2, hashFn); err != nil { + t.Fatalf("hash chunks: %v", err) + } + if first, second := <-completed, <-completed; first != 1 || second != 0 { + t.Fatalf("completion order = %d, %d, want 1, 0", first, second) + } + digest, err := store.AssembleDigest([]store.DigestFile{sources[0].digest}) + if err != nil { + t.Fatalf("assemble digest: %v", err) + } + const want = "sha256:da8dd289e7d66408bd7d202642f180f2c22bc508cf0fe1d6e9f0e0b7967f6a5e" + if digest != want { + t.Errorf("digest = %q, want %q", digest, want) + } +} + +func TestPublishDigestedFailureAndFreshRetry(t *testing.T) { + const id = "ck_00000000000000aa" + root := t.TempDir() + st, err := New(root, store.CheckpointIDRe) + if err != nil { + t.Fatalf("New: %v", err) + } + firstDigest := mustPublishDigested(t, st, id, "first") + + failed, err := st.Stage(id) + if err != nil { + t.Fatalf("stage failed publish: %v", err) + } + seedRecord(t, failed, "second") + secondMeta, err := os.ReadFile(filepath.Join(failed, store.MetaFile)) + if err != nil { + t.Fatalf("read second meta: %v", err) + } + blocker := filepath.Join(root, id, digestName(secondMeta)) + 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 != "" { + t.Fatalf("PublishDigested with blocked sidecar = %q, %v, want empty/error", digest, publishErr) + } + dir, meta, digest, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch previous generation: %v", err) + } + content, readErr := os.ReadFile(filepath.Join(dir, "disk.img")) + release() + if string(meta) != metaJSON("first") || digest != firstDigest { + t.Errorf("previous generation meta/digest = %q/%q, want first/%q", meta, digest, firstDigest) + } + if readErr != nil || string(content) != "first" { + t.Errorf("previous generation content = %q, %v, want first", content, readErr) + } + + if err = os.RemoveAll(blocker); err != nil { + t.Fatalf("remove digest blocker: %v", err) + } + if err = os.RemoveAll(failed); err != nil { + t.Fatalf("remove failed staging: %v", err) + } + secondDigest := mustPublishDigested(t, st, id, "second") + dir, meta, digest, release, err = st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch replacement generation: %v", err) + } + defer release() + content, readErr = os.ReadFile(filepath.Join(dir, "disk.img")) + if string(meta) != metaJSON("second") || digest != secondDigest || digest == firstDigest { + t.Errorf("replacement generation meta/digest = %q/%q, want second/%q", meta, digest, secondDigest) + } + if readErr != nil || string(content) != "second" { + t.Errorf("replacement generation content = %q, %v, want second", content, readErr) + } +} + +func TestSweepGenerationsPairsCurrentAndSupersededDigests(t *testing.T) { + const id = "ck_00000000000000aa" + root := t.TempDir() + st, err := New(root, store.CheckpointIDRe) + if err != nil { + t.Fatalf("New: %v", err) + } + firstMeta := []byte(metaJSON("first")) + firstGen := filepath.Join(root, id, store.ExportGen(firstMeta)) + firstSidecar := filepath.Join(root, id, digestName(firstMeta)) + mustPublishDigested(t, st, id, "first") + secondMeta := []byte(metaJSON("second")) + secondGen := filepath.Join(root, id, store.ExportGen(secondMeta)) + secondSidecar := filepath.Join(root, id, digestName(secondMeta)) + secondDigest := mustPublishDigested(t, st, id, "second") + for _, path := range []string{firstGen, firstSidecar, secondGen, secondSidecar} { + backdate(t, path) + } + + if err = st.SweepGenerations(); err != nil { + t.Fatalf("SweepGenerations: %v", err) + } + for _, path := range []string{firstGen, firstSidecar} { + if _, statErr := os.Stat(path); !os.IsNotExist(statErr) { + t.Errorf("superseded entry %s survived sweep: %v", filepath.Base(path), statErr) + } + } + for _, path := range []string{secondGen, secondSidecar} { + if _, statErr := os.Stat(path); statErr != nil { + t.Errorf("current entry %s was swept: %v", filepath.Base(path), statErr) + } + } + _, meta, digest, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch current generation: %v", err) + } + defer release() + if string(meta) != metaJSON("second") || digest != secondDigest { + t.Fatalf("current meta/digest = %q/%q, want second/%q", meta, digest, secondDigest) + } } // TestPublishRetriesAfterExpiredInstall: an expired uncommitted generation is @@ -99,7 +336,7 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) { mustPublish(t, st, id, "same") backdate(t, gen) mustPublish(t, st, id, "same") - dir, _, release, err := st.Fetch(t.Context(), id) + dir, _, _, release, err := st.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch after retried publish: %v", err) } @@ -121,23 +358,23 @@ func TestFetchLegacyFlatLayout(t *testing.T) { } seedRecord(t, filepath.Join(root, id), "legacy") - dir, meta, release, err := st.Fetch(t.Context(), id) + dir, meta, digest, release, err := st.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch legacy: %v", err) } release() - if string(meta) != metaJSON("legacy") || dir != filepath.Join(root, id, store.ExportDir) { - t.Fatalf("legacy fetch = %q %q, want the flat export dir", dir, meta) + if string(meta) != metaJSON("legacy") || digest != "" || dir != filepath.Join(root, id, store.ExportDir) { + t.Fatalf("legacy fetch = %q %q %q, want the flat export dir without digest", dir, meta, digest) } mustPublish(t, st, id, "modern") - dir, meta, release, err = st.Fetch(t.Context(), id) + dir, meta, digest, release, err = st.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch after re-publish: %v", err) } defer release() - if string(meta) != metaJSON("modern") || dir == filepath.Join(root, id, store.ExportDir) { - t.Fatalf("re-published fetch = %q %q, want a generation dir", dir, meta) + if string(meta) != metaJSON("modern") || digest != "" || dir == filepath.Join(root, id, store.ExportDir) { + t.Fatalf("re-published fetch = %q %q %q, want a generation dir without digest", dir, meta, digest) } } @@ -168,7 +405,7 @@ func TestPublishSweepsGenerationsBySupersessionAge(t *testing.T) { if _, statErr := os.Stat(gen2); statErr != nil { t.Fatalf("generation inside its supersession grace was reclaimed: %v", statErr) } - dir, meta, release, err := st.Fetch(t.Context(), id) + dir, meta, _, release, err := st.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch after sweep: %v", err) } @@ -212,7 +449,7 @@ func TestSweepSparesPublishingGeneration(t *testing.T) { if renameErr := os.Rename(tmp, filepath.Join(root, id, store.MetaFile)); renameErr != nil { t.Fatalf("commit meta: %v", renameErr) } - dir, gotMeta, release, err := writer.Fetch(t.Context(), id) + dir, gotMeta, digest, release, err := writer.Fetch(t.Context(), id) if err != nil { t.Fatalf("fetch committed generation: %v", err) } @@ -220,6 +457,9 @@ func TestSweepSparesPublishingGeneration(t *testing.T) { if string(gotMeta) != metaJSON("new") { t.Fatalf("fetch meta %q, want new generation", gotMeta) } + if digest != "" { + t.Fatalf("fetch digest %q, want empty", digest) + } if got, readErr := os.ReadFile(filepath.Join(dir, "disk.img")); readErr != nil || string(got) != "new" { t.Fatalf("fetch export %q, %v, want new generation", got, readErr) } @@ -240,7 +480,7 @@ func TestSweepSparesLegacyFallback(t *testing.T) { if err := st.SweepStaging(); err != nil { t.Fatalf("sweep: %v", err) } - if _, _, release, err := st.Fetch(t.Context(), id); err != nil { + if _, _, _, release, err := st.Fetch(t.Context(), id); err != nil { t.Fatalf("legacy record swept while still current: %v", err) } else { release() @@ -318,7 +558,7 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) { t.Fatalf("remove commit blocker: %v", removeErr) } mustPublish(t, writer, publishID, "fresh") - dir, gotMeta, release, err := writer.Fetch(t.Context(), publishID) + dir, gotMeta, digest, release, err := writer.Fetch(t.Context(), publishID) if err != nil { t.Fatalf("fetch retried publish: %v", err) } @@ -326,6 +566,9 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) { if string(gotMeta) != metaJSON("fresh") { t.Fatalf("fetch meta %q, want fresh", gotMeta) } + if digest != "" { + t.Fatalf("fetch digest %q, want empty", digest) + } if got, readErr := os.ReadFile(filepath.Join(dir, "disk.img")); readErr != nil || string(got) != "fresh" { t.Fatalf("fetch export %q, %v, want fresh", got, readErr) } @@ -470,6 +713,20 @@ func mustPublish(t *testing.T, st *Store, id, metaID string) { } } +func mustPublishDigested(t *testing.T, st *Store, id, metaID string) string { + t.Helper() + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("stage %s: %v", metaID, err) + } + seedRecord(t, staging, metaID) + digest, err := st.PublishDigested(t.Context(), staging, id) + if err != nil { + t.Fatalf("publish digested %s: %v", metaID, err) + } + return digest +} + func seedRecord(t *testing.T, dir, metaID string) { t.Helper() if err := os.MkdirAll(filepath.Join(dir, store.ExportDir), 0o750); err != nil { diff --git a/sandboxd/store/s3/digest.go b/sandboxd/store/s3/digest.go new file mode 100644 index 00000000..5d6f7566 --- /dev/null +++ b/sandboxd/store/s3/digest.go @@ -0,0 +1,62 @@ +package s3 + +import ( + "crypto/sha256" + "fmt" + "hash" + "io" + + "github.com/cocoonstack/sandbox/sandboxd/store" +) + +type digestReader struct { + body io.Reader + chunkHash hash.Hash + path string + size int64 + read int64 + chunkBytes int64 + chunks [][sha256.Size]byte + finished bool +} + +func newDigestReader(body io.Reader, path string, size int64) *digestReader { + return &digestReader{body: body, chunkHash: sha256.New(), path: path, size: size} +} + +func (r *digestReader) Read(p []byte) (int, error) { + n, err := r.body.Read(p) + r.read += int64(n) + for offset := 0; offset < n; { + remaining := int(store.DigestChunkSize - r.chunkBytes) + written := min(n-offset, remaining) + _, _ = r.chunkHash.Write(p[offset : offset+written]) + r.chunkBytes += int64(written) + offset += written + if r.chunkBytes == store.DigestChunkSize { + r.finishChunk() + } + } + return n, err +} + +func (r *digestReader) Digest() (store.DigestFile, error) { + if !r.finished { + if r.chunkBytes > 0 { + r.finishChunk() + } + r.finished = true + } + if r.read != r.size { + return store.DigestFile{}, fmt.Errorf("read %s: got %d bytes, want %d", r.path, r.read, r.size) + } + return store.DigestFile{Path: r.path, Size: r.size, Chunks: r.chunks}, nil +} + +func (r *digestReader) finishChunk() { + var sum [sha256.Size]byte + r.chunkHash.Sum(sum[:0]) + r.chunks = append(r.chunks, sum) + r.chunkHash.Reset() + r.chunkBytes = 0 +} diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 377099ae..b6dbfacd 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -28,6 +28,11 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store" ) +const ( + digestMetadataKey = "content-digest" + uploadPartSizeBytes = 16 << 20 +) + // Config selects the bucket and, for MinIO/R2-style endpoints, the // addressing mode. Credentials come from the standard AWS chain (env, // IAM role, web identity) — never from sandboxd's config file. @@ -39,13 +44,24 @@ type Config struct { ForcePathStyle bool `json:"force_path_style,omitempty"` } +type publishFile struct { + path string + key string + digestPath string +} + +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) +} + var _ store.Store = (*Store)(nil) // Store stages locally under stagingRoot and publishes to the bucket; // idRe names the instance's id namespace within the shared prefix. type Store struct { client *awss3.Client - tm *transfermanager.Client + tm transferManager bucket string prefix string staging string @@ -79,7 +95,7 @@ func New(ctx context.Context, cfg Config, stagingRoot string, idRe *regexp.Regex // Snapshot exports are hundreds of MB: multipart + concurrency keep a // publish/fetch bandwidth-bound instead of latency-bound. tm := transfermanager.New(client, func(o *transfermanager.Options) { - o.PartSizeBytes = 16 << 20 + o.PartSizeBytes = uploadPartSizeBytes o.Concurrency = 8 }) return &Store{client: client, tm: tm, bucket: cfg.Bucket, prefix: cfg.Prefix, staging: stagingRoot, idRe: idRe}, nil @@ -95,39 +111,12 @@ func (s *Store) Stage(id string) (string, error) { // overwrites keys a concurrent Fetch of the old generation is reading — // mixed-generation downloads become impossible, not just unlikely. func (s *Store) Publish(ctx context.Context, staging, id string) error { - metaRaw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // our own staging dir - if err != nil { - return fmt.Errorf("staging has no %s: %w", store.MetaFile, err) - } - gen := store.ExportGen(metaRaw) - g, gctx := errgroup.WithContext(ctx) - g.SetLimit(4) // files in parallel; each already multiparts internally - err = filepath.WalkDir(staging, func(path string, d os.DirEntry, err error) error { - if err != nil || d.IsDir() { - return err - } - rel, err := filepath.Rel(staging, path) - if err != nil { - return err - } - if rel == store.MetaFile { - return nil - } - key := s.key(id, gen+strings.TrimPrefix(rel, store.ExportDir)) - g.Go(func() error { return s.upload(gctx, key, path) }) - return nil - }) - if err != nil { - return err - } - if err = g.Wait(); err != nil { - return err - } - if err = s.uploadReader(ctx, s.key(id, store.MetaFile), bytes.NewReader(metaRaw)); err != nil { - return err - } - // Keep old generations because another node may have selected the previous meta; Delete reclaims them. - return os.RemoveAll(staging) + _, err := s.publish(ctx, staging, id, false) + return err +} + +func (s *Store) PublishDigested(ctx context.Context, staging, id string) (string, error) { + return s.publish(ctx, staging, id, true) } // Fetch materializes the export into a local cache generation keyed by the @@ -135,38 +124,28 @@ func (s *Store) Publish(ctx context.Context, staging, id string) error { // a new generation never disturbs a directory an in-flight clone is reading // (old generations are reaped at Delete and startup). Concurrent misses share // one download; release is a no-op; a missing id is ErrNotFound. -func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, func(), error) { - meta, err := s.ReadMeta(ctx, id) +func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, func(), error) { + meta, digest, err := s.readMeta(ctx, id) if err != nil { - return "", nil, nil, err + return "", nil, "", nil, err } gen := filepath.Join(s.staging, "cache", id, store.ExportGenHash(meta)) export := filepath.Join(gen, store.ExportDir) if _, statErr := os.Stat(export); statErr == nil { - return export, meta, func() {}, nil + return export, meta, digest, func() {}, nil } _, err, _ = s.fetches.Do(gen, func() (any, error) { return nil, s.populate(ctx, id, meta, gen) }) if err != nil { - return "", nil, nil, err + return "", nil, "", nil, err } - return export, meta, func() {}, nil + return export, meta, digest, func() {}, nil } func (s *Store) ReadMeta(ctx context.Context, id string) ([]byte, error) { - out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{ - Bucket: &s.bucket, Key: aws.String(s.key(id, store.MetaFile)), - }) - if err != nil { - var apiErr smithy.APIError - if errors.As(err, &apiErr) && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") { //nolint:goconst // AWS API error codes, compared as literals - return nil, store.ErrNotFound - } - return nil, fmt.Errorf("record %s: %w", id, err) - } - defer func() { _ = out.Body.Close() }() - return io.ReadAll(out.Body) + meta, _, err := s.readMeta(ctx, id) + return meta, err } func (s *Store) Metas(ctx context.Context) ([][]byte, error) { @@ -244,6 +223,113 @@ func (s *Store) SweepStaging() error { // bucket lifecycle policy handles invisible upload orphans. func (s *Store) SweepGenerations() error { return nil } +func (s *Store) readMeta(ctx context.Context, id string) ([]byte, string, error) { + out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{ + Bucket: &s.bucket, Key: aws.String(s.key(id, store.MetaFile)), + }) + if err != nil { + var apiErr smithy.APIError + if errors.As(err, &apiErr) && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") { //nolint:goconst // AWS API error codes, compared as literals + return nil, "", store.ErrNotFound + } + return nil, "", fmt.Errorf("record %s: %w", id, err) + } + defer func() { _ = out.Body.Close() }() + meta, err := io.ReadAll(out.Body) + if err != nil { + return nil, "", err + } + return meta, out.Metadata[digestMetadataKey], nil +} + +func (s *Store) publish(ctx context.Context, staging, id string, digested bool) (string, error) { + metaRaw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // our own staging dir + if err != nil { + return "", fmt.Errorf("staging has no %s: %w", store.MetaFile, err) + } + files, err := s.publishFiles(staging, id, store.ExportGen(metaRaw), digested) + if err != nil { + return "", err + } + digestFiles := make([]store.DigestFile, len(files)) + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(4) // files in parallel; each already multiparts internally + for i, file := range files { + g.Go(func() error { + if !digested { + return s.upload(gctx, file.key, file.path) + } + fileDigest, digestErr := s.uploadDigested(gctx, file.key, file.path, file.digestPath) + if digestErr == nil { + digestFiles[i] = fileDigest + } + return digestErr + }) + } + if err = g.Wait(); err != nil { + return "", err + } + digest := "" + var metadata map[string]string + if digested { + digest, err = store.AssembleDigest(digestFiles) + if err != nil { + return "", err + } + metadata = map[string]string{digestMetadataKey: digest} + } + if err = s.uploadReader(ctx, s.key(id, store.MetaFile), bytes.NewReader(metaRaw), int64(len(metaRaw)), metadata); err != nil { + return "", err + } + // Keep old generations because another node may have selected the previous meta; Delete reclaims them. + if err := os.RemoveAll(staging); err != nil { + return "", err + } + return digest, nil +} + +func (s *Store) publishFiles(staging, id, gen string, digested bool) ([]publishFile, error) { + root := staging + if digested { + root = filepath.Join(staging, store.ExportDir) + } + var files []publishFile + err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + if err != nil || entry.IsDir() { + return err + } + rel, err := filepath.Rel(root, path) + if err != nil { + return err + } + if !digested { + if rel == store.MetaFile { + return nil + } + files = append(files, publishFile{ + path: path, + key: s.key(id, gen+strings.TrimPrefix(rel, store.ExportDir)), + }) + return nil + } + info, err := entry.Info() + 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) + } + files = append(files, publishFile{ + path: path, + key: s.key(id, gen+"/"+rel), + digestPath: rel, + }) + return nil + }) + return files, err +} + // populate downloads one cache generation and installs it atomically. func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string) error { if _, err := os.Stat(gen); err == nil { @@ -301,11 +387,42 @@ func (s *Store) upload(ctx context.Context, key, path string) error { return err } defer func() { _ = f.Close() }() - return s.uploadReader(ctx, key, f) + info, err := f.Stat() + if err != nil { + return err + } + return s.uploadReader(ctx, key, f, info.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 { + return store.DigestFile{}, err + } + defer func() { _ = f.Close() }() + info, err := f.Stat() + if err != nil { + return store.DigestFile{}, err + } + if !info.Mode().IsRegular() { + return store.DigestFile{}, fmt.Errorf("digest export %s is not a regular file", digestPath) + } + reader := newDigestReader(f, digestPath, info.Size()) + if err := s.uploadReader(ctx, key, reader, info.Size(), nil); err != nil { + return store.DigestFile{}, err + } + return reader.Digest() } -func (s *Store) uploadReader(ctx context.Context, key string, body io.Reader) error { - if _, err := s.tm.UploadObject(ctx, &transfermanager.UploadObjectInput{Bucket: &s.bucket, Key: &key, Body: body}); err != nil { +func (s *Store) uploadReader(ctx context.Context, key string, body io.Reader, size int64, metadata map[string]string) error { + input := &transfermanager.UploadObjectInput{ + Bucket: &s.bucket, + Key: &key, + Body: body, + ContentLength: aws.Int64(size), + Metadata: metadata, + } + if _, err := s.tm.UploadObject(ctx, input); err != nil { return fmt.Errorf("upload %s: %w", key, err) } return nil diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index 1f435a53..9c70fd74 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -1,10 +1,14 @@ package s3 import ( + "bytes" "cmp" + "context" + "crypto/sha256" "encoding/xml" "fmt" "io" + "maps" "net/http" "net/http/httptest" "os" @@ -15,32 +19,15 @@ import ( "sync" "testing" + "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager" + "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/store/storetest" ) func TestS3BackendContract(t *testing.T) { fake := &fakeS3{objects: map[string][]byte{}} - ts := httptest.NewServer(fake) - t.Cleanup(ts.Close) - - t.Setenv("AWS_ACCESS_KEY_ID", "test") - t.Setenv("AWS_SECRET_ACCESS_KEY", "test") - // Seekable bodies + when_required keep PutObject payloads plain (no - // aws-chunked checksum trailers the fake would have to decode). - t.Setenv("AWS_REQUEST_CHECKSUM_CALCULATION", "when_required") - t.Setenv("AWS_RESPONSE_CHECKSUM_VALIDATION", "when_required") - - st, err := New(t.Context(), Config{ - Bucket: "testbucket", - Prefix: "ck/", - Endpoint: ts.URL, - Region: "us-east-1", - ForcePathStyle: true, - }, t.TempDir(), store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } + st := newTestStore(t, fake) storetest.RunContract(t, st) } @@ -53,24 +40,7 @@ func TestDeleteRetryConverges(t *testing.T) { "ck/" + id + "/" + store.MetaFile: []byte(`{"id":"` + id + `"}`), "ck/" + id + "/" + store.ExportDir + "/disk.img": []byte("bytes"), }} - ts := httptest.NewServer(fake) - t.Cleanup(ts.Close) - - t.Setenv("AWS_ACCESS_KEY_ID", "test") - t.Setenv("AWS_SECRET_ACCESS_KEY", "test") - t.Setenv("AWS_REQUEST_CHECKSUM_CALCULATION", "when_required") - t.Setenv("AWS_RESPONSE_CHECKSUM_VALIDATION", "when_required") - - st, err := New(t.Context(), Config{ - Bucket: "testbucket", - Prefix: "ck/", - Endpoint: ts.URL, - Region: "us-east-1", - ForcePathStyle: true, - }, t.TempDir(), store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } + st := newTestStore(t, fake) if err := st.Delete(t.Context(), id); err != nil { t.Fatalf("first delete: %v", err) } @@ -89,11 +59,6 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) { "ck/" + id + "/export-first/disk.img", "ck/" + id + "/export-second/disk.img", } - t.Setenv("AWS_ACCESS_KEY_ID", "test") - t.Setenv("AWS_SECRET_ACCESS_KEY", "test") - t.Setenv("AWS_REQUEST_CHECKSUM_CALCULATION", "when_required") - t.Setenv("AWS_RESPONSE_CHECKSUM_VALIDATION", "when_required") - for _, tc := range []struct { name string failList bool @@ -112,17 +77,9 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) { failList: tc.failList, failDelete: tc.failDelete, } - ts := httptest.NewServer(fake) - t.Cleanup(ts.Close) - st, err := New(t.Context(), Config{ - Bucket: "testbucket", Prefix: "ck/", Endpoint: ts.URL, - Region: "us-east-1", ForcePathStyle: true, - }, t.TempDir(), store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } + st := newTestStore(t, fake) - if err = st.Delete(t.Context(), id); err == nil { + if err := st.Delete(t.Context(), id); err == nil { t.Fatal("first Delete succeeded with an injected failure") } fake.mu.Lock() @@ -141,7 +98,7 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) { fake.deleteBatches = nil fake.mu.Unlock() - if err = st.Delete(t.Context(), id); err != nil { + if err := st.Delete(t.Context(), id); err != nil { t.Fatalf("second Delete: %v", err) } fake.mu.Lock() @@ -174,25 +131,8 @@ func TestFetchLegacyExportLayout(t *testing.T) { "ck/" + id + "/" + store.MetaFile: []byte(`{"id":"` + id + `"}`), "ck/" + id + "/" + store.ExportDir + "/disk.img": []byte("legacy-bytes"), }} - ts := httptest.NewServer(fake) - t.Cleanup(ts.Close) - - t.Setenv("AWS_ACCESS_KEY_ID", "test") - t.Setenv("AWS_SECRET_ACCESS_KEY", "test") - t.Setenv("AWS_REQUEST_CHECKSUM_CALCULATION", "when_required") - t.Setenv("AWS_RESPONSE_CHECKSUM_VALIDATION", "when_required") - - st, err := New(t.Context(), Config{ - Bucket: "testbucket", - Prefix: "ck/", - Endpoint: ts.URL, - Region: "us-east-1", - ForcePathStyle: true, - }, t.TempDir(), store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } - dir, _, release, err := st.Fetch(t.Context(), id) + st := newTestStore(t, fake) + dir, _, _, release, err := st.Fetch(t.Context(), id) if err != nil { t.Fatalf("Fetch: %v", err) } @@ -203,28 +143,196 @@ func TestFetchLegacyExportLayout(t *testing.T) { } } -func TestRepublishRetainsGenerationSelectedByAnotherStore(t *testing.T) { - const id = "ck_00000000000000aa" +func TestDigestReaderIgnoresReadBoundaries(t *testing.T) { + data := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize)) + data = append(data, 'y') + source := &patternReader{data: data, limits: []int{1, 31, 64 << 10, 7, 79 << 10}} + reader := newDigestReader(source, "boundary.bin", int64(len(data))) + if _, ok := any(reader).(io.Seeker); ok { + t.Fatal("digest reader exposes io.Seeker") + } + if _, err := io.CopyBuffer(io.Discard, reader, make([]byte, 113<<10)); err != nil { + t.Fatalf("read digest stream: %v", err) + } + file, err := reader.Digest() + if err != nil { + t.Fatalf("finish digest stream: %v", err) + } + digest, err := store.AssembleDigest([]store.DigestFile{file}) + if err != nil { + t.Fatalf("assemble digest: %v", err) + } + const want = "sha256:da8dd289e7d66408bd7d202642f180f2c22bc508cf0fe1d6e9f0e0b7967f6a5e" + if digest != want { + t.Errorf("digest = %q, want %q", digest, want) + } +} + +func TestUploadDigestedSetsContentLength(t *testing.T) { + data := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize)) + data = append(data, 'y') + path := filepath.Join(t.TempDir(), "boundary.bin") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatalf("write export: %v", err) + } + tm := &recordingTransferManager{} + st := &Store{tm: tm} + file, err := st.uploadDigested(t.Context(), "key", path, "boundary.bin") + if err != nil { + t.Fatalf("upload digested: %v", err) + } + if tm.contentLength == nil || *tm.contentLength != int64(len(data)) { + t.Fatalf("ContentLength = %v, want %d", tm.contentLength, len(data)) + } + if tm.read != int64(len(data)) { + t.Fatalf("uploaded bytes = %d, want %d", tm.read, len(data)) + } + digest, err := store.AssembleDigest([]store.DigestFile{file}) + if err != nil { + t.Fatalf("assemble digest: %v", err) + } + const want = "sha256:da8dd289e7d66408bd7d202642f180f2c22bc508cf0fe1d6e9f0e0b7967f6a5e" + if digest != want { + t.Errorf("digest = %q, want %q", digest, want) + } +} + +func TestPublishDigestedRetryAndMetaRequestAccounting(t *testing.T) { + const id = "ck_00000000000000dd" fake := &fakeS3{objects: map[string][]byte{}} - ts := httptest.NewServer(fake) - t.Cleanup(ts.Close) + st := newTestStore(t, fake) + meta := []byte(`{"id":"` + id + `","gen":1}`) + content := "retry-body" + exportKey := "ck/" + id + "/" + store.ExportGen(meta) + "/disk.img" + metaKey := "ck/" + id + "/" + store.MetaFile + fake.setPutFailures(exportKey, 1) - t.Setenv("AWS_ACCESS_KEY_ID", "test") - t.Setenv("AWS_SECRET_ACCESS_KEY", "test") - t.Setenv("AWS_REQUEST_CHECKSUM_CALCULATION", "when_required") - t.Setenv("AWS_RESPONSE_CHECKSUM_VALIDATION", "when_required") - cfg := Config{ - Bucket: "testbucket", Prefix: "ck/", Endpoint: ts.URL, - Region: "us-east-1", ForcePathStyle: true, + staging := stageRecord(t, st, id, meta, content) + digest, err := st.PublishDigested(t.Context(), staging, id) + if err != nil { + t.Fatalf("PublishDigested: %v", err) + } + chunk := sha256.Sum256([]byte(content)) + want, err := store.AssembleDigest([]store.DigestFile{{ + Path: "disk.img", Size: int64(len(content)), Chunks: [][sha256.Size]byte{chunk}, + }}) + if err != nil { + t.Fatalf("assemble expected digest: %v", err) + } + if digest != want { + t.Errorf("digest = %q, want %q", digest, want) + } + if got := fake.requestCount(http.MethodPut, exportKey); got != 2 { + t.Errorf("export PUT requests = %d, want 2", got) + } + if got := fake.contentLengths(exportKey); !slices.Equal(got, []int64{int64(len(content)), int64(len(content))}) { + t.Errorf("export content lengths = %v, want two %d-byte requests", got, len(content)) + } + if got := fake.objectMetadata(metaKey)[digestMetadataKey]; got != want { + t.Errorf("marker digest metadata = %q, want %q", got, want) + } + + fake.resetRequests() + _, _, fetchedDigest, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch miss: %v", err) + } + release() + if fetchedDigest != want { + t.Errorf("Fetch miss digest = %q, want %q", fetchedDigest, want) + } + assertSingleMetaGet(t, fake, metaKey) + + fake.resetRequests() + _, _, fetchedDigest, release, err = st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch hit: %v", err) + } + release() + if fetchedDigest != want { + t.Errorf("Fetch hit digest = %q, want %q", fetchedDigest, want) + } + assertSingleMetaGet(t, fake, metaKey) + + publishRecord(t, st, id, []byte(`{"id":"`+id+`","gen":2}`), "plain") + _, _, fetchedDigest, release, err = st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch plain replacement: %v", err) + } + release() + if fetchedDigest != "" { + t.Errorf("plain replacement digest = %q, want empty", fetchedDigest) } - reader, err := New(t.Context(), cfg, t.TempDir(), store.CheckpointIDRe) + if got := fake.objectMetadata(metaKey)[digestMetadataKey]; got != "" { + t.Errorf("plain marker retained digest metadata %q", got) + } +} + +func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) { + const id = "ck_00000000000000ee" + fake := &fakeS3{objects: map[string][]byte{}} + st := newTestStore(t, fake) + oldMeta := []byte(`{"id":"` + id + `","gen":1}`) + oldStaging := stageRecord(t, st, id, oldMeta, "old") + oldDigest, err := st.PublishDigested(t.Context(), oldStaging, id) + if err != nil { + t.Fatalf("publish old generation: %v", err) + } + + newMeta := []byte(`{"id":"` + id + `","gen":2}`) + newStaging := stageRecord(t, st, id, newMeta, "replacement") + newExportKey := "ck/" + id + "/" + store.ExportGen(newMeta) + "/disk.img" + metaKey := "ck/" + id + "/" + store.MetaFile + markerPuts := fake.requestCount(http.MethodPut, metaKey) + fake.setPutFailures(newExportKey, 100) + digest, err := st.PublishDigested(t.Context(), newStaging, id) + if err == nil { + t.Fatal("PublishDigested succeeded with injected export failure") + } + if digest != "" { + t.Errorf("failed PublishDigested returned digest %q", digest) + } + if got := fake.requestCount(http.MethodPut, metaKey); got != markerPuts { + t.Errorf("marker PUT requests after export failure = %d, want %d", got, markerPuts) + } + + dir, meta, fetchedDigest, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch old generation: %v", err) + } + defer release() + content, err := os.ReadFile(filepath.Join(dir, "disk.img")) //nolint:gosec // test path + if err != nil { + t.Fatalf("read old generation: %v", err) + } + if string(meta) != string(oldMeta) || string(content) != "old" || fetchedDigest != oldDigest { + t.Errorf("old generation = meta %q, content %q, digest %q", meta, content, fetchedDigest) + } + + fake.setPutFailures(newExportKey, 0) + newDigest, err := st.PublishDigested(t.Context(), newStaging, id) if err != nil { - t.Fatalf("new reader: %v", err) + t.Fatalf("retry PublishDigested: %v", err) } - writer, err := New(t.Context(), cfg, t.TempDir(), store.CheckpointIDRe) + dir, meta, fetchedDigest, release, err = st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch replacement: %v", err) + } + defer release() + content, err = os.ReadFile(filepath.Join(dir, "disk.img")) //nolint:gosec // test path if err != nil { - t.Fatalf("new writer: %v", err) + t.Fatalf("read replacement: %v", err) } + if string(meta) != string(newMeta) || string(content) != "replacement" || fetchedDigest != newDigest { + t.Errorf("replacement = meta %q, content %q, digest %q", meta, content, fetchedDigest) + } +} + +func TestRepublishRetainsGenerationSelectedByAnotherStore(t *testing.T) { + const id = "ck_00000000000000aa" + fake := &fakeS3{objects: map[string][]byte{}} + reader := newTestStore(t, fake) + writer := newTestStore(t, fake) publishRecord(t, writer, id, []byte(`{"id":"`+id+`","gen":1}`), "first") selected, err := reader.ReadMeta(t.Context(), id) if err != nil { @@ -273,6 +381,14 @@ func TestS3BackendContractRealEndpoint(t *testing.T) { } func publishRecord(t *testing.T, st *Store, id string, meta []byte, content string) { + t.Helper() + staging := stageRecord(t, st, id, meta, content) + if err := st.Publish(t.Context(), staging, id); err != nil { + t.Fatalf("publish record: %v", err) + } +} + +func stageRecord(t *testing.T, st *Store, id string, meta []byte, content string) string { t.Helper() staging, err := st.Stage(id) if err != nil { @@ -288,147 +404,338 @@ func publishRecord(t *testing.T, st *Store, id string, meta []byte, content stri if err := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil { t.Fatalf("write meta: %v", err) } - if err := st.Publish(t.Context(), staging, id); err != nil { - t.Fatalf("publish record: %v", err) + return staging +} + +func newTestStore(t *testing.T, fake *fakeS3) *Store { + t.Helper() + ts := httptest.NewServer(fake) + t.Cleanup(ts.Close) + t.Setenv("AWS_ACCESS_KEY_ID", "test") + t.Setenv("AWS_SECRET_ACCESS_KEY", "test") + // Keep PutObject bodies plain instead of adding aws-chunked trailers the fake would need to decode. + t.Setenv("AWS_REQUEST_CHECKSUM_CALCULATION", "when_required") + t.Setenv("AWS_RESPONSE_CHECKSUM_VALIDATION", "when_required") + st, err := New(t.Context(), Config{ + Bucket: "testbucket", Prefix: "ck/", Endpoint: ts.URL, + Region: "us-east-1", ForcePathStyle: true, + }, t.TempDir(), store.CheckpointIDRe) + if err != nil { + t.Fatalf("New: %v", err) } + return st +} + +func assertSingleMetaGet(t *testing.T, fake *fakeS3, metaKey string) { + t.Helper() + if got := fake.requestCount(http.MethodGet, metaKey); got != 1 { + t.Errorf("meta GET requests = %d, want 1", got) + } + if got := fake.requestCount(http.MethodHead, metaKey); got != 0 { + t.Errorf("meta HEAD requests = %d, want 0", got) + } +} + +type patternReader struct { + data []byte + limits []int + offset int + read int +} + +func (r *patternReader) Read(p []byte) (int, error) { + if r.offset == len(r.data) { + return 0, io.EOF + } + limit := r.limits[r.read%len(r.limits)] + r.read++ + n := min(len(p), limit, len(r.data)-r.offset) + copy(p, r.data[r.offset:r.offset+n]) + r.offset += n + return n, nil +} + +type recordingTransferManager struct { + contentLength *int64 + read int64 +} + +func (m *recordingTransferManager) UploadObject( + _ context.Context, + input *transfermanager.UploadObjectInput, + _ ...func(*transfermanager.Options), +) (*transfermanager.UploadObjectOutput, error) { + if input.ContentLength != nil { + length := *input.ContentLength + m.contentLength = &length + } + read, err := io.CopyBuffer(io.Discard, input.Body, make([]byte, 113<<10)) + m.read = read + return &transfermanager.UploadObjectOutput{}, err +} + +func (*recordingTransferManager) DownloadObject( + context.Context, + *transfermanager.DownloadObjectInput, + ...func(*transfermanager.Options), +) (*transfermanager.DownloadObjectOutput, error) { + return nil, fmt.Errorf("unexpected download") +} + +type fakeRequest struct { + method string + key string } -// fakeS3 implements just enough of the S3 REST surface (path-style) for -// the backend: PutObject, GetObject, DeleteObjects, ListObjectsV2. type fakeS3 struct { mu sync.Mutex - objects map[string][]byte // key -> body + objects map[string][]byte + metadata map[string]map[string]string + failPuts map[string]int + requests map[fakeRequest]int + putLengths map[string][]int64 + deleteBatches [][]string failList bool failDelete bool - deleteBatches [][]string } func (f *fakeS3) ServeHTTP(w http.ResponseWriter, r *http.Request) { f.mu.Lock() defer f.mu.Unlock() key := strings.TrimPrefix(r.URL.Path, "/testbucket/") + if f.requests == nil { + f.requests = map[fakeRequest]int{} + } + f.requests[fakeRequest{method: r.Method, key: key}]++ + query := r.URL.Query() switch { + case r.Method == http.MethodGet && query.Get("list-type") == "2": + f.handleListObjects(w, r) + case r.Method == http.MethodPost && query.Has("delete"): + f.handleDeleteObjects(w, r) case r.Method == http.MethodPut: - body, _ := io.ReadAll(r.Body) - f.objects[key] = body - case r.Method == http.MethodGet && r.URL.Query().Get("list-type") == "2": - if f.failList { - http.Error(w, "injected list failure", http.StatusBadRequest) - return - } - prefix := r.URL.Query().Get("prefix") - delim := r.URL.Query().Get("delimiter") - type object struct { - Key string `xml:"Key"` - Size int `xml:"Size"` - } - type commonPrefix struct { - Prefix string `xml:"Prefix"` - } - var result struct { - XMLName xml.Name `xml:"ListBucketResult"` - IsTruncated bool `xml:"IsTruncated"` - Contents []object - CommonPrefixes []commonPrefix - } - seen := map[string]bool{} - for k, v := range f.objects { - if !strings.HasPrefix(k, prefix) { - continue - } - if delim != "" { - if i := strings.Index(k[len(prefix):], delim); i >= 0 { - cp := k[:len(prefix)+i+1] - if !seen[cp] { - seen[cp] = true - result.CommonPrefixes = append(result.CommonPrefixes, commonPrefix{Prefix: cp}) - } - continue - } - } - result.Contents = append(result.Contents, object{Key: k, Size: len(v)}) - } - slices.SortFunc(result.Contents, func(a, b object) int { return cmp.Compare(a.Key, b.Key) }) - slices.SortFunc(result.CommonPrefixes, func(a, b commonPrefix) int { return cmp.Compare(a.Prefix, b.Prefix) }) - w.Header().Set("Content-Type", "application/xml") - _ = xml.NewEncoder(w).Encode(result) + f.handlePutObject(w, r, key) case r.Method == http.MethodHead: - body, ok := f.objects[key] - if !ok { - http.Error(w, "NoSuchKey", http.StatusNotFound) - return - } - w.Header().Set("Content-Length", strconv.Itoa(len(body))) + f.handleHeadObject(w, key) case r.Method == http.MethodGet: - body, ok := f.objects[key] - if !ok { - w.WriteHeader(http.StatusNotFound) - _, _ = w.Write([]byte(`NoSuchKey`)) - return + f.handleGetObject(w, r, key) + case r.Method == http.MethodDelete: + f.handleDeleteObject(w, key) + default: + http.Error(w, "unsupported", http.StatusNotImplemented) + } +} + +func (f *fakeS3) setPutFailures(key string, count int) { + f.mu.Lock() + defer f.mu.Unlock() + if f.failPuts == nil { + f.failPuts = map[string]int{} + } + f.failPuts[key] = count +} + +func (f *fakeS3) requestCount(method, key string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.requests[fakeRequest{method: method, key: key}] +} + +func (f *fakeS3) contentLengths(key string) []int64 { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.putLengths[key]) +} + +func (f *fakeS3) objectMetadata(key string) map[string]string { + f.mu.Lock() + defer f.mu.Unlock() + return maps.Clone(f.metadata[key]) +} + +func (f *fakeS3) resetRequests() { + f.mu.Lock() + defer f.mu.Unlock() + clear(f.requests) +} + +func (f *fakeS3) handlePutObject(w http.ResponseWriter, r *http.Request, key string) { + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "read body", http.StatusBadRequest) + return + } + if f.putLengths == nil { + f.putLengths = map[string][]int64{} + } + f.putLengths[key] = append(f.putLengths[key], r.ContentLength) + if f.failPuts[key] > 0 { + f.failPuts[key]-- + writeFakeError(w, http.StatusInternalServerError, "InternalError", "injected put failure") + return + } + if f.objects == nil { + f.objects = map[string][]byte{} + } + if f.metadata == nil { + f.metadata = map[string]map[string]string{} + } + f.objects[key] = bytes.Clone(body) + metadata := requestMetadata(r.Header) + if len(metadata) == 0 { + delete(f.metadata, key) + } else { + f.metadata[key] = metadata + } + w.Header().Set("ETag", `"put"`) +} + +func (f *fakeS3) handleListObjects(w http.ResponseWriter, r *http.Request) { + if f.failList { + http.Error(w, "injected list failure", http.StatusBadRequest) + return + } + prefix := r.URL.Query().Get("prefix") + delim := r.URL.Query().Get("delimiter") + type object struct { + Key string `xml:"Key"` + Size int `xml:"Size"` + } + type commonPrefix struct { + Prefix string `xml:"Prefix"` + } + var result struct { + XMLName xml.Name `xml:"ListBucketResult"` + IsTruncated bool `xml:"IsTruncated"` + Contents []object + CommonPrefixes []commonPrefix + } + seen := map[string]bool{} + for key, body := range f.objects { + if !strings.HasPrefix(key, prefix) { + continue } - // transfermanager sizes ranged downloads via HeadObject then GETs - // with a Range header; serve it so multi-part gets stay correct. - if rng := r.Header.Get("Range"); rng != "" { - var start, end int - if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end); err == nil && start < len(body) { - if end >= len(body) { - end = len(body) - 1 + if delim != "" { + if i := strings.Index(key[len(prefix):], delim); i >= 0 { + common := key[:len(prefix)+i+1] + if !seen[common] { + seen[common] = true + result.CommonPrefixes = append(result.CommonPrefixes, commonPrefix{Prefix: common}) } - w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(body))) - w.WriteHeader(http.StatusPartialContent) - _, _ = w.Write(body[start : end+1]) - return + continue } } - _, _ = w.Write(body) - case r.Method == http.MethodPost && r.URL.Query().Has("delete"): - var req struct { - Objects []struct { - Key string `xml:"Key"` - } `xml:"Object"` - } - body, _ := io.ReadAll(r.Body) - if err := xml.Unmarshal(body, &req); err != nil { - http.Error(w, "bad delete xml", http.StatusBadRequest) - return - } - batch := make([]string, len(req.Objects)) - for i := range req.Objects { - batch[i] = req.Objects[i].Key - } - f.deleteBatches = append(f.deleteBatches, batch) - // Strict-backend emulation: absent keys answer a per-entry NoSuchKey - // (AWS succeeds silently) so the client's tolerance stays exercised. - type delErr struct { - Key string `xml:"Key"` - Code string `xml:"Code"` - Message string `xml:"Message,omitempty"` - } - var result struct { - XMLName xml.Name `xml:"DeleteResult"` - Errors []delErr `xml:"Error"` - } - if f.failDelete { - result.Errors = append(result.Errors, delErr{ - Key: req.Objects[0].Key, Code: "AccessDenied", Message: "injected delete failure", - }) - w.Header().Set("Content-Type", "application/xml") - _ = xml.NewEncoder(w).Encode(result) + result.Contents = append(result.Contents, object{Key: key, Size: len(body)}) + } + slices.SortFunc(result.Contents, func(a, b object) int { return cmp.Compare(a.Key, b.Key) }) + slices.SortFunc(result.CommonPrefixes, func(a, b commonPrefix) int { return cmp.Compare(a.Prefix, b.Prefix) }) + w.Header().Set("Content-Type", "application/xml") + _ = xml.NewEncoder(w).Encode(result) +} + +func (f *fakeS3) handleHeadObject(w http.ResponseWriter, key string) { + body, ok := f.objects[key] + if !ok { + http.Error(w, "NoSuchKey", http.StatusNotFound) + return + } + writeMetadata(w.Header(), f.metadata[key]) + w.Header().Set("Content-Length", strconv.Itoa(len(body))) +} + +func (f *fakeS3) handleGetObject(w http.ResponseWriter, r *http.Request, key string) { + body, ok := f.objects[key] + if !ok { + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`NoSuchKey`)) + return + } + writeMetadata(w.Header(), f.metadata[key]) + if rng := r.Header.Get("Range"); rng != "" { + var start, end int + if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end); err == nil && start < len(body) { + end = min(end, len(body)-1) + w.Header().Set("Content-Length", strconv.Itoa(end-start+1)) + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(body))) + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write(body[start : end+1]) return } - for _, o := range req.Objects { - if _, ok := f.objects[o.Key]; !ok { - result.Errors = append(result.Errors, delErr{Key: o.Key, Code: "NoSuchKey"}) - continue - } - delete(f.objects, o.Key) - } + } + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + _, _ = w.Write(body) +} + +func (f *fakeS3) handleDeleteObjects(w http.ResponseWriter, r *http.Request) { + var req struct { + Objects []struct { + Key string `xml:"Key"` + } `xml:"Object"` + } + body, err := io.ReadAll(r.Body) + if err != nil || xml.Unmarshal(body, &req) != nil { + http.Error(w, "bad delete xml", http.StatusBadRequest) + return + } + batch := make([]string, len(req.Objects)) + for i := range req.Objects { + batch[i] = req.Objects[i].Key + } + f.deleteBatches = append(f.deleteBatches, batch) + type deleteError struct { + Key string `xml:"Key"` + Code string `xml:"Code"` + Message string `xml:"Message,omitempty"` + } + var result struct { + XMLName xml.Name `xml:"DeleteResult"` + Errors []deleteError `xml:"Error"` + } + if f.failDelete { + result.Errors = append(result.Errors, deleteError{ + Key: req.Objects[0].Key, Code: "AccessDenied", Message: "injected delete failure", + }) w.Header().Set("Content-Type", "application/xml") _ = xml.NewEncoder(w).Encode(result) - case r.Method == http.MethodDelete: - delete(f.objects, key) - w.WriteHeader(http.StatusNoContent) - default: - http.Error(w, "unsupported", http.StatusNotImplemented) + return } + for _, object := range req.Objects { + if _, ok := f.objects[object.Key]; !ok { + result.Errors = append(result.Errors, deleteError{Key: object.Key, Code: "NoSuchKey"}) + continue + } + delete(f.objects, object.Key) + delete(f.metadata, object.Key) + } + w.Header().Set("Content-Type", "application/xml") + _ = xml.NewEncoder(w).Encode(result) +} + +func (f *fakeS3) handleDeleteObject(w http.ResponseWriter, key string) { + delete(f.objects, key) + delete(f.metadata, key) + w.WriteHeader(http.StatusNoContent) +} + +func requestMetadata(header http.Header) map[string]string { + metadata := map[string]string{} + for key, values := range header { + key = strings.ToLower(key) + if name, ok := strings.CutPrefix(key, "x-amz-meta-"); ok && len(values) > 0 { + metadata[name] = values[0] + } + } + return metadata +} + +func writeMetadata(header http.Header, metadata map[string]string) { + for key, value := range metadata { + header.Set("x-amz-meta-"+key, value) + } +} + +func writeFakeError(w http.ResponseWriter, status int, code, message string) { + w.Header().Set("Content-Type", "application/xml") + w.WriteHeader(status) + _, _ = fmt.Fprintf(w, `%s%s`, code, message) } diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go index a4203f62..ebbaba7f 100644 --- a/sandboxd/store/store.go +++ b/sandboxd/store/store.go @@ -48,10 +48,13 @@ type Store interface { // in-process; request-path callers pass an uncancelable ctx so a started // publish finishes. Publish(ctx context.Context, staging, id string) error + // PublishDigested applies Publish semantics and returns the export digest. + PublishDigested(ctx context.Context, staging, id string) (string, error) // Fetch materializes a record's snapshot export as a local directory - // cocoon can clone from, plus the meta it resolved on the way, and a - // release to hold until the clone is done (the generation outlives it). - Fetch(ctx context.Context, id string) (dir string, meta []byte, release func(), err error) + // cocoon can clone from, plus the meta and digest resolved on the way, + // and a release to hold until clone ends (the generation outlives it). + // The digest is empty for records written by Publish. + Fetch(ctx context.Context, id string) (dir string, meta []byte, digest string, release func(), err error) // ReadMeta returns a record's metadata, or an error when the record // does not exist. ReadMeta(ctx context.Context, id string) ([]byte, error) diff --git a/sandboxd/store/storetest/storetest.go b/sandboxd/store/storetest/storetest.go index fcb545e7..b565381a 100644 --- a/sandboxd/store/storetest/storetest.go +++ b/sandboxd/store/storetest/storetest.go @@ -13,7 +13,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store" ) -// RunContract drives one backend through the full checkpoint lifecycle. +// RunContract drives one backend through plain and digested record lifecycles. func RunContract(t *testing.T, st store.Store) { t.Helper() ctx := t.Context() @@ -36,13 +36,16 @@ func RunContract(t *testing.T, st store.Store) { t.Fatalf("ReadMeta: %q, %v", raw, err) } - dir, meta, release, err := st.Fetch(ctx, id) + dir, meta, digest, release, err := st.Fetch(ctx, id) if err != nil { t.Fatalf("Fetch: %v", err) } if string(meta) != `{"id":"`+id+`"}` { t.Fatalf("Fetch meta: %q, want the published record", meta) } + if digest != "" { + t.Fatalf("Fetch digest after plain Publish: %q, want empty", digest) + } got, err := os.ReadFile(filepath.Join(dir, "disk.img")) //nolint:gosec // test path if err != nil || string(got) != "snapshot-bytes" { t.Fatalf("fetched export: %q, %v", got, err) @@ -81,13 +84,16 @@ func RunContract(t *testing.T, st store.Store) { if err = st.Publish(ctx, second, id); err != nil { t.Fatalf("Publish second: %v", err) } - dir, meta, release, err = st.Fetch(ctx, id) + dir, meta, digest, release, err = st.Fetch(ctx, id) if err != nil { t.Fatalf("Fetch second: %v", err) } if string(meta) != `{"id":"`+id+`","gen":2}` { t.Fatalf("Fetch meta after re-publish: %q, want the second generation", meta) } + if digest != "" { + t.Fatalf("Fetch digest after plain re-publish: %q, want empty", digest) + } if _, statErr := os.Stat(filepath.Join(dir, "disk.img")); !os.IsNotExist(statErr) { t.Errorf("first-generation file survived re-publish: %v", statErr) } @@ -102,21 +108,91 @@ func RunContract(t *testing.T, st store.Store) { if _, err = st.ReadMeta(ctx, id); !errors.Is(err, store.ErrNotFound) { t.Fatalf("ReadMeta after Delete: %v, want store.ErrNotFound", err) } - if _, _, _, err = st.Fetch(ctx, id); !errors.Is(err, store.ErrNotFound) { + if _, _, _, _, err = st.Fetch(ctx, id); !errors.Is(err, store.ErrNotFound) { t.Fatalf("Fetch after Delete: %v, want store.ErrNotFound", err) } if metas, err = st.Metas(ctx); err != nil || len(metas) != 0 { t.Fatalf("Metas after Delete: %d, %v", len(metas), err) } + runDigestContract(t, st) +} + +func runDigestContract(t *testing.T, st store.Store) { + t.Helper() + ctx := t.Context() + const id = "ck_00000000000000cc" + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage digested: %v", err) + } + writeExport(t, staging, "z.bin", "z") + writeExport(t, staging, "nested/a.bin", "a") + if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`"}`), 0o600); err != nil { + t.Fatalf("write digested meta: %v", err) + } + digest, err := st.PublishDigested(ctx, staging, id) + if err != nil { + t.Fatalf("PublishDigested: %v", err) + } + const want = "sha256:ad65b315de70e494c767969e65fc5de65c73846c4ebcdc7051abe08b056637ad" + if digest != want { + t.Fatalf("PublishDigested digest %q, want %q", digest, want) + } + _, _, fetchedDigest, release, err := st.Fetch(ctx, id) + if err != nil { + t.Fatalf("Fetch digested: %v", err) + } + if fetchedDigest != want { + t.Errorf("Fetch digest %q, want %q", fetchedDigest, want) + } + release() + rejectNonRegularReplacement(t, st, id, want) + if err = st.Delete(ctx, id); err != nil { + t.Fatalf("Delete digested: %v", err) + } +} + +func rejectNonRegularReplacement(t *testing.T, st store.Store, id, wantDigest string) { + t.Helper() + staging, err := st.Stage(id) + if err != nil { + t.Fatalf("Stage non-regular replacement: %v", err) + } + defer func() { _ = os.RemoveAll(staging) }() + export := filepath.Join(staging, store.ExportDir) + if err = os.MkdirAll(export, 0o750); err != nil { + t.Fatalf("mkdir non-regular export: %v", err) + } + if err = os.Symlink("target", filepath.Join(export, "link")); err != nil { + t.Fatalf("symlink export: %v", err) + } + if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`","gen":2}`), 0o600); err != nil { + t.Fatalf("write non-regular meta: %v", err) + } + if _, err = st.PublishDigested(t.Context(), staging, id); err == nil { + t.Fatal("PublishDigested accepted a non-regular export entry") + } + dir, meta, digest, release, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("Fetch after rejected replacement: %v", err) + } + defer release() + if string(meta) != `{"id":"`+id+`"}` || digest != wantDigest { + t.Errorf("committed meta/digest after rejection = %q/%q, want original/%q", meta, digest, wantDigest) + } + content, err := os.ReadFile(filepath.Join(dir, "nested", "a.bin")) //nolint:gosec // test path + if err != nil || string(content) != "a" { + t.Errorf("committed export after rejection = %q, %v, want a", content, err) + } } func writeExport(t *testing.T, staging, name, content string) { t.Helper() - dir := filepath.Join(staging, store.ExportDir) - if err := os.MkdirAll(dir, 0o750); err != nil { + path := filepath.Join(staging, store.ExportDir, name) + if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { t.Fatalf("mkdir export: %v", err) } - if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600); err != nil { + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { t.Fatalf("write export: %v", err) } } From 881bd1d656c4f9be76c25341aab72c42624965b9 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 11 Aug 2026 23:46:21 +0800 Subject: [PATCH 2/7] store/dir: drop the post-hash restat of export entries Staging is single-writer under the template lock and every chunk read already fails on a short read; the extra per-file Lstat guarded nothing reachable. --- sandboxd/store/dir/dir.go | 7 ------- 1 file changed, 7 deletions(-) diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index e3de6957..e84e8f88 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -397,13 +397,6 @@ func hashExport(ctx context.Context, root string) (string, error) { } files := make([]store.DigestFile, len(sources)) for i := range sources { - info, err := os.Lstat(sources[i].path) - if err != nil { - return "", fmt.Errorf("stat export entry %s: %w", sources[i].digest.Path, err) - } - if !info.Mode().IsRegular() || info.Size() != sources[i].digest.Size { - return "", fmt.Errorf("export entry %s changed while hashing", sources[i].digest.Path) - } files[i] = sources[i].digest } return store.AssembleDigest(files) From c10a9506963499607e4cbb151bb90750a3ee48d3 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 11 Aug 2026 23:46:36 +0800 Subject: [PATCH 3/7] review: name the dir chunk hasher func type (signatures) --- sandboxd/store/dir/dir.go | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index e84e8f88..6770a30c 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -45,6 +45,8 @@ type digestJob struct { chunk int } +type chunkHasher func(source *digestSource, chunk int, buffer []byte) error + var _ store.Store = (*Store)(nil) // Store keeps records as //{meta.json,export-/...}: the @@ -402,12 +404,7 @@ func hashExport(ctx context.Context, root string) (string, error) { return store.AssembleDigest(files) } -func hashChunks( - ctx context.Context, - sources []digestSource, - workers int, - hashFn func(*digestSource, int, []byte) error, -) error { +func hashChunks(ctx context.Context, sources []digestSource, workers int, hashFn chunkHasher) error { jobs := make(chan digestJob) group, groupCtx := errgroup.WithContext(ctx) for range workers { From 0d3540ecbd2472c0fc704e5827b4136fcef75bf0 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 11 Aug 2026 23:50:53 +0800 Subject: [PATCH 4/7] review: move helpers below exported sets, uncluster payload types (asl) --- protocol/wire/frame.go | 22 +++++++------- sandboxd/pool/pool.go | 68 +++++++++++++++++++++--------------------- sdk/go/port.go | 20 ++++++------- 3 files changed, 55 insertions(+), 55 deletions(-) diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 8dda469f..4f50ac25 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -682,6 +682,17 @@ func DecodeResponse(line []byte) (Response, error) { return dec(line) } +// AppendBulkRequest renders a data-carrying request frame — +// {"v":1,"op":,"data":""} plus newline — into buf, reused across +// calls on the bulk send paths (base64's alphabet needs no JSON escaping). +func AppendBulkRequest(buf []byte, op string, data []byte) []byte { + buf = append(buf[:0], requestHead...) + buf = append(buf, op...) + buf = append(buf, `","data":"`...) + buf = base64.StdEncoding.AppendEncode(buf, data) + return append(buf, '"', '}', '\n') +} + // fastBulk decodes a bulk frame's base64 data field by slicing it out // (base64's alphabet is JSON-escape-free) and skipping json.Unmarshal, which // otherwise dominates the download path. Only the exact canonical shape both @@ -711,17 +722,6 @@ func fastBulk(tag string, slow func([]byte) (Response, error), mk func([]byte) R } } -// AppendBulkRequest renders a data-carrying request frame — -// {"v":1,"op":,"data":""} plus newline — into buf, reused across -// calls on the bulk send paths (base64's alphabet needs no JSON escaping). -func AppendBulkRequest(buf []byte, op string, data []byte) []byte { - buf = append(buf[:0], requestHead...) - buf = append(buf, op...) - buf = append(buf, `","data":"`...) - buf = base64.StdEncoding.AppendEncode(buf, data) - return append(buf, '"', '}', '\n') -} - // encodeTagged marshals v as a flat object and splices the tag head in front // of its fields, so wire tags never live on the structs themselves. The // spare capacity byte lets Conn.Send append the newline without a copy. diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 6d0ce96e..cc1d7dd3 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -581,14 +581,6 @@ func (m *Manager) Sandboxes() []SandboxSummary { return out } -func summarize(sb *types.Sandbox) SandboxSummary { - return SandboxSummary{ - ID: sb.ID, Key: sb.Key, Deadline: sb.Deadline, - Hibernated: sb.HibernateSnap != "", Archived: sb.ArchiveCk != "", - FromCheckpoint: sb.FromCheckpoint, ClaimRef: sb.ClaimRef, - } -} - // WarmCounts is the per-pool-key-hash warm count, for gossiping placement. func (m *Manager) WarmCounts() map[string]int { m.mu.Lock() @@ -602,32 +594,6 @@ func (m *Manager) WarmCounts() map[string]int { return counts } -func loadEgressCA(cfg *config.EgressCAConfig) (*egress.CA, error) { - root, err := os.ReadFile(cfg.RootCert) //nolint:gosec // operator-configured ca path - if err != nil { - return nil, fmt.Errorf("read root cert: %w", err) - } - interCert, err := os.ReadFile(cfg.IntermediateCert) //nolint:gosec // operator-configured ca path - if err != nil { - return nil, fmt.Errorf("read intermediate cert: %w", err) - } - interKey, err := os.ReadFile(cfg.IntermediateKey) //nolint:gosec // operator-configured ca path - if err != nil { - return nil, fmt.Errorf("read intermediate key: %w", err) - } - return egress.LoadCA(root, interCert, interKey) -} - -// newStoreView builds one id-namespaced view of the configured backend. -// The dir default lives here rather than config.applyDefaults: tests build -// Config directly, skipping Load. -func newStoreView(ctx context.Context, cfg *config.Config, staging string, idRe *regexp.Regexp) (store.Store, error) { - if cs := cfg.CheckpointStore; cs != nil && cs.Kind == "s3" { - return s3.New(ctx, *cs.S3, filepath.Join(cfg.DataDir, staging), idRe) - } - return dir.New(cmp.Or(cfg.CheckpointDir, filepath.Join(cfg.DataDir, "checkpoints")), idRe) -} - // WithPeerHeal installs the healer ClaimCheckpointHeal pulls through, so a // peer transfer is paid only once nothing cheaper resolves the claim. func (m *Manager) WithPeerHeal(enabled bool, owners peer.Owners, token string) { @@ -685,6 +651,40 @@ func (m *Manager) goldensDir() string { return filepath.Join(m.dataDir, "goldens") } +func summarize(sb *types.Sandbox) SandboxSummary { + return SandboxSummary{ + ID: sb.ID, Key: sb.Key, Deadline: sb.Deadline, + Hibernated: sb.HibernateSnap != "", Archived: sb.ArchiveCk != "", + FromCheckpoint: sb.FromCheckpoint, ClaimRef: sb.ClaimRef, + } +} + +func loadEgressCA(cfg *config.EgressCAConfig) (*egress.CA, error) { + root, err := os.ReadFile(cfg.RootCert) //nolint:gosec // operator-configured ca path + if err != nil { + return nil, fmt.Errorf("read root cert: %w", err) + } + interCert, err := os.ReadFile(cfg.IntermediateCert) //nolint:gosec // operator-configured ca path + if err != nil { + return nil, fmt.Errorf("read intermediate cert: %w", err) + } + interKey, err := os.ReadFile(cfg.IntermediateKey) //nolint:gosec // operator-configured ca path + if err != nil { + return nil, fmt.Errorf("read intermediate key: %w", err) + } + return egress.LoadCA(root, interCert, interKey) +} + +// newStoreView builds one id-namespaced view of the configured backend. +// The dir default lives here rather than config.applyDefaults: tests build +// Config directly, skipping Load. +func newStoreView(ctx context.Context, cfg *config.Config, staging string, idRe *regexp.Regexp) (store.Store, error) { + if cs := cfg.CheckpointStore; cs != nil && cs.Kind == "s3" { + return s3.New(ctx, *cs.S3, filepath.Join(cfg.DataDir, staging), idRe) + } + return dir.New(cmp.Or(cfg.CheckpointDir, filepath.Join(cfg.DataDir, "checkpoints")), idRe) +} + func dirExists(path string) bool { fi, err := os.Stat(path) return err == nil && fi.IsDir() diff --git a/sdk/go/port.go b/sdk/go/port.go index d03b6fb4..9b4a2d87 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -98,16 +98,6 @@ func (s *Sandbox) DialPort(ctx context.Context, port uint16) (*PortConn, error) return s.openStream(ctx, &wire.PortForward{Port: port}) } -type previewRequest struct { - Token string `json:"token"` - Port uint16 `json:"port"` - TTLSeconds int `json:"ttl_seconds,omitempty"` -} - -type previewResponse struct { - URL string `json:"url"` -} - // PreviewURL mints a shareable URL that serves the sandbox's guest HTTP // port from a plain browser, valid for ttl (the node clamps it to the // claim's lease). Requires the node to have preview configured. @@ -191,6 +181,16 @@ func closeWrite(conn net.Conn) { } } +type previewRequest struct { + Token string `json:"token"` + Port uint16 `json:"port"` + TTLSeconds int `json:"ttl_seconds,omitempty"` +} + +type previewResponse struct { + URL string `json:"url"` +} + type portAddr string func (a portAddr) Network() string { return "silkd" } From 4a0e0d18d6fc6f891a51cae4acdda023ad6d7195 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 11 Aug 2026 23:52:05 +0800 Subject: [PATCH 5/7] review: compress touched-file comments to the why --- protocol/wire/frame.go | 12 +++--------- sandboxd/pool/pool.go | 4 +--- sdk/go/port.go | 3 +-- 3 files changed, 5 insertions(+), 14 deletions(-) diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 4f50ac25..102a7856 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -693,12 +693,8 @@ func AppendBulkRequest(buf []byte, op string, data []byte) []byte { return append(buf, '"', '}', '\n') } -// fastBulk decodes a bulk frame's base64 data field by slicing it out -// (base64's alphabet is JSON-escape-free) and skipping json.Unmarshal, which -// otherwise dominates the download path. Only the exact canonical shape both -// producers emit — {"type":"","data":""} — takes the slice; -// anything else falls back to slow, the full parse, so no byte of a frame -// escapes validation. +// fastBulk slices the base64 data out of a canonical bulk frame, skipping the +// json.Unmarshal that dominates downloads; any other shape falls back to slow. func fastBulk(tag string, slow func([]byte) (Response, error), mk func([]byte) Response) func([]byte) (Response, error) { head := []byte(`{"type":"` + tag + `","data":"`) return func(line []byte) (Response, error) { @@ -712,9 +708,7 @@ func fastBulk(tag string, slow func([]byte) (Response, error), mk func([]byte) R } out := make([]byte, base64.StdEncoding.DecodedLen(len(b64))) n, err := base64.StdEncoding.Decode(out, b64) - // Decode skips CR/LF (raw control bytes JSON forbids); a canonical - // payload decodes to exactly the encoded length, so any skip falls - // back to the full parse. + // Decode skips CR/LF, so a length mismatch means non-canonical input. if err != nil || base64.StdEncoding.EncodedLen(n) != len(b64) { return slow(line) } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index cc1d7dd3..fe333769 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -675,9 +675,7 @@ func loadEgressCA(cfg *config.EgressCAConfig) (*egress.CA, error) { return egress.LoadCA(root, interCert, interKey) } -// newStoreView builds one id-namespaced view of the configured backend. -// The dir default lives here rather than config.applyDefaults: tests build -// Config directly, skipping Load. +// The dir default lives here, not config.applyDefaults: tests build Config directly. func newStoreView(ctx context.Context, cfg *config.Config, staging string, idRe *regexp.Regexp) (store.Store, error) { if cs := cfg.CheckpointStore; cs != nil && cs.Kind == "s3" { return s3.New(ctx, *cs.S3, filepath.Join(cfg.DataDir, staging), idRe) diff --git a/sdk/go/port.go b/sdk/go/port.go index 9b4a2d87..8d5bf525 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -173,8 +173,7 @@ func (s *Sandbox) proxyConn(ctx context.Context, local net.Conn, port uint16) { <-done } -// closeWrite half-closes conn's write side when it supports it, so the peer -// sees EOF while the tail still drains. +// Half-close so the peer sees EOF while the tail still drains. func closeWrite(conn net.Conn) { if cw, ok := conn.(interface{ CloseWrite() error }); ok { _ = cw.CloseWrite() From be16043712225dc80a50e4987d60e63ed37ec88a Mon Sep 17 00:00:00 2001 From: CMGS Date: Wed, 12 Aug 2026 00:00:36 +0800 Subject: [PATCH 6/7] store: unify digest chunk counting, fold the root check into the export walk --- sandboxd/store/digest.go | 16 ++++++++++------ sandboxd/store/dir/dir.go | 23 +++++++++-------------- 2 files changed, 19 insertions(+), 20 deletions(-) diff --git a/sandboxd/store/digest.go b/sandboxd/store/digest.go index 15244417..7b79e042 100644 --- a/sandboxd/store/digest.go +++ b/sandboxd/store/digest.go @@ -23,6 +23,14 @@ type DigestFile struct { Chunks [][sha256.Size]byte // indexed by ascending file offset } +// ChunkCount returns how many DigestChunkSize chunks size spans. +func ChunkCount(size int64) int64 { + if size == 0 { + return 0 + } + return (size-1)/DigestChunkSize + 1 +} + // AssembleDigest returns the canonical v2 digest of indexed file chunks. func AssembleDigest(files []DigestFile) (string, error) { files = slices.Clone(files) @@ -40,12 +48,8 @@ func AssembleDigest(files []DigestFile) (string, error) { if file.Size < 0 { return "", fmt.Errorf("digest file %q has negative size %d", file.Path, file.Size) } - var wantChunks int64 - if file.Size > 0 { - wantChunks = (file.Size-1)/DigestChunkSize + 1 - } - if int64(len(file.Chunks)) != wantChunks { - return "", fmt.Errorf("digest file %q has %d chunks, want %d for size %d", file.Path, len(file.Chunks), wantChunks, file.Size) + if int64(len(file.Chunks)) != ChunkCount(file.Size) { + return "", fmt.Errorf("digest file %q has %d chunks, want %d for size %d", file.Path, len(file.Chunks), ChunkCount(file.Size), file.Size) } var frame [8]byte diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index 6770a30c..765ffca8 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -442,19 +442,18 @@ func hashChunks(ctx context.Context, sources []digestSource, workers int, hashFn } func collectDigestSources(root string) ([]digestSource, error) { - info, err := os.Lstat(root) - if err != nil { - return nil, err - } - if !info.IsDir() { - return nil, fmt.Errorf("export root is not a directory") - } var sources []digestSource - err = filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { + err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { if walkErr != nil { return walkErr } - if path == root || entry.IsDir() { + if path == root { + if !entry.IsDir() { + return fmt.Errorf("export root is not a directory") + } + return nil + } + if entry.IsDir() { return nil } rel, relErr := filepath.Rel(root, path) @@ -469,14 +468,10 @@ func collectDigestSources(root string) ([]digestSource, error) { if !entryInfo.Mode().IsRegular() { return fmt.Errorf("unsupported export entry %s (%s)", rel, entryInfo.Mode().Type()) } - chunks := entryInfo.Size() / store.DigestChunkSize - if entryInfo.Size()%store.DigestChunkSize != 0 { - chunks++ - } sources = append(sources, digestSource{ path: path, digest: store.DigestFile{ - Path: rel, Size: entryInfo.Size(), Chunks: make([][sha256.Size]byte, int(chunks)), + Path: rel, Size: entryInfo.Size(), Chunks: make([][sha256.Size]byte, store.ChunkCount(entryInfo.Size())), }, }) return nil From 91eafcf64638eaa2d004bb0a5a91374ac6e33176 Mon Sep 17 00:00:00 2001 From: CMGS Date: Wed, 12 Aug 2026 00:20:05 +0800 Subject: [PATCH 7/7] review: cut narrative and duplicated comment content repo-wide Restatements, perf-advertising clauses, and cross-duplicated godoc go; contract sentences (locking, crash/retry, security invariants) stay. --- e2e/cmd/rpcbench/main.go | 3 +-- e2e/cmd/smoke/main.go | 6 ++---- mcp/tools.go | 3 --- protocol/wire/frame.go | 3 +-- sandboxd/config/config.go | 7 +++---- sandboxd/egress/proxy.go | 7 +------ sandboxd/engine/engine.go | 8 ++------ sandboxd/engine/portconn.go | 7 +++---- sandboxd/mesh/mesh.go | 10 +++------- sandboxd/pool/checkpoint.go | 18 +++++------------ sandboxd/pool/hibernate.go | 9 +++------ sandboxd/pool/pool.go | 5 ++--- sandboxd/pool/reconcile.go | 6 ++---- sandboxd/pool/refill.go | 9 ++++----- sandboxd/pool/template.go | 1 - sandboxd/server/metrics.go | 2 +- sandboxd/server/relay.go | 3 +-- sandboxd/server/server.go | 2 +- sandboxd/store/dir/dir.go | 4 +--- sandboxd/store/peer/peer.go | 2 +- sandboxd/store/peer/probe.go | 7 +++---- sandboxd/store/peer/transport.go | 3 +-- sandboxd/types/api.go | 7 +++---- sandboxd/types/types.go | 3 +-- sandboxd/utils/utils.go | 3 +-- sdk/go/checkpoint.go | 2 +- sdk/go/client.go | 33 ++++++++++++++------------------ sdk/go/port.go | 3 +-- sdk/go/pty.go | 4 +--- sdk/go/silkd/conn.go | 3 +-- 30 files changed, 64 insertions(+), 119 deletions(-) diff --git a/e2e/cmd/rpcbench/main.go b/e2e/cmd/rpcbench/main.go index f60f52dd..bf112299 100644 --- a/e2e/cmd/rpcbench/main.go +++ b/e2e/cmd/rpcbench/main.go @@ -108,8 +108,7 @@ func run(addr, token, template string, n int) error { return nil } -// statRPC drives one fs_stat over the upgraded conn and closes it — the -// protocol is one RPC per connection. +// statRPC: the protocol is one RPC per connection. func statRPC(conn net.Conn) error { defer func() { _ = conn.Close() }() sc := silkd.NewConn(conn) diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index e389ea42..c9d3f5d6 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -289,7 +289,7 @@ func smokeGit(ctx context.Context, sb *sandbox.Sandbox) error { // smokeEgress claims a second sandbox on the egress lane and pins the // positive half of silkd's lane detection: git must actually run there, so a // push with no remote fails inside git — never with the none-lane -// unimplemented guard. Needs no reachable network. +// unimplemented guard. func smokeEgress(ctx context.Context, client *sandbox.Client, template string) error { sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetEgress)) if err != nil { @@ -388,8 +388,7 @@ func smokeFork(ctx context.Context, sb *sandbox.Sandbox) error { // smokePromote proves promote-to-template through the owner-bound handle: // claiming from it must clone the parent's state — on a real node a cold // boot of this never-registered image ref would fail, so success itself -// proves the golden path — and delete removes it. The name-based -// DeleteTemplate covers the node-local Client surface too. +// proves the golden path — and delete removes it. func smokePromote(ctx context.Context, client *sandbox.Client, sb *sandbox.Sandbox) error { tpl, err := sb.Promote(ctx, "smoke-tpl:v1") if err != nil { @@ -656,7 +655,6 @@ func isSilkdKind(err error, kind string) bool { return errors.As(err, &er) && er.Kind == kind } -// lspWrite frames one JSON-RPC message the LSP way (Content-Length header). func lspWrite(w io.Writer, body string) error { _, err := fmt.Fprintf(w, "Content-Length: %d\r\n\r\n%s", len(body), body) return err diff --git a/mcp/tools.go b/mcp/tools.go index 7e4d1d4c..ff151a13 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -111,8 +111,6 @@ func toolCreateSandbox(ctx context.Context, s *server, raw json.RawMessage) (str return jsonText(map[string]any{"sandbox_id": sb.ID, "deadline": sb.Deadline}), nil } -// sandboxArg is the shared sandbox_id field of the tool argument structs; -// parseAndBox reads it through id(). type sandboxArg struct { SandboxID string `json:"sandbox_id"` } @@ -388,7 +386,6 @@ func toolNodeInfo(ctx context.Context, s *server, _ json.RawMessage) (string, er return jsonText(info), nil } -// boxArg resolves a bare sandbox_id argument to a live handle. func (s *server) boxArg(raw json.RawMessage) (*sandbox.Sandbox, error) { _, sb, err := parseAndBox[sandboxArg](s, raw) return sb, err diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 102a7856..d223ebe7 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -668,8 +668,7 @@ func NewFrameScanner(r io.Reader) *bufio.Scanner { } // DecodeResponse parses one frame into its type's concrete Go type. Byte -// fields are freshly allocated per frame, so callers may retain them; pooling -// them would require a copy-out at every retention site first. +// fields are freshly allocated per frame, so callers may retain them. func DecodeResponse(line []byte) (Response, error) { typ, err := frameTag(line, respTagHead, "type") if err != nil { diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 0c091362..dc5da56a 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -117,7 +117,7 @@ type TenantSpec struct { // that tenant answers 401), and a // node serving the egress lane can only redirect egress claims to peers if it // too has an egress attachment (a no-egress node answers 409 rather than -// redirecting). Both are acceptable for a homogeneous cluster. +// redirecting). type MeshConfig struct { NodeID string `json:"node_id"` // unique name; defaults to Bind Bind string `json:"bind"` // memberlist host:port @@ -254,9 +254,8 @@ type Config struct { // addressing fields (never payloads) to /audit.jsonl. AuditLog bool `json:"audit_log,omitempty"` - // MaxForkCount caps children per fork call — each child is a full-RAM VM, - // so this bounds a single request's memory blast radius to the node's - // capacity. Defaults to 16. + // MaxForkCount caps children per fork call — each child is a full-RAM VM. + // Defaults to 16. MaxForkCount int `json:"max_fork_count,omitempty"` // RefillConcurrency caps concurrent VM provisioning node-wide — warm-pool diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index 2b671e89..b18a177b 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -136,10 +136,7 @@ func (p *Proxy) untrack(conn net.Conn) { p.connMu.Unlock() } -// serveConnect gates an HTTPS/opaque tunnel by host. A plain allow hijacks and -// splices the end-to-end TLS to the origin untouched. A matched rule with -// Intercept instead terminates the TLS (interception, see intercept.go) so the -// request is filtered by method and the secret injected; deny answers a typed 403. +// serveConnect gates an HTTPS/opaque tunnel by host. func (p *Proxy) serveConnect(w http.ResponseWriter, r *http.Request) { host := hostOnly(r.Host) // Host-gate the interception decision: the tunnel's CONNECT verb is not the @@ -177,8 +174,6 @@ func (p *Proxy) serveConnect(w http.ResponseWriter, r *http.Request) { splice(client, upstream) } -// serveForward gates an absolute-form plaintext request by host and method, -// injects the rule's credential, and relays it to the origin. func (p *Proxy) serveForward(w http.ResponseWriter, r *http.Request) { if !r.URL.IsAbs() { http.Error(w, "egress: proxy requires an absolute-form request URI", http.StatusBadRequest) diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index fb645f83..43bfccca 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -1,10 +1,6 @@ // Package engine drives VM lifecycle through the cocoon CLI and dials the -// in-guest silkd over hybrid vsock. -// -// cocoon runs as a subprocess deliberately: the CLI is cocoon's stable -// contract (it exports no lifecycle library), it is the exact interface every -// latency figure was measured through, and no lifecycle call sits on the -// warm-claim path. +// in-guest silkd over hybrid vsock. The CLI is cocoon's only stable contract +// (it exports no lifecycle library); no lifecycle call sits on the warm-claim path. package engine import ( diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index ddfe2ee6..c0b34b0f 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -61,7 +61,7 @@ func (g *guestPortConn) Read(p []byte) (int, error) { g.pending = frame.Data case "done", "": return 0, io.EOF // the guest closed the forwarded port - default: // error frame or anything terminal + default: return 0, fmt.Errorf("port stream ended: %s", frame.Type) } } @@ -87,9 +87,8 @@ func (g *guestPortConn) Write(p []byte) (int, error) { } // fastPortData slices the canonical data frame's base64 out without a JSON -// parse — json.Unmarshal otherwise dominates the download relay, and the -// SDK's fastBulk sets the contract: only the exact canonical shape takes the -// slice, anything else falls back to the full parse. +// parse; the SDK's fastBulk sets the contract: only the exact canonical +// shape takes the slice, anything else falls back to the full parse. func fastPortData(line []byte) ([]byte, bool) { after, ok := bytes.CutPrefix(line, portDataHead) if !ok { diff --git a/sandboxd/mesh/mesh.go b/sandboxd/mesh/mesh.go index a59f0e89..5eefcc37 100644 --- a/sandboxd/mesh/mesh.go +++ b/sandboxd/mesh/mesh.go @@ -105,9 +105,8 @@ func (m *Mesh) Join(seeds []string) error { // UpdateSelf republishes this node's warm-pool counts and promoted-template // set, bumping the epoch so peers adopt the new view. An unchanged view is -// not republished — the periodic tick would otherwise gossip a fresh epoch -// every second for nothing. templates must arrive sorted: the unchanged -// compare is order-sensitive. +// not republished. templates must arrive sorted: the unchanged compare is +// order-sensitive. func (m *Mesh) UpdateSelf(ctx context.Context, pools map[string]int, templates []string) { m.updateMu.Lock() defer m.updateMu.Unlock() @@ -120,7 +119,7 @@ func (m *Mesh) UpdateSelf(ctx context.Context, pools map[string]int, templates [ m.mu.Unlock() // Persist the candidate before publishing it: memberlist gossips self the // instant it enters the view, so a crash before the write would strand peers - // on an epoch a backwards-clock restart can't beat. Hold old state on failure. + // on an epoch a backwards-clock restart can't beat. if err := m.persistEpoch(epoch); err != nil { log.WithFunc("mesh.UpdateSelf").Warnf(ctx, "persist epoch: %v", err) return @@ -185,8 +184,6 @@ func (m *Mesh) Candidates(keyHash string) []string { case 1: return []string{pool[0].addr} } - // Power-of-two-choices: sample two, order by warmer. This is load - // spreading, not security — a weak PRNG is the right tool. i := rand.IntN(len(pool)) //nolint:gosec // placement jitter, not crypto j := rand.IntN(len(pool) - 1) //nolint:gosec // placement jitter, not crypto if j >= i { @@ -247,7 +244,6 @@ func (m *Mesh) Shutdown() error { return m.ml.Shutdown() } -// persistEpoch durably records the epoch. func (m *Mesh) persistEpoch(epoch uint64) error { return storeEpoch(m.epochPath, epoch) } diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index 053be4ed..099f3b4e 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -32,9 +32,8 @@ var ( ) // Checkpoint captures a claimed sandbox's state under a fresh id; the source -// keeps running (a hibernated one is captured from its wake image). Branches -// clone that exact state, and a source can be checkpointed again — a tree. -// tenant attributes the record; empty means the operator (root). +// keeps running (a hibernated one is captured from its wake image). tenant +// attributes the record; empty means the operator (root). func (m *Manager) Checkpoint(ctx context.Context, id string, cred Cred, name, tenant string) (types.Checkpoint, error) { sb, ok := m.resolve(id, cred) if !ok { @@ -64,9 +63,8 @@ func (m *Manager) Checkpoint(ctx context.Context, id string, cred Cred, name, te // is the capability to branch. func (m *Manager) ClaimCheckpoint(ctx context.Context, ckptID string, ttl time.Duration, tenant string) (*types.Sandbox, error) { // Reject a bad or unknown id before recLock: a rejected id must not leave - // a lock-map entry (only a delete evicts one). Checkpoints are immutable, - // so this parse stands in for the fetched meta below. It precedes quota: - // a full node must still answer "not here", or the tiers never run. + // a lock-map entry (only a delete evicts one). It precedes quota: a full + // node must still answer "not here". ckpt, err := m.loadCheckpoint(ctx, ckptID) if err != nil { return nil, err @@ -126,10 +124,6 @@ func (m *Manager) Checkpoints(ctx context.Context, tenant string) ([]types.Check // broadcasts to peers when fleet-scoped so a healed copy does not outlive it. // A tenant may delete only its own records — anything else answers // ErrUnknownCheckpoint, never a hint the id exists; root deletes anything. -// Existence is checked under the record lock (heal broke the "local miss -// means truly gone" assumption), plus vetoIfHealPending for a heal whose -// transfer runs unlocked. Every exit evicts the lock entry (recDoneEvict) — -// a checkpoint id is effectively one-shot, so the map must not grow per call. func (m *Manager) DeleteCheckpoint(ctx context.Context, ckptID, tenant string, scope DeleteScope) error { // Reject a bad id before recLock: a rejected id must not leave a // lock-map entry. @@ -382,8 +376,7 @@ func (m *Manager) pinnedArchiveCks() map[string]struct{} { } // sweepExpiredCheckpoints ages out checkpoints older than the configured -// TTL; explicit deletes never wait for it. It runs detached from the Run -// loop, so the guard keeps a slow backend from stacking sweeps. +// TTL; explicit deletes never wait for it. It runs detached from the Run loop. func (m *Manager) sweepExpiredCheckpoints(ctx context.Context) { if !m.ckptSweeping.CompareAndSwap(false, true) { return @@ -425,7 +418,6 @@ func (m *Manager) deleteCkLocked(ctx context.Context, ckID string) error { return nil } -// loadCheckpoint reads and parses a checkpoint's meta from the local store. func (m *Manager) loadCheckpoint(ctx context.Context, ckptID string) (types.Checkpoint, error) { if !store.CheckpointIDRe.MatchString(ckptID) { return types.Checkpoint{}, ErrUnknownCheckpoint diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index b9231c7b..0fbc7b15 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -15,8 +15,7 @@ import ( // Hibernate atomically snapshots a claimed sandbox and stops its VM, freeing // memory; the next agent access wakes it. Idempotent on an already-hibernated -// sandbox. When to hibernate is the caller's policy — the node only provides -// the transition. +// sandbox. func (m *Manager) Hibernate(ctx context.Context, id string, cred Cred) error { sb, ok := m.resolve(id, cred) if !ok { @@ -298,10 +297,8 @@ func (m *Manager) setPendingSnap(sb *types.Sandbox, snap string) claimSnapshot { // resolvePendingSnap settles a hibernate intent whose engine result was never // confirmed: the snapshot's presence decides whether to adopt it as -// HibernateSnap or clear the intent. Reports an adopted (completed but -// unrecorded) hibernate so the caller bills it; an unusable snapshot list -// keeps the intent and errors. No-op field read when nothing is pending. -// The caller holds sb.Transition. +// HibernateSnap or clear the intent. An unusable snapshot list keeps the +// intent and errors. The caller holds sb.Transition. func (m *Manager) resolvePendingSnap(ctx context.Context, sb *types.Sandbox) (adopted bool, err error) { if sb.PendingSnap == "" { return false, nil diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index fe333769..4d063ef5 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -272,9 +272,8 @@ type Manager struct { // maxClaims caps live claims node-wide (0 = unlimited); tenantMax holds // every configured tenant's cap (0 = unlimited) and doubles as the set of - // known tenants; tenantLive counts live claims per tenant so admission - // stays O(1). usage is the always-on billing event stream, audit the - // config-gated request tap. + // known tenants; tenantLive counts live claims per tenant. usage is the + // always-on billing event stream, audit the config-gated request tap. maxClaims int draining bool // guarded by m.mu; deliberately not persisted tenantMax map[string]int diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index bc548234..00d06d52 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -46,8 +46,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { switch { case sb.ArchiveCk != "": // Archived: no local VM by design; the store ck is the durable - // state. Adopt the stub so the id/token survive restart and the - // first exec wakes it (wakeArchived). + // state. Adopt the stub; the first exec wakes it (wakeArchived). m.claimed[id] = sb m.tenantDelta(sb.Tenant, 1) continue @@ -65,8 +64,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { // Stopped under a journaled hibernate intent whose commit never // landed: the intent names the wake image, adopt it. A verified- // missing image means the hibernate never completed — fall - // through and drop; an unverifiable list adopts (a failed wake - // beats a destroyed claim). + // through and drop; an unverifiable list adopts. sb.HibernateSnap, sb.PendingSnap = sb.PendingSnap, "" default: continue diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index 191d10f1..6818daf4 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -33,9 +33,9 @@ func (m *Manager) refillOnce(ctx context.Context) { return } now := time.Now() - // Spawning VMs into a full node is what turned a capacity limit into an - // outage: every doomed attempt still takes the global rtnl lock. The park - // gates spawns only — sweeping removed pools is pure bookkeeping. + // Spawning VMs into a full node still takes the global rtnl lock even when + // doomed to fail. The park gates spawns only — sweeping removed pools is + // pure bookkeeping. parked := now.Before(m.atCapacityUntil) inFlight := 0 for key, p := range m.pools { @@ -144,7 +144,7 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string) { } // noteCapacityLocked parks refill node-wide for a capacity failure, reporting -// whether this call parked it so the episode is logged once. m.mu held. +// whether this call parked it. m.mu held. func (m *Manager) noteCapacityLocked(now time.Time, reason string) bool { first := !now.Before(m.atCapacityUntil) m.atCapacityUntil = now.Add(capacityBackoff) @@ -424,7 +424,6 @@ func (m *Manager) vsockOf(ctx context.Context, name string) (string, error) { return vm.VsockSocket, nil } -// findVM returns the engine's record for name, if it still lists one. func (m *Manager) findVM(ctx context.Context, name string) (types.VMRecord, bool, error) { vms, err := m.eng.List(ctx, name) if err != nil { diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index 35b70527..407b1573 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -28,7 +28,6 @@ type templateRecord struct { // Promote publishes a claimed sandbox as a template under (template, parent // net, parent size); later claims for that key clone from it. Re-promoting the // same name replaces it, and the caller owns its lifecycle (DeleteTemplate). -// tenant attributes the record; empty means the operator (root). func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, tenant string) (types.PoolKey, string, error) { sb, ok := m.resolve(id, cred) if !ok { diff --git a/sandboxd/server/metrics.go b/sandboxd/server/metrics.go index ed41e599..638321b3 100644 --- a/sandboxd/server/metrics.go +++ b/sandboxd/server/metrics.go @@ -16,7 +16,7 @@ type SandboxListResponse struct { // handleMetrics renders Prometheus text format by hand — counters and // gauges only, no client library. Latency rides as *_seconds_total next to -// its count, so dashboards derive averages without histogram machinery. +// its count. func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { pools, g := s.mgr.Info() c := s.mgr.Counters() diff --git a/sandboxd/server/relay.go b/sandboxd/server/relay.go index afc9714d..73c10367 100644 --- a/sandboxd/server/relay.go +++ b/sandboxd/server/relay.go @@ -148,8 +148,7 @@ type auditTee struct { // WriteTo hands the splice back to io.Copy's fast path once the first line // is captured: it reads through Read until the record fires, then delegates -// the rest of the stream to a direct copy — an audit-enabled relay only pays -// the tee for the request frame, not the payload. +// the rest of the stream to a direct copy. func (t *auditTee) WriteTo(w io.Writer) (int64, error) { var total int64 buf := make([]byte, 4096) diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index da6bd581..fd7c4f36 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -267,7 +267,7 @@ func (s *Server) redirectClaim(ctx context.Context, w http.ResponseWriter, req t // aggregated/control-plane teardown works without holding the per-sandbox token; // a per-sandbox token releases only its own claim, unchanged. A tenant token is // neither — it is not the root api_token, so it resolves as a (non-matching) -// sandbox token and 404s. Tenants never get operator release. +// sandbox token and 404s. func (s *Server) handleRelease(w http.ResponseWriter, r *http.Request) { token, ok := sandboxToken(w, r) if !ok { diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index 765ffca8..48836c2e 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -51,9 +51,7 @@ var _ store.Store = (*Store)(nil) // Store keeps records as //{meta.json,export-/...}: the // generation dir is immutable and the meta rename is the atomic commit -// pointer, so readers and writers need no cross-process locks. idRe names -// the instance's id namespace — two instances (checkpoints, templates) -// share a root without seeing each other's records. +// pointer, so readers and writers need no cross-process locks. type Store struct { root string idRe *regexp.Regexp diff --git a/sandboxd/store/peer/peer.go b/sandboxd/store/peer/peer.go index a439618a..37988245 100644 --- a/sandboxd/store/peer/peer.go +++ b/sandboxd/store/peer/peer.go @@ -19,7 +19,7 @@ const healBudget = 30 * time.Minute type Owners func(id string) []string // Validate checks a staged pull before Pull trusts the owner that sent it; an -// error tries the next owner. The healer does not know the record's shape. +// error tries the next owner. type Validate func(staging string) error // Healer pulls a record this node does not hold into a caller-provided diff --git a/sandboxd/store/peer/probe.go b/sandboxd/store/peer/probe.go index 57527762..1fc9b419 100644 --- a/sandboxd/store/peer/probe.go +++ b/sandboxd/store/peer/probe.go @@ -70,10 +70,9 @@ type HTTPProber struct { } // Owners fans a HEAD out to every peer and returns up to maxRedirectOwners -// addresses that answered 200. Concurrent calls for one id share a fan-out; -// a short positive cache serves a hot id's repeat redirects. Cache staleness -// fails safely: a redirect to a peer that just deleted the record 404s, and -// the client's no_redirect fallback heals from the origin. +// addresses that answered 200. Cache staleness fails safely: a redirect to a +// peer that just deleted the record 404s, and the client's no_redirect +// fallback heals from the origin. func (p *HTTPProber) Owners(ctx context.Context, id string) []string { owners, start, hit := p.cacheLookup(id) if hit { diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index 2f7acf78..8a78d9aa 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -35,8 +35,7 @@ const ( recordTrailer = ".record-complete" ) -// ErrNotFound reports that a peer does not hold the requested record. Declared -// here so the transport does not import the store package it is a backend for. +// ErrNotFound reports that a peer does not hold the requested record. var ErrNotFound = errors.New("peer does not hold record") // Puller fetches a record from a peer into a local directory. diff --git a/sandboxd/types/api.go b/sandboxd/types/api.go index b56a4b4f..43ae7202 100644 --- a/sandboxd/types/api.go +++ b/sandboxd/types/api.go @@ -27,7 +27,7 @@ type ClaimRequest struct { NoRedirect bool `json:"no_redirect,omitempty"` // ClaimRef is an opaque caller reference (the aggregated apiserver passes // the k8s "/") recorded on the claim so the read path can - // map a listed sandbox back to the name it was claimed under. Optional. + // map a listed sandbox back to the name it was claimed under. ClaimRef string `json:"claim_ref,omitempty"` } @@ -58,9 +58,8 @@ type ClaimResponse struct { // ForkRequest is the wire body of POST /v1/sandboxes/{id}/fork. The // Authorization header carries an api or tenant token (forking creates node // resources, like a claim); Token proves ownership of the source sandbox. -// TTLSeconds applies to every child; zero means the server default — a -// lease is a per-sandbox resource bound, so children never inherit the -// parent's remainder. +// TTLSeconds applies to every child; zero means the server default, and +// children never inherit the parent's remainder. type ForkRequest struct { Token string `json:"token"` Count int `json:"count"` diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index 83dc5a40..0216d01e 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -75,8 +75,7 @@ func (e Engine) Validate() error { } } -// Size is a T-shirt resource tier. Free-form CPU/memory would fragment the -// warm pools, so only tiers are accepted. +// Size is a T-shirt resource tier. type Size string // SizeSpec is the concrete allocation behind a tier. Memory is in cocoon flag diff --git a/sandboxd/utils/utils.go b/sandboxd/utils/utils.go index df1a0a6d..4594215f 100644 --- a/sandboxd/utils/utils.go +++ b/sandboxd/utils/utils.go @@ -60,8 +60,7 @@ func WriteFileSync(path string, data []byte, perm os.FileMode) error { // DecodeStrictJSON decodes one JSON value into v, refusing unknown fields, // duplicated keys, and trailing data — for hand-edited operator input, where -// a typo must fail instead of silently changing what was configured (an -// unknown or repeated key is dropped or last-wins under a lenient decode). +// a typo must fail instead of silently changing what was configured. func DecodeStrictJSON(raw []byte, v any) error { if err := rejectDuplicateKeys(json.NewDecoder(bytes.NewReader(raw))); err != nil { return err diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index 3e3d28af..bca4c039 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -51,7 +51,7 @@ func (ck *Checkpoint) New(ctx context.Context, opts ...Option) (*Sandbox, error) // (offline, partitioned, or joined later) keeps serving branches from its // replica until the node's checkpoint_ttl_hours ages it out; with that TTL // at its default of 0 (keep forever), an unreachable peer's replica has no -// cleanup bound at all. See the cluster docs' placement lifecycle. +// cleanup bound at all. func (ck *Checkpoint) Delete(ctx context.Context) error { return doNoContent(ctx, ck.c, http.MethodDelete, ck.addr, "/v1/checkpoints/"+ck.ID, nil, ck.c.apiToken, "delete checkpoint") } diff --git a/sdk/go/client.go b/sdk/go/client.go index 637531f9..bdf2dd96 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -43,11 +43,10 @@ type Client struct { // New claims a sandbox for template. Without options the node serves its // defaults: the no-network lane and the smallest size tier. New returns when -// the sandbox's silkd is reachable; a warm pool hit is milliseconds, a cold -// key can take the full boot. Against a cluster, a warm miss redirects to a -// peer that holds one, which New follows transparently; if every candidate -// fails transiently (full, mid-heal, unreachable), New falls back to the -// origin once so it provisions or heals locally. +// the sandbox's silkd is reachable. Against a cluster, a warm miss redirects +// to a peer that holds one, which New follows transparently; if every +// candidate fails transiently (full, mid-heal, unreachable), New falls back +// to the origin once so it provisions or heals locally. func (c *Client) New(ctx context.Context, template string, opts ...Option) (*Sandbox, error) { claim := claimRequest{Template: template} for _, opt := range opts { @@ -84,9 +83,8 @@ func (c *Client) Lookup(ctx context.Context, id, token string) (*Sandbox, error) // Attach binds a handle to an already-claimed sandbox whose owner data-plane // address is already known (e.g. delivered by the L3 apiserver as annotations), -// with no lookup round-trip. Use it to exec/agent into a sandbox claimed out of -// band. ownerAddr is the node's sandboxd data-plane address; token is the -// per-sandbox credential. +// with no lookup round-trip. ownerAddr is the node's sandboxd data-plane +// address; token is the per-sandbox credential. func (c *Client) Attach(ownerAddr, id, token string) *Sandbox { return &Sandbox{ID: id, token: token, c: c, owner: ownerAddr} } @@ -95,8 +93,6 @@ func (c *Client) Attach(ownerAddr, id, token string) *Sandbox { // does not hold it but the mesh's gossip names an owner, the delete follows // the redirect there (one hop); gossip lags a fresh promote by about a tick, // so right after promoting prefer Template.Delete on the returned handle. -// The options pick the template's key axes exactly as a claim would -// (network lane, size); the same defaults apply. func (c *Client) DeleteTemplate(ctx context.Context, template string, opts ...Option) error { claim := claimRequest{Template: template} for _, opt := range opts { @@ -290,15 +286,6 @@ func retryTransient(err error) bool { } } -// redirectFallback walks candidates via claimAt, retrying broadly (retryAny) -// so one wrong candidate doesn't cost a candidate that would still succeed. -// If every candidate is exhausted and the last failure was transient -// (retryTransient), it gives origin one more no_redirect attempt — the node -// that issued the redirect provisions or heals locally instead of leaving -// the claim stuck on stale gossip. A definitive last failure (a bad request, -// an auth rejection, a conflict) skips the fallback: origin would fail the -// same way. A second-level redirect (a compliant server never sends one -// once no_redirect is set) fails the candidate rather than being followed. // claimFollow runs the claim protocol from origin: claim there, and on a // redirect re-encode with no_redirect and follow via redirectFallback. Only // the fallback error carries the verb — first-contact errors return raw. @@ -326,6 +313,14 @@ func claimFollow(origin, verb string, encode func(noRedirect bool) ([]byte, erro return addr, target, nil } +// redirectFallback walks candidates via claimAt, retrying broadly (retryAny). +// If every candidate is exhausted and the last failure was transient +// (retryTransient), it gives origin one more no_redirect attempt — the node +// that issued the redirect provisions or heals locally instead of leaving +// the claim stuck on stale gossip. A definitive last failure (a bad request, +// an auth rejection, a conflict) skips the fallback: origin would fail the +// same way. A second-level redirect (a compliant server never sends one +// once no_redirect is set) fails the candidate rather than being followed. func redirectFallback(origin string, candidates []string, claimAt func(addr string) (claimResponse, error)) (string, claimResponse, error) { claimNoRedirect := func(target string) (claimResponse, error) { cr, err := claimAt(target) diff --git a/sdk/go/port.go b/sdk/go/port.go index 8d5bf525..b239f86b 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -23,8 +23,7 @@ var _ net.Conn = (*PortConn)(nil) // PortConn is a net.Conn to a TCP port inside the sandbox, relayed over the // silkd protocol (works on the no-network lane — the vsock relay is its only // transport). Read returns io.EOF when the guest server closes; CloseWrite -// half-closes the guest socket. Deadlines are not supported (bound the -// DialPort ctx instead). +// half-closes the guest socket. type PortConn struct { conn *silkd.Conn stop func() diff --git a/sdk/go/pty.go b/sdk/go/pty.go index 6aa9cf2d..c340ce62 100644 --- a/sdk/go/pty.go +++ b/sdk/go/pty.go @@ -18,9 +18,7 @@ type PtyOpts struct { User string } -// Pty is an open pseudo-terminal in the sandbox. Read yields terminal output -// (EOF when the shell exits), Write feeds input, Resize adjusts the window, -// and Close (or canceling the OpenPty ctx) ends the session. +// Pty is an open pseudo-terminal in the sandbox. type Pty struct { PID uint32 diff --git a/sdk/go/silkd/conn.go b/sdk/go/silkd/conn.go index 5e3ed7aa..5ff9c63c 100644 --- a/sdk/go/silkd/conn.go +++ b/sdk/go/silkd/conn.go @@ -58,8 +58,7 @@ func (c *Conn) Close() error { return c.rwc.Close() } -// sendBulk renders a data/stdin frame into the reused buffer, skipping the -// json.Marshal alloc + escape rescan. +// sendBulk renders a data/stdin frame into the reused buffer. func (c *Conn) sendBulk(op string, payload []byte) error { c.wmu.Lock() defer c.wmu.Unlock()