amadan

carlos / platform Public

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

Staging read the whole build into memory, so a ~99 MB build ran the root
stage pass (MemoryMax=256M) out of memory and a console promote waited for
the hourly adopt pass. stageArtifactFrom now streams the artifact through
OpenQualified into a per-call temp file beside the destination, hashing as
it goes, and renames only bytes matching the manifest's sha256. A mismatch
or failed read leaves the staged copy untouched. Astra: round 1 REJECT (a
shared temp name let a concurrent writer's partial bytes be published),
fixed; round 2 ACCEPT.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Squashed from stage-stream-artifact into main.
claude cd987f6ca876572ee718566d48e3431f796fd8a4 parent a3bf7d6
5 files changed, +355 −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 +245 −0
  • internal/release/versions.go +84 −8

CI failed — 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..316dcd6a9
--- /dev/null
+++ b/internal/release/stage_stream_test.go
@@ -0,0 +1,245 @@
+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 left, _ := filepath.Glob(path + ".tmp*"); len(left) != 0 {
+ t.Errorf("temp files left behind: %v", left)
+ }
+ })
+ }
+}
+
+// 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")
+ }
+}
+
+// Two writers of one version overlap (Astra, stage-stream-artifact review):
+// A has written and hashed the whole build when B, a download that later
+// fails, starts writing. With a shared temp file B's truncate-and-write lands
+// in A's file before A renames it, and A publishes B's partial bytes under a
+// passing hash. Each writer's own temp file keeps A's bytes A's.
+func TestStageArtifactFromOverlappingWriters(t *testing.T) {
+ opt := t.TempDir()
+ good := []byte("the good build, whole")
+ sum := sha256.Sum256(good)
+ want := hex.EncodeToString(sum[:])
+
+ bWrote, release, bDone := make(chan struct{}), make(chan struct{}), make(chan error, 1)
+ a := &overlapReader{body: good, atEOF: func() {
+ go func() {
+ bDone <- stageArtifactFrom(opt, "bab", "v1", "km", want, func() (io.ReadCloser, error) {
+ return io.NopCloser(&stallingReader{wrote: bWrote, release: release}), nil
+ })
+ }()
+ <-bWrote
+ }}
+ if err := stageArtifactFrom(opt, "bab", "v1", "km", want, func() (io.ReadCloser, error) {
+ return io.NopCloser(a), nil
+ }); err != nil {
+ t.Fatalf("writer A: %v", err)
+ }
+ close(release)
+ if err := <-bDone; err == nil {
+ t.Fatal("writer B's failed download reported success")
+ }
+ path := VersionPath(opt, "bab", "v1", "km")
+ if got, _ := os.ReadFile(path); !bytes.Equal(got, good) {
+ t.Fatalf("staged %q, want writer A's verified bytes %q", got, good)
+ }
+ if left, _ := filepath.Glob(path + ".tmp*"); len(left) != 0 {
+ t.Errorf("temp files left behind: %v", left)
+ }
+}
+
+// overlapReader yields body, then runs atEOF once before reporting EOF.
+type overlapReader struct {
+ body []byte
+ atEOF func()
+}
+
+func (o *overlapReader) Read(b []byte) (int, error) {
+ if len(o.body) > 0 {
+ n := copy(b, o.body)
+ o.body = o.body[n:]
+ return n, nil
+ }
+ if o.atEOF != nil {
+ o.atEOF()
+ o.atEOF = nil
+ }
+ return 0, io.EOF
+}
+
+// stallingReader yields a partial build, signals once it has been written
+// (the second Read only comes after the first chunk's write), then waits for
+// release and fails.
+type stallingReader struct {
+ n int
+ wrote chan struct{}
+ release chan struct{}
+}
+
+func (s *stallingReader) Read(b []byte) (int, error) {
+ s.n++
+ if s.n == 1 {
+ return copy(b, "partial"), nil
+ }
+ close(s.wrote)
+ <-s.release
+ return 0, errors.New("connection reset")
+}
diff --git a/internal/release/versions.go b/internal/release/versions.go
index eea8b9bbd..8f44118e0 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,101 @@ 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()
+ // A temp file of its own, not WriteFileAtomic's shared dst+".tmp": a
+ // download is long enough for a concurrent writer of the same version (an
+ // unserialized pass, the agent) to truncate a shared temp file between
+ // this call's hash and its rename, which would publish the other writer's
+ // partial bytes under a passing hash.
+ out, err := os.CreateTemp(filepath.Dir(dst), filepath.Base(dst)+".tmp-*")
+ if err != nil {
+ return err
+ }
+ tmp := out.Name()
+ 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)
+ }
+ // CreateTemp makes the file 0600; the binary must be 0755 to exec.
+ 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: