amadan

carlos / platform Public

cli: carlos run's second Ctrl-C ends a request in flight at once (run spec §5.7)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 09b2a3653a54acad541ebb3690dfefb7dca301c3 parent 03d617d
2 files changed, +262 −48
  • run.go +128 −22
  • run_test.go +134 −26
diff --git a/run.go b/run.go
index b467189f0..f9981fc77 100644
--- a/run.go
+++ b/run.go
@@ -352,9 +352,13 @@ func runConflict(err error, msg string) bool {
// arrive is sent again a few times.
func ackRun(ctx context.Context, c runClient, t runTerm, id string, lastSeq int64) {
var err error
- for try := 0; try < runAckTries; try++ {
+ for try := 0; try < runAckTries && ctx.Err() == nil; try++ {
if try > 0 {
- <-t.after(runAckWait)
+ select {
+ case <-t.after(runAckWait):
+ case <-ctx.Done():
+ return
+ }
}
err = c.ack(ctx, id, lastSeq)
var ce *runConsoleError
@@ -380,10 +384,16 @@ 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
- var cancel runCancel
+ w := &runWatch{ch: t.interrupts}
+ cancel := runCancel{w: w}
var failingSince time.Time
for {
- p, err := c.page(ctx, id, after)
+ var p runs.Page
+ var err error
+ w.run(ctx, func(ctx context.Context) { p, err = c.page(ctx, id, after) })
+ if w.stopped() {
+ return stopWaiting(ctx, t, c, id, &cancel)
+ }
if err != nil {
var ce *runConsoleError
switch {
@@ -431,7 +441,10 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp
continue
}
if p.Result != nil {
- ackRun(ctx, c, t, id, p.Result.LastSeq)
+ 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)
+ }
code, lines := runExit(host, *p.Result)
if p.Result.Cancelled && p.Result.Started.IsZero() {
// Cancelled before any box launched it: there was no command.
@@ -451,13 +464,18 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp
return 130
case provision.RunQueued:
if !expireAt.IsZero() && !t.now().Before(expireAt) {
- state, err := c.expire(ctx, id)
+ var state string
+ var err error
+ w.run(ctx, func(ctx context.Context) { state, err = c.expire(ctx, id) })
+ if w.stopped() {
+ return stopWaiting(ctx, t, c, id, &cancel)
+ }
var ce *runConsoleError
switch {
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.
- _ = c.ack(ctx, id, 0)
+ w.run(ctx, func(ctx context.Context) { _ = c.ack(ctx, id, 0) })
t.say(runNotPickedUp(host))
return 3
case err == nil || errors.As(err, &ce):
@@ -474,20 +492,83 @@ func followRun(ctx context.Context, c runClient, t runTerm, host, id string, exp
}
}
+// runWatch is the follow loop's view of Ctrl-C. Between requests the loop
+// takes a Ctrl-C itself; while a request is in flight run takes it, so the
+// second one can end that request at once (run spec §5.7). n counts every
+// Ctrl-C taken, and handled counts those the loop has acted on.
+type runWatch struct {
+ ch <-chan os.Signal
+ n, handled int
+}
+
+// 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) {
+ rctx, cancel := context.WithCancel(ctx)
+ stop, fin := make(chan struct{}), make(chan struct{})
+ go func() {
+ defer close(fin)
+ for {
+ select {
+ case <-w.ch:
+ w.n++
+ if w.n >= 2 && !aborted {
+ aborted = true
+ cancel()
+ }
+ case <-stop:
+ return
+ }
+ }
+ }()
+ f(rctx)
+ close(stop)
+ <-fin
+ cancel()
+ return aborted
+}
+
+// stopped reports whether the second Ctrl-C has been taken.
+func (w *runWatch) stopped() bool { return w.n >= 2 }
+
+// take claims a Ctrl-C that is waiting, taken by run or still in the channel.
+func (w *runWatch) take() bool {
+ if w.n > w.handled {
+ w.handled++
+ return true
+ }
+ select {
+ case <-w.ch:
+ w.n++
+ w.handled++
+ return true
+ default:
+ return false
+ }
+}
+
// runCancel is the state of the cancel a Ctrl-C asks for.
type runCancel struct {
+ w *runWatch
asked, done, said bool
+ abandoned bool // the second Ctrl-C cut the request off: do not send it again
}
// 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.
+// the cancel still happens (run spec §5.7); it is said once. A cancel the
+// second Ctrl-C cut off is not sent again: that Ctrl-C said stop waiting.
func (k *runCancel) send(ctx context.Context, t runTerm, c runClient, id string) {
- if !k.asked || k.done {
+ if !k.asked || k.done || k.abandoned {
+ 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) {
+ k.abandoned = true
return
}
- _, err := c.cancel(ctx, id)
var ce *runConsoleError
switch {
case err == nil, runConflict(err, fmt.Sprintf("run %s has already ended", id)):
@@ -501,30 +582,55 @@ func (k *runCancel) send(ctx context.Context, t runTerm, c runClient, id string)
}
}
+// press acts on one Ctrl-C that has been taken. 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.
+func (k *runCancel) press(ctx context.Context, t runTerm, c runClient, id string) bool {
+ if k.asked {
+ k.send(ctx, t, c, id)
+ return true
+ }
+ k.asked = true
+ t.say(fmt.Sprintf("cancelling run %s. press Ctrl-C again to stop waiting", id))
+ k.send(ctx, t, c, id)
+ return false
+}
+
+// stopWaiting is the exit once the second Ctrl-C has been taken, possibly
+// while a request was in flight and the first not yet acted on.
+func stopWaiting(ctx context.Context, t runTerm, c runClient, id string, k *runCancel) int {
+ for k.w.take() {
+ if k.press(ctx, t, c, id) {
+ break
+ }
+ }
+ return 130
+}
+
// 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.
+// and returns false so following goes on; the second stops waiting and
+// returns true. 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:
+ if !k.w.take() {
select {
case <-t.after(d):
k.send(ctx, t, c, id)
return false
- case <-t.interrupts:
+ case <-k.w.ch:
+ k.w.n++
+ k.w.handled++
}
}
- if k.asked {
- k.send(ctx, t, c, id)
+ if k.press(ctx, t, c, id) {
+ return true
+ }
+ // A second Ctrl-C already waiting is not left for the next poll.
+ if k.w.take() {
+ k.press(ctx, t, c, id)
return true
}
- k.asked = true
- t.say(fmt.Sprintf("cancelling run %s. press Ctrl-C again to stop waiting", id))
- k.send(ctx, t, c, id)
return false
}
diff --git a/run_test.go b/run_test.go
index c9031fb8a..07da6011c 100644
--- a/run_test.go
+++ b/run_test.go
@@ -812,27 +812,63 @@ func TestFollowRunInterrupts(t *testing.T) {
})
}
-// gatedRun holds one page call open, as a request in flight: entered is
-// closed when the call arrives, and the call returns when release is closed.
+// gatedRun holds one request open, as a request in flight: entered is closed
+// when the held call arrives, and the call returns when release is closed or
+// its context is cancelled, as a real request does.
type gatedRun struct {
*scriptedRun
- holdCall int // the page call (1-based) to hold
- entered chan struct{}
- release chan struct{}
- ctxErr error // the held call's context error when it returned
+ 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{}
+ 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
+}
+
+func (g *gatedRun) hold(ctx context.Context) error {
+ g.held++
+ close(g.entered)
+ select {
+ case <-g.release:
+ case <-ctx.Done():
+ }
+ g.ctxErr = ctx.Err()
+ return g.ctxErr
}
func (g *gatedRun) page(ctx context.Context, id string, after int64) (runs.Page, error) {
- if g.pagesRead()+1 == g.holdCall {
- close(g.entered)
- <-g.release
- g.ctxErr = ctx.Err()
+ if g.pagesRead()+1 == g.holdPage {
+ if err := g.hold(ctx); err != nil {
+ g.scriptedRun.mu.Lock()
+ g.scriptedRun.pageCalls++
+ g.scriptedRun.mu.Unlock()
+ return runs.Page{}, err
+ }
}
return g.scriptedRun.page(ctx, id, after)
}
-func newGatedRun(s *scriptedRun, holdCall int) *gatedRun {
- return &gatedRun{scriptedRun: s, holdCall: holdCall, entered: make(chan struct{}), release: make(chan struct{})}
+func (g *gatedRun) cancel(ctx context.Context, id string) (string, error) {
+ if g.holdCancel && g.held == 0 {
+ g.scriptedRun.mu.Lock()
+ g.scriptedRun.cancels++
+ g.scriptedRun.mu.Unlock()
+ return "", g.hold(ctx)
+ }
+ return g.scriptedRun.cancel(ctx, id)
+}
+
+func (g *gatedRun) ack(ctx context.Context, id string, last int64) error {
+ if g.holdAck && g.held == 0 {
+ g.scriptedRun.acked = append(g.scriptedRun.acked, last)
+ return g.hold(ctx)
+ }
+ return g.scriptedRun.ack(ctx, id, last)
+}
+
+func newGatedRun(s *scriptedRun) *gatedRun {
+ return &gatedRun{scriptedRun: s, entered: make(chan struct{}), release: make(chan struct{})}
}
func awaitFollowRun(t *testing.T, done <-chan int, what string) int {
@@ -896,16 +932,34 @@ func TestFollowRunCtrlCOnceCancelsTwiceExits(t *testing.T) {
}
}
-func TestFollowRunCtrlCWhileAPageIsInFlight(t *testing.T) {
+func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
picked := runs.Page{State: "picked"}
done := runs.Page{State: "done", Result: &runs.Result{Cancelled: true, Started: time.Unix(1_700_000_000, 0)}}
- t.Run("the first is taken when the request returns, before the next poll", func(t *testing.T) {
- f := newTermFx()
- g := newGatedRun(&scriptedRun{pages: []runs.Page{picked, done}}, 1)
+ start := func(g *gatedRun, f *termFx) <-chan int {
codes := make(chan int, 1)
go func() {
codes <- followRun(context.Background(), g, f.term(), "shop.example.com", "run_a", time.Time{})
}()
+ return codes
+ }
+ // promptly is how long a second Ctrl-C may take to end a request that
+ // never answers; the held request would otherwise wait for ever.
+ const promptly = 500 * time.Millisecond
+ quick := func(t *testing.T, codes <-chan int) int {
+ t.Helper()
+ select {
+ case code := <-codes:
+ return code
+ case <-time.After(promptly):
+ t.Fatal("the second Ctrl-C did not end the request in flight")
+ return 0
+ }
+ }
+ t.Run("the first is taken when the request returns, before the next poll", func(t *testing.T) {
+ f := newTermFx()
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked, done}})
+ g.holdPage = 1
+ codes := start(g, f)
awaitSignal(t, g.entered, "the first page request was never made")
f.sig <- os.Interrupt
// The request is not abandoned, and nothing is cancelled until it is back.
@@ -929,32 +983,86 @@ func TestFollowRunCtrlCWhileAPageIsInFlight(t *testing.T) {
t.Fatalf("stderr %q", got)
}
})
- t.Run("the second is taken when the request returns, with no further poll", func(t *testing.T) {
+ t.Run("the second ends a page request in flight at once", func(t *testing.T) {
f := newTermFx()
f.sig <- os.Interrupt // the first, handled after the first page
- g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}}, 2)
- codes := make(chan int, 1)
- go func() {
- codes <- followRun(context.Background(), g, f.term(), "shop.example.com", "run_a", time.Time{})
- }()
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
+ g.holdPage = 2
+ codes := start(g, f)
awaitSignal(t, g.entered, "the second page request was never made")
if got := g.cancelsMade(); got != 1 {
t.Fatalf("%d cancels before the second page request, want 1", got)
}
- f.sig <- os.Interrupt
- close(g.release)
- if code := awaitFollowRun(t, codes, "the second Ctrl-C did not stop waiting"); code != 130 {
+ f.sig <- os.Interrupt // the request is never released
+ if code := quick(t, codes); code != 130 {
t.Fatalf("exit %d, want 130", code)
}
+ if !errors.Is(g.ctxErr, context.Canceled) {
+ t.Fatalf("the request in flight ended with %v, want it cut off", g.ctxErr)
+ }
if got := g.pagesRead(); got != 2 {
t.Fatalf("%d page requests, want 2: no poll after the second Ctrl-C", got)
}
if got := g.cancelsMade(); got != 1 {
- t.Fatalf("%d cancels, want 1", got)
+ t.Fatalf("%d cancels, want 1: the one already sent stays sent and is not repeated", got)
}
if len(g.acked) != 0 {
t.Fatalf("acked %v", g.acked)
}
+ 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("both arriving during one page request still cancel, then stop", func(t *testing.T) {
+ f := newTermFx()
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
+ g.holdPage = 1
+ codes := start(g, f)
+ awaitSignal(t, g.entered, "the first page request was never made")
+ f.sig <- os.Interrupt
+ f.sig <- os.Interrupt
+ if code := quick(t, codes); code != 130 {
+ t.Fatalf("exit %d, want 130", code)
+ }
+ if got := g.cancelsMade(); got != 1 {
+ t.Fatalf("%d cancels, want 1: the first Ctrl-C still asks for the cancel", got)
+ }
+ 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("the second ends a cancel request in flight and does not send it again", func(t *testing.T) {
+ f := newTermFx()
+ f.sig <- os.Interrupt
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
+ g.holdCancel = true
+ codes := start(g, f)
+ awaitSignal(t, g.entered, "the cancel request was never made")
+ f.sig <- os.Interrupt
+ if code := quick(t, codes); code != 130 {
+ t.Fatalf("exit %d, want 130", code)
+ }
+ if got := g.cancelsMade(); got != 1 {
+ t.Fatalf("%d cancels, want 1", got)
+ }
+ if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
+ 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) {
+ f := newTermFx()
+ g := newGatedRun(&scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{LastSeq: 3}}}})
+ g.holdAck = true
+ codes := start(g, f)
+ awaitSignal(t, g.entered, "the ack request was never made")
+ f.sig <- os.Interrupt
+ f.sig <- os.Interrupt
+ if code := quick(t, codes); code != 130 {
+ t.Fatalf("exit %d, want 130", code)
+ }
+ if got := g.cancelsMade(); got != 1 {
+ t.Fatalf("%d cancels, want 1", got)
+ }
})
}