amadan

carlos / platform Public

release: stream the artifact to disk while staging (#269)

Staging read the whole build into memory (GetQualified -> io.ReadAll),
hashed it, then wrote it. Under the root stage pass's MemoryMax=256M a
~99 MB build peaked at the cap and the pass was killed, so a console
promote waited for the next hourly adopt pass.

stageFrom now opens the artifact with OpenQualified (OpenObject plus
GetQualified's migration hint) and stageArtifactFrom copies it into the
temp file beside the destination while hashing. Only bytes that hash to
the manifest's sha256 are renamed into place; a mismatch or a failed
read removes the temp file and leaves the staged copy alone, as before.
The version dir is checked before the build is opened, so a sandboxed
agent still fails on the mkdir and relays to root without downloading.
Human

Prompt

File and fix: staging a release reads the whole build into memory, so a
~99 MB build kills the 256 MB stage pass.

From this commit's own message. The prompt ledger has no entries for it — which is what a squashed, rebased or amended commit looks like, and what every commit looks like in a repo without a ledger.

Agent notes
A failed download can now leave an empty version dir for a fresh
version; sweep.go's treeStats already collects empty husks. The edge's
static-site extract and vet still read artifacts whole (out of scope,
noted on #269). The legacy tripwire gained Open so it watches the
streamed read too.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9b022aea041b3b63e51140ca73b763475c644f36 parent abcc732
5 files changed, +274 −15
  • internal/release/adopt.go +6 −7
  • internal/release/qualified.go +12 −0
  • internal/release/qualified_test.go +8 −0
  • internal/release/stage_stream_test.go +169 −0
  • internal/release/versions.go +79 −8

CI cancelled — superseded by a newer push — run details

diff --git a/internal/release/adopt.go b/internal/release/adopt.go
index 7078040b3..f7aafa322 100644
--- a/internal/release/adopt.go
+++ b/internal/release/adopt.go
@@ -1854,14 +1854,13 @@ func stageFrom(ctx context.Context, s Store, optDir, acct, app, ver, host string
// catches up.
return nil, fmt.Errorf("release %s has no %s build — skipping", ver, adoptPlatform)
}
- body, _, err := GetQualified(ctx, s, ArtifactKey(acct, app, ver, art.File))
+ // Streamed to disk and hashed on the way (#269): a build read whole
+ // outgrew the stage pass's memory cap.
+ err := stageArtifactFrom(optDir, acct, ver, app, art.SHA256, func() (io.ReadCloser, error) {
+ rc, _, err := OpenQualified(ctx, s, ArtifactKey(acct, app, ver, art.File))
+ return rc, err
+ })
if err != nil {
- return nil, fmt.Errorf("download %s: %w", ver, err)
- }
- if got := etagOf(body); got != art.SHA256 {
- return nil, fmt.Errorf("download %s: sha256 mismatch (refusing to stage)", ver)
- }
- if err := StageArtifact(optDir, acct, ver, app, body); err != nil {
return nil, err
}
if chk.verified != nil {
diff --git a/internal/release/qualified.go b/internal/release/qualified.go
index cb5c3a076..72f71f2ca 100644
--- a/internal/release/qualified.go
+++ b/internal/release/qualified.go
@@ -26,6 +26,7 @@ import (
"errors"
"fmt"
"github.com/carlosframework/platform/internal/controlname"
+ "io"
"os"
"strings"
)
@@ -88,6 +89,17 @@ func GetQualified(ctx context.Context, s Store, key string) (body []byte, etag s
return body, etag, err
}
+// OpenQualified is GetQualified for a streaming read (OpenObject): the same
+// key, the same hint on a miss. Staging reads a build this way (#269) so a
+// large one never sits in memory whole.
+func OpenQualified(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) {
+ rc, etag, err := OpenObject(ctx, s, key)
+ if err != nil && errors.Is(err, os.ErrNotExist) {
+ return nil, "", fmt.Errorf("%w%s", err, migrationHint(key))
+ }
+ return rc, etag, err
+}
+
// migrationHint turns a miss on an account-qualified APP key into the exact
// command that fixes it, for the one deployment shape that produces a
// surprising 404: upgraded from a pre-accounts binary, objects still at
diff --git a/internal/release/qualified_test.go b/internal/release/qualified_test.go
index 755ca68ca..1419154c9 100644
--- a/internal/release/qualified_test.go
+++ b/internal/release/qualified_test.go
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "io"
"os"
"strings"
"testing"
@@ -50,6 +51,13 @@ func (l *legacyTripwire) Get(ctx context.Context, key string) ([]byte, string, e
return l.Store.Get(ctx, key)
}
+// Open is tripped too: staging streams the artifact (#269), and without this
+// a streamed read would reach the store under the tripwire unwatched.
+func (l *legacyTripwire) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ l.tripped("read", key)
+ return OpenObject(ctx, l.Store, key)
+}
+
func (l *legacyTripwire) List(ctx context.Context, prefix string) ([]string, error) {
l.tripped("listed", prefix)
return l.Store.List(ctx, prefix)
diff --git a/internal/release/stage_stream_test.go b/internal/release/stage_stream_test.go
new file mode 100644
index 000000000..f5c60fb4d
--- /dev/null
+++ b/internal/release/stage_stream_test.go
@@ -0,0 +1,169 @@
+package release
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "io"
+ "os"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "testing"
+)
+
+// wholeArtifactRefuser fails the test if anything reads a release artifact
+// whole through Get. Manifests and pointers are small and still come through
+// Get; a build has to be streamed through Open (#269).
+type wholeArtifactRefuser struct {
+ Store
+ t *testing.T
+}
+
+func (w wholeArtifactRefuser) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.Contains(key, "/releases/") && !strings.HasSuffix(key, "manifest.json") {
+ w.t.Errorf("artifact %s was read whole into memory; staging must stream it", key)
+ }
+ return w.Store.Get(ctx, key)
+}
+
+func (w wholeArtifactRefuser) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ return OpenObject(ctx, w.Store, key)
+}
+
+func TestStagingStreamsTheArtifact(t *testing.T) {
+ dir := NewDirStore(t.TempDir())
+ ctx := context.Background()
+ opt := t.TempDir()
+ body := []byte("the whole build")
+ shipRelease(t, dir, "bab", "km", "v1", adoptPlatform, body)
+
+ if _, err := ensureStaged(ctx, wholeArtifactRefuser{Store: dir, t: t}, opt, "bab", "km", "v1", "h.example.com"); err != nil {
+ t.Fatalf("ensureStaged: %v", err)
+ }
+ got, err := os.ReadFile(VersionPath(opt, "bab", "v1", "km"))
+ if err != nil || !bytes.Equal(got, body) {
+ t.Fatalf("staged %q (err %v), want %q", got, err, body)
+ }
+}
+
+// patternReader yields n bytes of a fixed pattern without holding them.
+type patternReader struct{ n int64 }
+
+func (p *patternReader) Read(b []byte) (int, error) {
+ if p.n <= 0 {
+ return 0, io.EOF
+ }
+ if int64(len(b)) > p.n {
+ b = b[:p.n]
+ }
+ for i := range b {
+ b[i] = byte(i * 31)
+ }
+ p.n -= int64(len(b))
+ return len(b), nil
+}
+
+// The regression itself: a ~99 MB build read whole peaked past the stage
+// pass's 256 MB cap. Streamed, staging a build allocates a copy buffer, not
+// the build.
+func TestStageArtifactFromKeepsMemoryFlat(t *testing.T) {
+ if testing.Short() {
+ t.Skip("writes a 96 MB file")
+ }
+ const size = 96 << 20
+ h := sha256.New()
+ if _, err := io.Copy(h, &patternReader{n: size}); err != nil {
+ t.Fatal(err)
+ }
+ want := hex.EncodeToString(h.Sum(nil))
+ opt := t.TempDir()
+
+ runtime.GC()
+ var before, after runtime.MemStats
+ runtime.ReadMemStats(&before)
+ err := stageArtifactFrom(opt, "bab", "v1", "big", want, func() (io.ReadCloser, error) {
+ return io.NopCloser(&patternReader{n: size}), nil
+ })
+ runtime.ReadMemStats(&after)
+ if err != nil {
+ t.Fatalf("stageArtifactFrom: %v", err)
+ }
+ if grew := after.TotalAlloc - before.TotalAlloc; grew > 8<<20 {
+ t.Errorf("staging a %d MB build allocated %d MB; it must stream", size>>20, grew>>20)
+ }
+ if fi, err := os.Stat(VersionPath(opt, "bab", "v1", "big")); err != nil || fi.Size() != size || fi.Mode().Perm() != 0o755 {
+ t.Fatalf("staged file: %v, err %v", fi, err)
+ }
+}
+
+// failingReader yields some bytes, then fails, like a connection dropped
+// partway through a download.
+type failingReader struct{ sent bool }
+
+func (f *failingReader) Read(b []byte) (int, error) {
+ if f.sent {
+ return 0, errors.New("connection reset")
+ }
+ f.sent = true
+ return copy(b, "half a build"), nil
+}
+
+// Bytes that fail to arrive whole, or arrive wrong, never replace what was
+// staged under the name and leave no temp file behind, exactly as verifying
+// in memory first did.
+func TestStageArtifactFromLeavesTheStagedCopyOnFailure(t *testing.T) {
+ good := []byte("new build")
+ sum := sha256.Sum256(good)
+ for _, tc := range []struct {
+ name string
+ want string
+ r func() io.Reader
+ mismatch bool
+ }{
+ {"wrong bytes", hex.EncodeToString(sum[:]), func() io.Reader { return strings.NewReader("not the build") }, true},
+ {"read fails partway", hex.EncodeToString(sum[:]), func() io.Reader { return &failingReader{} }, false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ opt := t.TempDir()
+ if err := StageArtifact(opt, "bab", "v1", "km", []byte("old build")); err != nil {
+ t.Fatal(err)
+ }
+ err := stageArtifactFrom(opt, "bab", "v1", "km", tc.want, func() (io.ReadCloser, error) {
+ return io.NopCloser(tc.r()), nil
+ })
+ if err == nil || !strings.Contains(err.Error(), "download v1") {
+ t.Fatalf("err = %v, want a download error", err)
+ }
+ if errors.Is(err, errArtifactMismatch) != tc.mismatch {
+ t.Errorf("errors.Is(err, errArtifactMismatch) = %v, want %v", !tc.mismatch, tc.mismatch)
+ }
+ path := VersionPath(opt, "bab", "v1", "km")
+ if got, _ := os.ReadFile(path); string(got) != "old build" {
+ t.Errorf("staged copy is %q, want the old build untouched", got)
+ }
+ if _, err := os.Stat(path + ".tmp"); !errors.Is(err, os.ErrNotExist) {
+ t.Errorf("temp file left behind: %v", err)
+ }
+ })
+ }
+}
+
+// A box that cannot write the version store (the sandboxed agent) fails on
+// the directory before it opens the build, so it never downloads one only to
+// relay the stage to root.
+func TestStageArtifactFromChecksTheStoreBeforeOpening(t *testing.T) {
+ opt := filepath.Join(t.TempDir(), "not-a-dir")
+ if err := os.WriteFile(opt, nil, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ err := stageArtifactFrom(opt, "bab", "v1", "km", "", func() (io.ReadCloser, error) {
+ t.Error("the build was opened though the version store cannot be written")
+ return io.NopCloser(strings.NewReader("")), nil
+ })
+ if err == nil {
+ t.Fatal("staging into an unwritable store succeeded")
+ }
+}
diff --git a/internal/release/versions.go b/internal/release/versions.go
index eea8b9bbd..82941dede 100644
--- a/internal/release/versions.go
+++ b/internal/release/versions.go
@@ -14,7 +14,10 @@ package release
import (
"context"
+ "crypto/sha256"
+ "encoding/hex"
"encoding/json"
+ "errors"
"fmt"
"io"
"net"
@@ -126,28 +129,96 @@ func WriteFileAtomic(dst string, data []byte, mode os.FileMode) error {
// StageArtifact puts verified artifact bytes into the store under ver, so
// hosts can be pointed at it. Idempotent: re-staging overwrites.
func StageArtifact(optDir, acct, ver, app string, data []byte) error {
+ dst, err := stagePath(optDir, acct, ver, app)
+ if err != nil {
+ return err
+ }
+ return WriteFileAtomic(dst, data, 0o755)
+}
+
+// errArtifactMismatch is stageArtifactFrom refusing bytes that do not hash
+// to the manifest's sha256. Nothing is left behind when it is returned.
+var errArtifactMismatch = errors.New("sha256 mismatch")
+
+// stageArtifactFrom is StageArtifact for a build read from the store as a
+// stream (#269). The bytes go straight into the temp file beside the
+// destination and are hashed on the way, so memory stays flat whatever the
+// build's size: the root stage pass runs under MemoryMax=256M, and reading a
+// ~99 MB build whole peaked past it. Only bytes that hash to wantSHA are
+// moved into place. A mismatch or a failed read removes the temp file and
+// leaves whatever was staged under the name untouched, exactly as verifying
+// in memory first did.
+//
+// open is called only once the version dir is writable, so a sandboxed
+// agent fails on the mkdir (stageBlocked relays that to root) without
+// opening the build at all.
+func stageArtifactFrom(optDir, acct, ver, app, wantSHA string, open func() (io.ReadCloser, error)) error {
+ dst, err := stagePath(optDir, acct, ver, app)
+ if err != nil {
+ return err
+ }
+ in, err := open()
+ if err != nil {
+ return fmt.Errorf("download %s: %w", ver, err)
+ }
+ defer in.Close()
+ tmp := dst + ".tmp"
+ out, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o755)
+ if err != nil {
+ return err
+ }
+ h := sha256.New()
+ if _, err := io.Copy(io.MultiWriter(out, h), in); err != nil {
+ out.Close()
+ os.Remove(tmp)
+ return fmt.Errorf("download %s: %w", ver, err)
+ }
+ if err := out.Close(); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ if hex.EncodeToString(h.Sum(nil)) != wantSHA {
+ os.Remove(tmp)
+ return fmt.Errorf("download %s: %w (refusing to stage)", ver, errArtifactMismatch)
+ }
+ // O_CREATE's mode is masked by umask; set it explicitly so the binary is +x.
+ if err := os.Chmod(tmp, 0o755); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ if err := os.Rename(tmp, dst); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ return nil
+}
+
+// stagePath validates the names, makes the version dir traversable, and
+// returns where the build goes.
+func stagePath(optDir, acct, ver, app string) (string, error) {
if !ValidVersion(ver) {
- return fmt.Errorf("invalid version %q (allowed: letters, digits, . _ -)", ver)
+ return "", fmt.Errorf("invalid version %q (allowed: letters, digits, . _ -)", ver)
}
if !ValidApp(app) {
- return fmt.Errorf("invalid app %q", app)
+ return "", fmt.Errorf("invalid app %q", app)
}
if acct != "" && !ValidAccount(acct) {
- return fmt.Errorf("invalid account %q", acct)
+ return "", fmt.Errorf("invalid account %q", acct)
}
- dir := filepath.Dir(VersionPath(optDir, acct, ver, app))
+ dst := VersionPath(optDir, acct, ver, app)
+ dir := filepath.Dir(dst)
if err := os.MkdirAll(dir, 0o755); err != nil {
- return err
+ return "", err
}
// MkdirAll's mode is masked by the caller's umask, so a tight operator
// umask (e.g. 077 from a hardened shell) leaves the version dir 0700 —
// the sandboxed service user then can't traverse in to exec its binary
// (203/EXEC "Permission denied", found staging a canary on Eleven).
- // Force it, like the write below forces the binary's +x.
+ // Force it, like the write forces the binary's +x.
if err := os.Chmod(dir, 0o755); err != nil {
- return err
+ return "", err
}
- return WriteFileAtomic(VersionPath(optDir, acct, ver, app), data, 0o755)
+ return dst, nil
}
// StagedManifestPath is the staged binary's SIBLING file: