carlos / platform Public

cli: carlos run retries a failed page, paces a catch-up, resends an unanswered cancel (review fix 1)

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

claude 2 weeks ago 6b8dad2cefd9066cef6552a97abb61cfa3ab901f parent 1e35205

3 files changed, +271 −47

apimode.go +12 −5
run.go +107 −30
run_test.go +152 −12

CI failed — run details

diff --git a/apimode.go b/apimode.go

index c4ce9029d..86f988b07 100644

--- a/apimode.go

+++ b/apimode.go

@@ -806,26 +806,33 @@ func apiErrorFromBody(status int, body []byte) error {

// doesn't: a caller here needs the FULL body (to try apiErrorFromBody's

// json-first decode), not just a pre-picked message.

func (s *apiSession) getJSON(ctx context.Context, path string) (int, []byte, error) {

+ status, _, body, err := s.getJSONHeader(ctx, path)

+ return status, body, err

+// getJSONHeader is getJSON that also hands back the response headers, for

+// the caller that reads one (Retry-After).

+func (s *apiSession) getJSONHeader(ctx context.Context, path string) (int, http.Header, []byte, error) {

req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.c.base+path, nil)

if err != nil {

- return 0, nil, err

+ return 0, nil, nil, err

}

if s.c.token != "" {

req.Header.Set("Authorization", "Bearer "+s.c.token)

}

res, err := s.c.httpc.Do(req)

if err != nil {

- return 0, nil, err

+ return 0, nil, nil, err

}

defer res.Body.Close()

if merr := consoleMoved(s.c.base, res.StatusCode, res.Header.Get("Location")); merr != nil {

- return res.StatusCode, nil, offerConsoleMoved(merr)

+ return res.StatusCode, res.Header, nil, offerConsoleMoved(merr)

}

body, err := io.ReadAll(io.LimitReader(res.Body, 1<<20))

if err != nil {

- return res.StatusCode, nil, err

+ return res.StatusCode, res.Header, nil, err

}

- return res.StatusCode, body, nil

+ return res.StatusCode, res.Header, body, nil

}

// sendJSON runs one JSON request over the given method and hands back the

diff --git a/run.go b/run.go

index 5d19855b7..b467189f0 100644

--- a/run.go

+++ b/run.go

@@ -139,8 +139,9 @@ type runClient interface {

// runConsoleError is a 4xx the console answered, with its message: a

// refusal by the console, exit 2.

type runConsoleError struct {

- status int

- msg string

+ status int

+ msg string

+ retryAfter time.Duration // a 429's Retry-After, when the console sent one

}

func (e *runConsoleError) Error() string { return e.msg }

@@ -271,9 +272,16 @@ func copyRunFilePart(mw *multipart.Writer, f runFileHandle, path string, want ru

}

func (c runAPI) page(ctx context.Context, id string, after int64) (runs.Page, error) {

- status, body, err := c.s.getJSON(ctx, c.path("/"+url.PathEscape(id)+fmt.Sprintf("?after=%d", after)))

+ status, h, body, err := c.s.getJSONHeader(ctx, c.path("/"+url.PathEscape(id)+fmt.Sprintf("?after=%d", after)))

var out runs.Page

- return out, runAnswer(status, body, err, http.StatusOK, &out)

+ err = runAnswer(status, body, err, http.StatusOK, &out)

+ var ce *runConsoleError

+ if errors.As(err, &ce) && status == http.StatusTooManyRequests {

+ if n, perr := strconv.Atoi(h.Get("Retry-After")); perr == nil && n > 0 {

+ ce.retryAfter = time.Duration(n) * time.Second

+ }

+ return out, err

}

func (c runAPI) cancel(ctx context.Context, id string) (string, error) {

@@ -311,6 +319,25 @@ const (

runAckWait = time.Second

)

+// How the follow loop treats a console that is slow to answer. A page that

+// fails for a reason that may pass (a 5xx, a lost connection, a 429) is asked

+// for again; after runPageGiveUp of failing in a row the follow is given up

+// (the run carries on, and --follow picks it up again). A 429 waits longer,

+// or as long as the console's Retry-After says, at most runRateWaitMax. A

+// page cut short is followed after a short pause, so that catching up on a

+// long run cannot spend the console's per-token budget in a burst.

+const (

+ runPageGiveUp = 2 * time.Minute

+ runRateWait = 5 * time.Second

+ runRateWaitMax = time.Minute

+ runMorePause = 200 * time.Millisecond

+// runNotPickedUp is RUN-C55.

+func runNotPickedUp(host string) string {

+ return fmt.Sprintf("%s: not picked up within 60 seconds. the run did not start. try again", host)

// runConflict reports whether err is the console's 409 with exactly msg.

func runConflict(err error, msg string) bool {

var ce *runConsoleError

@@ -337,11 +364,11 @@ func ackRun(ctx context.Context, c runClient, t runTerm, id string, lastSeq int6

case errors.As(err, &ce) && ce.status == http.StatusNotFound:

return

case errors.As(err, &ce) && !runConflict(err, fmt.Sprintf("run %s has not finished yet", id)):

- t.say(fmt.Sprintf("carlos run: %v", err))

+ t.say(fmt.Sprintf("carlos run: %s", err))

return

}

- t.say(fmt.Sprintf("carlos run: %v", err))

+ t.say(fmt.Sprintf("carlos run: %s", err))

}

// followRun prints a run's output until its result, acknowledges it, and

@@ -353,7 +380,8 @@ func ackRun(ctx context.Context, c runClient, t runTerm, id string, lastSeq int6

// following; the second stops waiting (§5.7).

func followRun(ctx context.Context, c runClient, t runTerm, host, id string, expireAt time.Time) int {

var after int64

- cancelled := false

+ var cancel runCancel

+ var failingSince time.Time

for {

p, err := c.page(ctx, id, after)

if err != nil {

@@ -362,13 +390,29 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp

case errors.As(err, &ce) && ce.status == http.StatusNotFound:

t.say(fmt.Sprintf("%s has no run %s, or this console has no carlos run yet", host, id))

return 2

- case errors.As(err, &ce):

- t.say(fmt.Sprintf("carlos run: %v", err))

+ case errors.As(err, &ce) && ce.status != http.StatusTooManyRequests:

+ t.say(fmt.Sprintf("carlos run: %s", err))

return 2

}

- t.say(fmt.Sprintf("carlos run: %v", err))

- return 1

+ // A 5xx, a 429 or a connection that failed may pass: ask again,

+ // for a while.

+ if failingSince.IsZero() {

+ failingSince = t.now()

+ }

+ if t.now().Sub(failingSince) >= runPageGiveUp {

+ t.say(fmt.Sprintf("carlos run: %s", err))

+ return 1

+ }

+ wait := time.Second

+ if ce != nil {

+ wait = min(max(runRateWait, ce.retryAfter), runRateWaitMax)

+ }

+ if interrupted(ctx, t, c, id, &cancel, wait) {

+ return 130

+ }

+ continue

}

+ failingSince = time.Time{}

for _, ch := range p.Chunks {

if ch.Seq <= after {

continue

@@ -381,6 +425,9 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp

after = ch.Seq

}

if p.More {

+ if interrupted(ctx, t, c, id, &cancel, runMorePause) {

+ return 130

+ }

continue

}

if p.Result != nil {

@@ -397,7 +444,7 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp

}

switch p.State {

case provision.RunExpired:

- t.say(fmt.Sprintf("%s: not picked up within 60 seconds. the run did not start. try again", host))

+ t.say(runNotPickedUp(host))

return 3

case provision.RunCancelled:

t.say(fmt.Sprintf("%s: run %s was cancelled before it started", host, id))

@@ -408,7 +455,10 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp

var ce *runConsoleError

switch {

case err == nil && state == provision.RunExpired:

- t.say(fmt.Sprintf("%s: not picked up within 60 seconds. the run did not start. try again", host))

+ // The run's uploaded files go now, not with the sweep.

+ // The ack is a courtesy: nothing hangs on it.

+ _ = c.ack(ctx, id, 0)

+ t.say(runNotPickedUp(host))

return 3

case err == nil || errors.As(err, &ce):

// A box picked the run first (409), or the console will

@@ -418,36 +468,63 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp

}

- if interrupted(ctx, t, c, id, &cancelled) {

+ if interrupted(ctx, t, c, id, &cancel, time.Second) {

return 130

}

-// interrupted waits out the poll interval and handles a Ctrl-C that arrives

-// first, or was already waiting: a pending Ctrl-C is taken before the next

-// tick, never after it. The first cancels the run and returns false so

-// following goes on; the second returns true: stop waiting.

-func interrupted(ctx context.Context, t runTerm, c runClient, id string, cancelled *bool) bool {

+// runCancel is the state of the cancel a Ctrl-C asks for.

+type runCancel struct {

+ asked, done, said bool

+// send makes the cancel, if one was asked for and has not been answered. A

+// cancel retried after a lost 200 finds the run already ended, which is what

+// was asked for. One that did not arrive is sent again on the next call, so

+// the cancel still happens (run spec §5.7); it is said once.

+func (k *runCancel) send(ctx context.Context, t runTerm, c runClient, id string) {

+ if !k.asked || k.done {

+ return

+ }

+ _, err := c.cancel(ctx, id)

+ var ce *runConsoleError

+ switch {

+ case err == nil, runConflict(err, fmt.Sprintf("run %s has already ended", id)):

+ k.done = true

+ case errors.As(err, &ce):

+ k.done = true

+ t.say(fmt.Sprintf("carlos run: %s", err))

+ case !k.said:

+ k.said = true

+ t.say(fmt.Sprintf("carlos run: %s", err))

+ }

+// interrupted waits out d, the pause before the next request, and handles a

+// Ctrl-C that arrives first or was already waiting: a pending Ctrl-C is

+// taken before the next tick, never after it. The first asks for the cancel

+// and returns false so following goes on; the second tries the cancel once

+// more if it is unanswered and returns true: stop waiting. A cancel still

+// unanswered when the wait ends is sent again.

+func interrupted(ctx context.Context, t runTerm, c runClient, id string, k *runCancel, d time.Duration) bool {

select {

case <-t.interrupts:

default:

select {

- case <-t.after(time.Second):

+ case <-t.after(d):

+ k.send(ctx, t, c, id)

return false

case <-t.interrupts:

}

- if *cancelled {

+ if k.asked {

+ k.send(ctx, t, c, id)

return true

}

- *cancelled = true

+ k.asked = true

t.say(fmt.Sprintf("cancelling run %s. press Ctrl-C again to stop waiting", id))

- // A cancel retried after a lost 200 finds the run already ended, which is

- // what was asked for.

- if _, err := c.cancel(ctx, id); err != nil && !runConflict(err, fmt.Sprintf("run %s has already ended", id)) {

- t.say(fmt.Sprintf("carlos run: %v", err))

- }

+ k.send(ctx, t, c, id)

return false

}

@@ -460,7 +537,7 @@ func runExit(host string, r runs.Result) (int, []string) {

}

switch {

case r.Refused == runs.RefusedExpired:

- return 3, append(lines, fmt.Sprintf("%s: not picked up within 60 seconds. the run did not start. try again", host))

+ return 3, append(lines, runNotPickedUp(host))

case r.Refused == runs.RefusedFlat:

return 3, append(lines, fmt.Sprintf("%s: the box is not on the nested layout yet, so carlos run cannot run there", host))

case r.Refused == runs.RefusedClaim:

@@ -502,10 +579,10 @@ func startRun(ctx context.Context, c runClient, t runTerm, o runOpts, app string

t.say(err.Error())

return 2

case errors.As(err, &ce):

- t.say(fmt.Sprintf("carlos run: %v", err))

+ t.say(fmt.Sprintf("carlos run: %s", err))

return 2

}

- t.say(fmt.Sprintf("carlos run: %v", err))

+ t.say(fmt.Sprintf("carlos run: %s", err))

return 1

}

t.say(fmt.Sprintf("run %s on %s. to follow it again: carlos run --app %s --instance %s --follow %s", created.ID, o.instance, app, o.instance, created.ID))

diff --git a/run_test.go b/run_test.go

index 6fd238cbb..b85fa539f 100644

--- a/run_test.go

+++ b/run_test.go

@@ -332,17 +332,19 @@ func TestRunAPICreateRefusesFilesThatDisagreeWithMeta(t *testing.T) {

// scriptedRun is a runClient that answers pages from a script.

type scriptedRun struct {

- pages []runs.Page

- pageErr error

- expireTo string

- expireErr error

- cancelErr error

- createErr error

- ackErrs []error // one per ack call, in order; nil once spent

- acked []int64

- cancels int

- expires int

- created runs.Meta

+ pages []runs.Page

+ pageErr error // every page call fails with it

+ pageErrs []error // the first page calls fail with these, in order

+ cancelErrs []error // one per cancel call, in order; cancelErr once spent

+ expireTo string

+ expireErr error

+ cancelErr error

+ createErr error

+ ackErrs []error // one per ack call, in order; nil once spent

+ acked []int64

+ cancels int

+ expires int

+ created runs.Meta

}

func (s *scriptedRun) create(_ context.Context, m runs.Meta, _ []string) (runs.Created, error) {

@@ -353,6 +355,13 @@ func (s *scriptedRun) create(_ context.Context, m runs.Meta, _ []string) (runs.C

return runs.Created{ID: "run_a", State: "queued"}, nil

}

func (s *scriptedRun) page(context.Context, string, int64) (runs.Page, error) {

+ if len(s.pageErrs) > 0 {

+ err := s.pageErrs[0]

+ s.pageErrs = s.pageErrs[1:]

+ if err != nil { // nil: this call is answered from pages

+ return runs.Page{}, err

+ }

if s.pageErr != nil {

return runs.Page{}, s.pageErr

}

@@ -364,6 +373,11 @@ func (s *scriptedRun) page(context.Context, string, int64) (runs.Page, error) {

}

func (s *scriptedRun) cancel(context.Context, string) (string, error) {

s.cancels++

+ if len(s.cancelErrs) > 0 {

+ err := s.cancelErrs[0]

+ s.cancelErrs = s.cancelErrs[1:]

+ return "cancel", err

+ }

return "cancel", s.cancelErr

}

func (s *scriptedRun) expire(context.Context, string) (string, error) {

@@ -486,6 +500,9 @@ func TestFollowRunExpiresAQueuedRunAfterSixtySeconds(t *testing.T) {

if code != 3 || s.expires != 1 || !strings.Contains(f.err.String(), "shop.example.com: not picked up within 60 seconds. the run did not start. try again") {

t.Fatalf("exit %d expires %d stderr %q", code, s.expires, f.err.String())

}

+ if len(s.acked) != 1 || s.acked[0] != 0 {

+ t.Fatalf("acked %v (the uploaded files go now, not with the sweep)", s.acked)

+ }

if f.now.Sub(time.Unix(1_700_000_000, 0)) < 60*time.Second {

t.Fatal("expired before 60 seconds")

}

@@ -564,6 +581,109 @@ func TestFollowRunPageErrors(t *testing.T) {

}

+func TestFollowRunRetriesAPageThatMayPass(t *testing.T) {

+ done := runs.Page{State: "done", Chunks: []runs.Chunk{chunk(1, "o", "out\n")}, Result: &runs.Result{Exit: 9, LastSeq: 1}}

+ t.Run("a 503 then success keeps following", func(t *testing.T) {

+ f := newTermFx()

+ start := f.now

+ s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{errors.New("could not read the run. try again"), errors.New("connection reset")}}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 {

+ t.Fatalf("exit %d stderr %q", code, f.err.String())

+ }

+ if f.out.String() != "out\n" || f.err.String() != "" || len(s.acked) != 1 || f.now.Sub(start) != 2*time.Second {

+ t.Fatalf("stdout %q stderr %q acked %v waited %s", f.out.String(), f.err.String(), s.acked, f.now.Sub(start))

+ }

+ })

+ t.Run("a 429 waits as long as Retry-After says", func(t *testing.T) {

+ f := newTermFx()

+ start := f.now

+ s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{&runConsoleError{status: http.StatusTooManyRequests, msg: "too many run requests. wait a minute and try again", retryAfter: 17 * time.Second}}}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 || f.now.Sub(start) != 17*time.Second {

+ t.Fatalf("exit %d waited %s", code, f.now.Sub(start))

+ }

+ })

+ t.Run("a 429 without Retry-After waits longer than a second, and no longer than a minute", func(t *testing.T) {

+ f := newTermFx()

+ start := f.now

+ s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{

+ &runConsoleError{status: http.StatusTooManyRequests, msg: "slow down"},

+ &runConsoleError{status: http.StatusTooManyRequests, msg: "slow down", retryAfter: time.Hour},

+ }}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 || f.now.Sub(start) != runRateWait+runRateWaitMax {

+ t.Fatalf("exit %d waited %s", code, f.now.Sub(start))

+ }

+ })

+ t.Run("a page that keeps failing is given up on after two minutes with exit 1", func(t *testing.T) {

+ f := newTermFx()

+ start := f.now

+ s := &scriptedRun{pageErr: errors.New("could not read the run. try again")}

+ code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})

+ if waited := f.now.Sub(start); code != 1 || waited < 2*time.Minute || waited > 2*time.Minute+5*time.Second || f.err.String() != "carlos run: could not read the run. try again\n" {

+ t.Fatalf("exit %d waited %s stderr %q", code, waited, f.err.String())

+ }

+ })

+ t.Run("the bound counts failures in a row", func(t *testing.T) {

+ f := newTermFx()

+ var errs []error

+ for range 100 {

+ errs = append(errs, errors.New("blip"))

+ }

+ // 100s of failures, a page that works, 100s more, then the result:

+ // 200s in all, but never two minutes in a row.

+ all := append(append(append([]error{}, errs...), nil), errs...)

+ s := &scriptedRun{pageErrs: append(all, nil), pages: []runs.Page{{State: "picked"}, done}}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 {

+ t.Fatalf("exit %d stderr %q", code, f.err.String())

+ }

+ })

+ t.Run("a refusal that cannot pass is not retried", func(t *testing.T) {

+ f := newTermFx()

+ s := &scriptedRun{pageErrs: []error{&runConsoleError{status: http.StatusForbidden, msg: "no"}}, pages: []runs.Page{done}}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 2 || f.err.String() != "carlos run: no\n" {

+ t.Fatalf("exit %d stderr %q", code, f.err.String())

+ }

+ })

+ t.Run("a failing page does not stop Ctrl-C", func(t *testing.T) {

+ f := newTermFx()

+ f.sig <- os.Interrupt

+ s := &scriptedRun{pageErr: errors.New("connection refused")}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels < 1 {

+ t.Fatalf("exit %d cancels %d", code, s.cancels)

+ }

+ })

+func TestFollowRunPacesAPageCutShort(t *testing.T) {

+ f := newTermFx()

+ start := f.now

+ res := &runs.Result{LastSeq: 3}

+ s := &scriptedRun{pages: []runs.Page{

+ {Chunks: []runs.Chunk{chunk(1, "o", "a")}, Result: res, More: true},

+ {Chunks: []runs.Chunk{chunk(2, "o", "b")}, Result: res, More: true},

+ {Chunks: []runs.Chunk{chunk(3, "o", "c")}, Result: res},

+ }}

+ followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})

+ if f.out.String() != "abc" || f.now.Sub(start) != 2*runMorePause {

+ t.Fatalf("stdout %q paused %s", f.out.String(), f.now.Sub(start))

+ }

+func TestRunAPIPageReadsRetryAfter(t *testing.T) {

+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

+ w.Header().Set("Retry-After", "7")

+ w.WriteHeader(http.StatusTooManyRequests)

+ json.NewEncoder(w).Encode(map[string]string{"error": "too many run requests. wait a minute and try again"})

+ }))

+ defer ts.Close()

+ c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "shop.example.com"}

+ _, err := c.page(context.Background(), "run_a", 0)

+ var ce *runConsoleError

+ if !errors.As(err, &ce) || ce.status != http.StatusTooManyRequests || ce.retryAfter != 7*time.Second {

+ t.Fatalf("%#v", err)

+ }

func TestFollowRunAck(t *testing.T) {

notDone := &runConsoleError{status: http.StatusConflict, msg: "run run_a has not finished yet"}

gone := &runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"}

@@ -626,7 +746,7 @@ func TestFollowRunInterrupts(t *testing.T) {

t.Fatalf("exit %d cancels %d acked %v", code, s.cancels, s.acked)

}

})

- t.Run("a cancel retried after a lost 200 is success", func(t *testing.T) {

+ t.Run("a cancel that finds the run already ended is silent success", func(t *testing.T) {

f := newTermFx()

f.sig <- os.Interrupt

s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already ended"}}

@@ -637,6 +757,26 @@ func TestFollowRunInterrupts(t *testing.T) {

t.Fatalf("stderr %q", got)

}

})

+ t.Run("a cancel that did not arrive is sent again on the next poll, and said once", func(t *testing.T) {

+ f := newTermFx()

+ f.sig <- os.Interrupt

+ s := &scriptedRun{pages: []runs.Page{queued, queued, queued, done}, cancelErrs: []error{errors.New("connection reset"), errors.New("connection reset"), nil}}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 3 {

+ t.Fatalf("exit %d cancels %d", code, s.cancels)

+ }

+ if got := strings.Count(f.err.String(), "carlos run: connection reset"); got != 1 {

+ t.Fatalf("said %d times: %q", got, f.err.String())

+ }

+ })

+ t.Run("the second Ctrl-C tries an unanswered cancel once more", func(t *testing.T) {

+ f := newTermFx()

+ f.sig <- os.Interrupt

+ s := &scriptedRun{pages: []runs.Page{queued}, cancelErrs: []error{errors.New("connection reset"), nil}}

+ if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 2 {

+ t.Fatalf("exit %d cancels %d", code, s.cancels)

+ }

+ })

t.Run("a cancel that failed is said", func(t *testing.T) {

f := newTermFx()

f.sig <- os.Interrupt