amadan

carlos / platform Public

runs: the chunker: ordered, capped, durable output chunks (run spec §5.5, codex 5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 449981af1e63a463d58b39acbc4d04286dff5f33 parent a188a2b
2 files changed, +363 −0
  • internal/runs/chunker.go +178 −0
  • internal/runs/chunker_test.go +185 −0
diff --git a/internal/runs/chunker.go b/internal/runs/chunker.go
new file mode 100644
index 000000000..57b59c77d
--- /dev/null
+++ b/internal/runs/chunker.go
@@ -0,0 +1,178 @@
+package runs
+
+import (
+ "context"
+ "errors"
+ "io"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// errBadStream is what Close reports when a writer was made for a stream
+// that is neither StreamOut nor StreamErr. ChunkKey trusts its stream, so the
+// chunker refuses anything else before it reaches a key.
+var errBadStream = errors.New("runs: chunker stream is not out or err")
+
+// Chunker turns a run's two streams into chunks (run spec §5.5):
+// out/<seq>-<o|e>, one seq for both streams in the order the bytes were
+// read. A chunk holds one stream's bytes: a write on the other stream cuts
+// it, so the follower interleaves exactly as the agent read. A chunk is
+// written every second, or as soon as it holds 256 KB. Past 64 MB in total
+// nothing more is written and Close says truncated; writes never fail, so
+// the command is never blocked by a refused pipe. Each chunk is a
+// create-only Put retried until it lands or the context ends, under the
+// lock, so seqs are written in order and a slow bucket slows the copy.
+type Chunker struct {
+ ctx context.Context
+ s release.Store
+ acct, app, id string
+ every time.Duration
+ max int64
+
+ mu sync.Mutex
+ stream string
+ pending []byte
+ seq int64
+ total int64
+ truncated bool
+ err error
+ closed bool
+
+ stop chan struct{}
+ done chan struct{}
+ closeOnce sync.Once
+}
+
+// NewChunker starts a chunker for run id with the spec's cadence and cap.
+// An account, app or run id the store helpers would refuse (ErrBadApp,
+// ErrBadID) gives a chunker that writes nothing and whose Close returns that
+// error, so it can never write outside runs/<id>/out/.
+func NewChunker(ctx context.Context, s release.Store, acct, app, id string) *Chunker {
+ return newChunker(ctx, s, acct, app, id, ChunkEvery, MaxOutputBytes)
+}
+
+func newChunker(ctx context.Context, s release.Store, acct, app, id string, every time.Duration, max int64) *Chunker {
+ c := &Chunker{ctx: ctx, s: s, acct: acct, app: app, id: id, every: every, max: max,
+ stop: make(chan struct{}), done: make(chan struct{})}
+ if err := checkRun(acct, app, id); err != nil {
+ c.err, c.truncated = err, true
+ close(c.done)
+ return c
+ }
+ go c.tick()
+ return c
+}
+
+func (c *Chunker) tick() {
+ defer close(c.done)
+ t := time.NewTicker(c.every)
+ defer t.Stop()
+ for {
+ select {
+ case <-c.stop:
+ return
+ case <-t.C:
+ c.mu.Lock()
+ c.flushLocked()
+ c.mu.Unlock()
+ }
+ }
+}
+
+type chunkStream struct {
+ c *Chunker
+ stream string
+}
+
+func (w chunkStream) Write(p []byte) (int, error) {
+ w.c.write(w.stream, p)
+ return len(p), nil
+}
+
+// Stream is the writer for one stream, StreamOut or StreamErr. Its Write
+// never fails and never returns short.
+func (c *Chunker) Stream(stream string) io.Writer { return chunkStream{c: c, stream: stream} }
+
+func (c *Chunker) write(stream string, p []byte) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if c.closed {
+ return
+ }
+ if stream != StreamOut && stream != StreamErr {
+ if c.err == nil {
+ c.err, c.truncated, c.pending = errBadStream, true, nil
+ }
+ return
+ }
+ if c.err != nil {
+ c.truncated = c.truncated || len(p) > 0
+ return
+ }
+ if room := c.max - c.total; int64(len(p)) > room {
+ p = p[:room]
+ c.truncated = true
+ }
+ for len(p) > 0 {
+ if c.stream != stream && len(c.pending) > 0 {
+ c.flushLocked()
+ if c.err != nil {
+ return
+ }
+ }
+ c.stream = stream
+ n := min(len(p), ChunkBytes-len(c.pending))
+ c.pending = append(c.pending, p[:n]...)
+ c.total += int64(n)
+ p = p[n:]
+ if len(c.pending) >= ChunkBytes {
+ c.flushLocked()
+ }
+ }
+}
+
+// flushLocked writes the pending chunk under the next seq. A Put that
+// answers ErrCASConflict was written by an earlier attempt whose answer was
+// lost; the seq is spent either way.
+func (c *Chunker) flushLocked() {
+ if len(c.pending) == 0 || c.err != nil {
+ return
+ }
+ seq := c.seq + 1
+ key := ChunkKey(c.acct, c.app, c.id, seq, c.stream)
+ backoff := 100 * time.Millisecond
+ for {
+ _, err := c.s.Put(c.ctx, key, c.pending, "")
+ if err == nil || errors.Is(err, release.ErrCASConflict) {
+ break
+ }
+ select {
+ case <-c.ctx.Done():
+ c.err, c.truncated, c.pending = err, true, nil
+ return
+ case <-time.After(backoff):
+ }
+ backoff = min(backoff*2, 5*time.Second)
+ }
+ c.seq, c.pending = seq, nil
+}
+
+// Close stops the ticker and writes the last chunk. Call it after both
+// streams reached their end (CmdHandle.Stop drains them); writes after Close
+// are dropped. It does not return until every chunk is written or chunking
+// gave up, so a caller may write result.json once it returns. It returns the
+// last seq written (0: no output), whether output was cut off, and the
+// failure that ended chunking, if one did. It is safe to call twice.
+func (c *Chunker) Close() (int64, bool, error) {
+ c.closeOnce.Do(func() {
+ close(c.stop)
+ <-c.done
+ })
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.flushLocked()
+ c.closed = true
+ return c.seq, c.truncated, c.err
+}
diff --git a/internal/runs/chunker_test.go b/internal/runs/chunker_test.go
new file mode 100644
index 000000000..dc3297952
--- /dev/null
+++ b/internal/runs/chunker_test.go
@@ -0,0 +1,185 @@
+package runs
+
+import (
+ "context"
+ "errors"
+ "io"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const tid = "run_abcdefghijklmnopqrstuvwxyz"
+
+func chunksOf(t *testing.T, s release.Store) []string {
+ t.Helper()
+ keys, err := ListChunks(context.Background(), s, "bab", "titogo", tid, 0, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []string
+ for _, k := range keys {
+ raw, _, err := s.Get(context.Background(), k)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, stream, _ := ParseChunkKey(k, OutPrefix("bab", "titogo", tid))
+ out = append(out, stream+":"+string(raw))
+ }
+ return out
+}
+
+func TestChunkerKeepsReadOrderAcrossStreams(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
+ io.WriteString(c.Stream(StreamOut), "a")
+ io.WriteString(c.Stream(StreamErr), "b")
+ io.WriteString(c.Stream(StreamOut), "c")
+ io.WriteString(c.Stream(StreamOut), "d")
+ last, truncated, err := c.Close()
+ if err != nil || truncated || last != 3 {
+ t.Fatalf("Close = %d %v %v", last, truncated, err)
+ }
+ if got := strings.Join(chunksOf(t, s), " "); got != "o:a e:b o:cd" {
+ t.Fatalf("chunks %q", got)
+ }
+}
+
+func TestChunkerFlushesOnTheTickAndAtTheSizeLimit(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ c := newChunker(context.Background(), s, "bab", "titogo", tid, 20*time.Millisecond, MaxOutputBytes)
+ io.WriteString(c.Stream(StreamOut), "tick")
+ deadline := time.Now().Add(2 * time.Second)
+ for len(chunksOf(t, s)) == 0 && time.Now().Before(deadline) {
+ time.Sleep(10 * time.Millisecond)
+ }
+ if got := chunksOf(t, s); len(got) != 1 || got[0] != "o:tick" {
+ t.Fatalf("after a tick: %q", got)
+ }
+ c2 := newChunker(context.Background(), s, "bab", "titogo", "run_bbbbbbbbbbbbbbbbbbbbbbbbbb", time.Hour, MaxOutputBytes)
+ io.WriteString(c2.Stream(StreamOut), strings.Repeat("x", ChunkBytes+1))
+ keys, _ := ListChunks(context.Background(), s, "bab", "titogo", "run_bbbbbbbbbbbbbbbbbbbbbbbbbb", 0, 0)
+ if len(keys) != 1 {
+ t.Fatalf("a full chunk must flush at once: %q", keys)
+ }
+ if last, _, _ := c2.Close(); last != 2 {
+ t.Fatalf("last %d, want 2 (one full chunk, one byte)", last)
+ }
+ c.Close()
+}
+
+func TestChunkerStopsAtTheCapAndSaysSo(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, 10)
+ n, err := io.WriteString(c.Stream(StreamOut), "0123456789abcdef")
+ if n != 16 || err != nil {
+ t.Fatalf("a write past the cap must not fail the stream: %d %v", n, err)
+ }
+ io.WriteString(c.Stream(StreamErr), "more")
+ _, truncated, err := c.Close()
+ if !truncated || err != nil {
+ t.Fatalf("truncated %v err %v", truncated, err)
+ }
+ if got := strings.Join(chunksOf(t, s), ""); got != "o:0123456789" {
+ t.Fatalf("chunks %q", got)
+ }
+}
+
+// failingStore fails the first n Puts.
+type failingStore struct {
+ release.Store
+ mu sync.Mutex
+ fail int
+}
+
+func (f *failingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ f.mu.Lock()
+ if f.fail > 0 {
+ f.fail--
+ f.mu.Unlock()
+ return "", errors.New("slow down")
+ }
+ f.mu.Unlock()
+ return f.Store.Put(ctx, key, body, ifMatch)
+}
+
+func TestChunkerRetriesAFailedPut(t *testing.T) {
+ s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 2}
+ c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
+ io.WriteString(c.Stream(StreamOut), "once")
+ last, _, err := c.Close()
+ if err != nil || last != 1 {
+ t.Fatalf("Close = %d %v", last, err)
+ }
+ if got := chunksOf(t, s.Store); len(got) != 1 || got[0] != "o:once" {
+ t.Fatalf("chunks %q", got)
+ }
+}
+
+func TestChunkerGivesUpWhenItsContextEnds(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 1 << 30}
+ c := newChunker(ctx, s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
+ go func() { time.Sleep(50 * time.Millisecond); cancel() }()
+ io.WriteString(c.Stream(StreamOut), strings.Repeat("x", ChunkBytes))
+ _, truncated, err := c.Close()
+ if err == nil || !truncated {
+ t.Fatalf("a chunk that never lands: truncated %v err %v", truncated, err)
+ }
+}
+
+func TestChunkerRefusesAnInvalidRunAtConstruction(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ for _, tc := range []struct {
+ acct, app, id string
+ want error
+ }{
+ {"bab", "titogo", "run_../../x", ErrBadID},
+ {"bab", "titogo", "", ErrBadID},
+ {"bab", "../other", tid, ErrBadApp},
+ {"../bab", "titogo", tid, ErrBadApp},
+ } {
+ c := NewChunker(context.Background(), s, tc.acct, tc.app, tc.id)
+ n, err := io.WriteString(c.Stream(StreamOut), "x")
+ if n != 1 || err != nil {
+ t.Fatalf("%+v: a refused chunker still swallows writes: %d %v", tc, n, err)
+ }
+ if _, _, err := c.Close(); !errors.Is(err, tc.want) {
+ t.Fatalf("%+v: Close err %v, want %v", tc, err, tc.want)
+ }
+ }
+ if keys, _ := s.List(context.Background(), ""); len(keys) != 0 {
+ t.Fatalf("a refused chunker wrote %q", keys)
+ }
+}
+
+func TestChunkerRefusesAnUnknownStream(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
+ io.WriteString(c.Stream("../x"), "bad")
+ _, _, err := c.Close()
+ if err == nil {
+ t.Fatal("an unknown stream must surface in Close")
+ }
+ if keys, _ := s.List(context.Background(), ""); len(keys) != 0 {
+ t.Fatalf("wrote %q", keys)
+ }
+}
+
+func TestChunkerDropsWritesAfterClose(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
+ io.WriteString(c.Stream(StreamOut), "a")
+ last, _, _ := c.Close()
+ io.WriteString(c.Stream(StreamOut), "late")
+ last2, _, _ := c.Close()
+ if last != 1 || last2 != 1 {
+ t.Fatalf("last %d then %d", last, last2)
+ }
+ if got := chunksOf(t, s); len(got) != 1 || got[0] != "o:a" {
+ t.Fatalf("chunks %q", got)
+ }
+}