amadan

carlos / platform Public

cli: carlos run's second Ctrl-C ends a cancel resend, bounds late cancels, leaves a finished run alone (review fix 1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4e9d5add2efa684f550d9eed5b2a084cd092a65d parent 09b2a36
2 files changed, +156 −35
  • run.go +49 −16
  • run_test.go +107 −19

CI cancelled — superseded by a newer push — run details

diff --git a/run.go b/run.go
index f9981fc77..55ce87994 100644
--- a/run.go
+++ b/run.go
@@ -372,6 +372,9 @@ func ackRun(ctx context.Context, c runClient, t runTerm, id string, lastSeq int6
return
}
}
+ if ctx.Err() != nil {
+ return // cut off by Ctrl-C: nothing hangs on the ack
+ }
t.say(fmt.Sprintf("carlos run: %s", err))
}
@@ -384,14 +387,14 @@ 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
- w := &runWatch{ch: t.interrupts}
- cancel := runCancel{w: w}
+ watch := &runWatch{ch: t.interrupts}
+ cancel := runCancel{w: watch}
var failingSince time.Time
for {
var p runs.Page
var err error
- w.run(ctx, func(ctx context.Context) { p, err = c.page(ctx, id, after) })
- if w.stopped() {
+ watch.run(ctx, func(ctx context.Context) { p, err = c.page(ctx, id, after) })
+ if watch.stopped() {
return stopWaiting(ctx, t, c, id, &cancel)
}
if err != nil {
@@ -441,10 +444,9 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp
continue
}
if p.Result != nil {
- w.run(ctx, func(ctx context.Context) { ackRun(ctx, c, t, id, p.Result.LastSeq) })
- if w.stopped() {
- return stopWaiting(ctx, t, c, id, &cancel)
- }
+ // The result is in hand and the ack is a courtesy: Ctrl-C during
+ // it ends the ack, not the exit code, and cancels nothing.
+ watch.run(ctx, func(ctx context.Context) { ackRun(ctx, c, t, id, p.Result.LastSeq) })
code, lines := runExit(host, *p.Result)
if p.Result.Cancelled && p.Result.Started.IsZero() {
// Cancelled before any box launched it: there was no command.
@@ -466,8 +468,8 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp
if !expireAt.IsZero() && !t.now().Before(expireAt) {
var state string
var err error
- w.run(ctx, func(ctx context.Context) { state, err = c.expire(ctx, id) })
- if w.stopped() {
+ watch.run(ctx, func(ctx context.Context) { state, err = c.expire(ctx, id) })
+ if watch.stopped() {
return stopWaiting(ctx, t, c, id, &cancel)
}
var ce *runConsoleError
@@ -475,7 +477,7 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp
case err == nil && state == provision.RunExpired:
// The run's uploaded files go now, not with the sweep.
// The ack is a courtesy: nothing hangs on it.
- w.run(ctx, func(ctx context.Context) { _ = c.ack(ctx, id, 0) })
+ watch.run(ctx, func(ctx context.Context) { _ = c.ack(ctx, id, 0) })
t.say(runNotPickedUp(host))
return 3
case err == nil || errors.As(err, &ce):
@@ -502,9 +504,26 @@ type runWatch struct {
}
// run calls f with a context that the second Ctrl-C cancels, and reports
-// whether it did. Ctrl-Cs that arrive meanwhile are counted for the loop.
-func (w *runWatch) run(ctx context.Context, f func(context.Context)) (aborted bool) {
+// whether it did. Ctrl-Cs that arrive meanwhile are counted for the loop. If
+// the second has already been taken, the context starts cancelled: no request
+// gets going after the user said stop.
+func (w *runWatch) run(ctx context.Context, f func(context.Context)) bool {
+ return w.do(ctx, 0, f)
+}
+
+// do is run with a deadline when limit is set. A request made after the
+// second Ctrl-C is bounded this way instead of started cancelled: it is still
+// attempted, and a further Ctrl-C ends it.
+func (w *runWatch) do(ctx context.Context, limit time.Duration, f func(context.Context)) (aborted bool) {
rctx, cancel := context.WithCancel(ctx)
+ if limit > 0 {
+ var stopTimer context.CancelFunc
+ rctx, stopTimer = context.WithTimeout(rctx, limit)
+ defer stopTimer()
+ } else if w.n >= 2 {
+ aborted = true
+ cancel()
+ }
stop, fin := make(chan struct{}), make(chan struct{})
go func() {
defer close(fin)
@@ -547,6 +566,10 @@ func (w *runWatch) take() bool {
}
}
+// runLateCancelWait is how long a cancel sent after the second Ctrl-C may
+// take. A variable so a test can shorten it.
+var runLateCancelWait = 3 * time.Second
+
// runCancel is the state of the cancel a Ctrl-C asks for.
type runCancel struct {
w *runWatch
@@ -564,8 +587,16 @@ func (k *runCancel) send(ctx context.Context, t runTerm, c runClient, id string)
return
}
var err error
- aborted := k.w.run(ctx, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
- if aborted && errors.Is(err, context.Canceled) {
+ var aborted bool
+ // A cancel sent after the second Ctrl-C is still attempted, but only for
+ // a moment: the user has said stop waiting.
+ late := k.w.stopped()
+ if late {
+ aborted = k.w.do(ctx, runLateCancelWait, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
+ } else {
+ aborted = k.w.run(ctx, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
+ }
+ if (aborted && errors.Is(err, context.Canceled)) || (late && errors.Is(err, context.DeadlineExceeded)) {
k.abandoned = true
return
}
@@ -617,7 +648,9 @@ func interrupted(ctx context.Context, t runTerm, c runClient, id string, k *runC
select {
case <-t.after(d):
k.send(ctx, t, c, id)
- return false
+ // A second Ctrl-C during the resend ends the wait here, not
+ // after another request.
+ return k.w.stopped()
case <-k.w.ch:
k.w.n++
k.w.handled++
diff --git a/run_test.go b/run_test.go
index 07da6011c..e9976b795 100644
--- a/run_test.go
+++ b/run_test.go
@@ -817,18 +817,22 @@ func TestFollowRunInterrupts(t *testing.T) {
// its context is cancelled, as a real request does.
type gatedRun struct {
*scriptedRun
- holdPage int // the page call (1-based) to hold, 0 for none
- holdCancel bool // hold the first cancel call
- holdAck bool // hold the first ack call
- entered chan struct{}
+ holdPage int // the page call (1-based) to hold, 0 for none
+ holdCancel int // the cancel call (1-based) to hold, 0 for none
+ holdAck bool // hold the first ack call
+ entered chan struct{} // the held page or ack call has arrived
+ cancelIn chan struct{} // the held cancel call has arrived
release chan struct{}
- ctxErr error // the held call's context error when it returned
- held int // calls made to a held route, counted when one is held
+ ctxErr error // the held page's context error when it returned
+ cancelCtx error // the held cancel's context error as it arrived
+ acksHeld int
}
-func (g *gatedRun) hold(ctx context.Context) error {
- g.held++
- close(g.entered)
+func (g *gatedRun) hold(ctx context.Context, entered chan struct{}) error {
+ if entered == g.cancelIn {
+ g.cancelCtx = ctx.Err()
+ }
+ close(entered)
select {
case <-g.release:
case <-ctx.Done():
@@ -839,7 +843,7 @@ func (g *gatedRun) hold(ctx context.Context) error {
func (g *gatedRun) page(ctx context.Context, id string, after int64) (runs.Page, error) {
if g.pagesRead()+1 == g.holdPage {
- if err := g.hold(ctx); err != nil {
+ if err := g.hold(ctx, g.entered); err != nil {
g.scriptedRun.mu.Lock()
g.scriptedRun.pageCalls++
g.scriptedRun.mu.Unlock()
@@ -850,25 +854,26 @@ func (g *gatedRun) page(ctx context.Context, id string, after int64) (runs.Page,
}
func (g *gatedRun) cancel(ctx context.Context, id string) (string, error) {
- if g.holdCancel && g.held == 0 {
+ if g.holdCancel != 0 && g.cancelsMade()+1 == g.holdCancel {
g.scriptedRun.mu.Lock()
g.scriptedRun.cancels++
g.scriptedRun.mu.Unlock()
- return "", g.hold(ctx)
+ return "", g.hold(ctx, g.cancelIn)
}
return g.scriptedRun.cancel(ctx, id)
}
func (g *gatedRun) ack(ctx context.Context, id string, last int64) error {
- if g.holdAck && g.held == 0 {
+ if g.holdAck && g.acksHeld == 0 {
+ g.acksHeld++
g.scriptedRun.acked = append(g.scriptedRun.acked, last)
- return g.hold(ctx)
+ return g.hold(ctx, g.entered)
}
return g.scriptedRun.ack(ctx, id, last)
}
func newGatedRun(s *scriptedRun) *gatedRun {
- return &gatedRun{scriptedRun: s, entered: make(chan struct{}), release: make(chan struct{})}
+ return &gatedRun{scriptedRun: s, entered: make(chan struct{}), cancelIn: make(chan struct{}), release: make(chan struct{})}
}
func awaitFollowRun(t *testing.T, done <-chan int, what string) int {
@@ -962,6 +967,12 @@ func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
codes := start(g, f)
awaitSignal(t, g.entered, "the first page request was never made")
f.sig <- os.Interrupt
+ // Wait until the watcher has taken the Ctrl-C, then give the loop a
+ // moment to act on it: it must not, while the request is open.
+ for end := time.Now().Add(2 * time.Second); len(f.sig) > 0 && time.Now().Before(end); {
+ time.Sleep(time.Millisecond)
+ }
+ time.Sleep(50 * time.Millisecond)
// The request is not abandoned, and nothing is cancelled until it is back.
if got := g.cancelsMade(); got != 0 {
t.Fatalf("%d cancels while the page request was in flight", got)
@@ -1035,9 +1046,9 @@ func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
f := newTermFx()
f.sig <- os.Interrupt
g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
- g.holdCancel = true
+ g.holdCancel = 1
codes := start(g, f)
- awaitSignal(t, g.entered, "the cancel request was never made")
+ awaitSignal(t, g.cancelIn, "the cancel request was never made")
f.sig <- os.Interrupt
if code := quick(t, codes); code != 130 {
t.Fatalf("exit %d, want 130", code)
@@ -1049,7 +1060,7 @@ func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
t.Fatalf("stderr %q: a cut-off cancel is not an error to report", got)
}
})
- t.Run("the second ends an ack in flight", func(t *testing.T) {
+ t.Run("both during the ack end the ack, not the exit code, and cancel nothing", func(t *testing.T) {
f := newTermFx()
g := newGatedRun(&scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{LastSeq: 3}}}})
g.holdAck = true
@@ -1057,11 +1068,88 @@ func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
awaitSignal(t, g.entered, "the ack request was never made")
f.sig <- os.Interrupt
f.sig <- os.Interrupt
+ if code := quick(t, codes); code != 0 {
+ t.Fatalf("exit %d, want the command's own 0", code)
+ }
+ if got := g.cancelsMade(); got != 0 {
+ t.Fatalf("%d cancels sent to a finished run", got)
+ }
+ if got := f.err.String(); got != "" {
+ t.Fatalf("stderr %q", got)
+ }
+ })
+ t.Run("a second Ctrl-C during the cancel resend ends the wait before another request", func(t *testing.T) {
+ f := newTermFx()
+ f.sig <- os.Interrupt
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}, cancelErrs: []error{errors.New("connection reset")}})
+ g.holdCancel = 2 // the first cancel fails fast; its resend never answers
+ g.holdPage = 3 // any page request after that would hang too
+ codes := start(g, f)
+ awaitSignal(t, g.cancelIn, "the cancel was never sent again")
+ f.sig <- os.Interrupt
if code := quick(t, codes); code != 130 {
t.Fatalf("exit %d, want 130", code)
}
+ if got := g.pagesRead(); got != 2 {
+ t.Fatalf("%d page requests, want 2: none after the second Ctrl-C", got)
+ }
+ })
+ late := func(t *testing.T) {
+ old := runLateCancelWait
+ runLateCancelWait = 100 * time.Millisecond
+ t.Cleanup(func() { runLateCancelWait = old })
+ }
+ t.Run("a cancel first sent after the second Ctrl-C is attempted, but only briefly", func(t *testing.T) {
+ late(t)
+ f := newTermFx()
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
+ g.holdPage = 1
+ g.holdCancel = 1 // never answers
+ codes := start(g, f)
+ awaitSignal(t, g.entered, "the page request was never made")
+ f.sig <- os.Interrupt
+ f.sig <- os.Interrupt
+ select {
+ case code := <-codes:
+ if code != 130 {
+ t.Fatalf("exit %d, want 130", code)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("a cancel sent after the second Ctrl-C held the exit")
+ }
if got := g.cancelsMade(); got != 1 {
- t.Fatalf("%d cancels, want 1", got)
+ t.Fatalf("%d cancels, want 1: still attempted", got)
+ }
+ if g.cancelCtx != nil {
+ t.Fatalf("the cancel was sent on a context already %v: it was not attempted", g.cancelCtx)
+ }
+ if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
+ t.Fatalf("stderr %q", got)
+ }
+ })
+ t.Run("a cancel resent on the second Ctrl-C is attempted, but only briefly", func(t *testing.T) {
+ late(t)
+ f := newTermFx()
+ f.sig <- os.Interrupt
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}, cancelErrs: []error{errors.New("connection reset")}})
+ g.holdPage = 2 // the second Ctrl-C arrives during this page request
+ g.holdCancel = 2 // the resend never answers
+ codes := start(g, f)
+ awaitSignal(t, g.entered, "the second page request was never made")
+ f.sig <- os.Interrupt
+ select {
+ case code := <-codes:
+ if code != 130 {
+ t.Fatalf("exit %d, want 130", code)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("a cancel resent after the second Ctrl-C held the exit")
+ }
+ if got := g.cancelsMade(); got != 2 {
+ t.Fatalf("%d cancels, want 2: the resend is still attempted", got)
+ }
+ if g.cancelCtx != nil {
+ t.Fatalf("the resend was sent on a context already %v: it was not attempted", g.cancelCtx)
}
})
}