amadan
release: each staging download gets its own temp file (#269)
Astra (round 1, P1): every writer shared dst+".tmp" with O_TRUNC. A concurrent writer of the same version (an unserialized pass, the agent) could truncate it between one writer's hash and its rename, publishing partial bytes under a passing hash. Streaming stretched that window to the whole download. stageArtifactFrom now writes to os.CreateTemp beside the destination and renames only its own file. TestStageArtifactFromOverlappingWriters overlaps a verified writer with a failing one deterministically; with a shared temp name it stages the partial bytes.
Agent notes
A pass killed mid-download now leaves <app>.tmp-<n> rather than overwriting one shared name next time; the version dir's sweep takes it with the dir.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+86
−5
internal/release/stage_stream_test.go+78 −2internal/release/versions.go+8 −3
CI passed — run details
diff --git a/internal/release/stage_stream_test.go b/internal/release/stage_stream_test.go| index f5c60fb4d..316dcd6a9 100644 |
| --- a/internal/release/stage_stream_test.go |
| +++ b/internal/release/stage_stream_test.go |
| @@ -144,8 +144,8 @@ func TestStageArtifactFromLeavesTheStagedCopyOnFailure(t *testing.T) { |
| 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) |
| + if left, _ := filepath.Glob(path + ".tmp*"); len(left) != 0 { |
| + t.Errorf("temp files left behind: %v", left) |
| } |
| }) |
| } |
| @@ -167,3 +167,79 @@ func TestStageArtifactFromChecksTheStoreBeforeOpening(t *testing.T) { |
| 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 82941dede..8f44118e0 100644 |
| --- a/internal/release/versions.go |
| +++ b/internal/release/versions.go |
| @@ -162,11 +162,16 @@ func stageArtifactFrom(optDir, acct, ver, app, wantSHA string, open func() (io.R |
| 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) |
| + // 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() |
| @@ -181,7 +186,7 @@ func stageArtifactFrom(optDir, acct, ver, app, wantSHA string, open func() (io.R |
| 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. |
| + // 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 |