carlos / platform Public

branch:

Clone

git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1	package main
2
3	import (
4	"bytes"
5	"context"
6	"crypto/sha256"
7	"encoding/hex"
8	"encoding/json"
9	"errors"
10	"io"
11	"mime/multipart"
12	"net/http"
13	"net/http/httptest"
14	"net/url"
15	"os"
16	"path/filepath"
17	"slices"
18	"strconv"
19	"strings"
20	"sync"
21	"sync/atomic"
22	"testing"
23	"time"
24
25	"github.com/carlosframework/platform/internal/runs"
26	)
27
28	func TestParseRunArgs(t *testing.T) {
29	ok, err := parseRunArgs([]string{"--app", "shop", "--instance", "h.example.com", "--file", "a.csv", "--file", "b.csv", "--timeout", "2m", "--", "import", "--dry-run"}, io.Discard)
30	if err != nil {
31	t.Fatal(err)
32	}
33	if ok.app != "shop" \|\| ok.instance != "h.example.com" \|\| len(ok.files) != 2 \|\| ok.timeout != 2*time.Minute \|\| strings.Join(ok.argv, " ") != "import --dry-run" {
34	t.Fatalf("opts %+v", ok)
35	}
36	def, err := parseRunArgs([]string{"--instance", "h.example.com", "--", "x"}, io.Discard)
37	if err != nil \|\| def.timeout != 30*time.Minute {
38	t.Fatalf("default timeout %s, %v", def.timeout, err)
39	}
40	for args, want := range map[string]string{
41	"--app shop --instance h.example.com import --dry-run": "",
42	"--app shop -- import": "--app and --instance name the instance to run against",
43	"--instance h.example.com --timeout 7h -- x": "--timeout must be between 1s and 6h",
44	"--instance h.example.com --timeout 0s -- x": "--timeout must be between 1s and 6h",
45	"--instance h.example.com --follow run_x -- x": "--follow takes a run id and no command after --",
46	"--instance h.example.com --follow run_x --file a": "--follow takes a run id and no command after --",
47	} {
48	_, err := parseRunArgs(strings.Fields(args), io.Discard)
49	if err == nil {
50	t.Errorf("%q accepted", args)
51	continue
52	}
53	if want == "" && !errors.Is(err, errRunUsage) {
54	t.Errorf("%q: %v, want the usage error", args, err)
55	}
56	if want != "" && err.Error() != want {
57	t.Errorf("%q: %q, want %q", args, err, want)
58	}
59	}
60	}
61
62	func TestRunMetaForStatsEveryFile(t *testing.T) {
63	dir := t.TempDir()
64	p := filepath.Join(dir, "checkins.csv")
65	if err := os.WriteFile(p, []byte("a,b\n"), 0o600); err != nil {
66	t.Fatal(err)
67	}
68	m, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: 90 * time.Second, files: []string{p}})
69	if err != nil \|\| m.TimeoutS != 90 \|\| len(m.Files) != 1 \|\| m.Files[0].Name != "checkins.csv" \|\| m.Files[0].Size != 4 {
70	t.Fatalf("meta %+v, %v", m, err)
71	}
72	missing := filepath.Join(dir, "nope.csv")
73	if _, err := runMetaFor(runOpts{files: []string{missing}}); !errors.Is(err, errRunLocalFile) \|\| !strings.HasPrefix(err.Error(), "could not read "+missing+": ") {
74	t.Fatalf("missing file: %v", err)
75	}
76	}
77
78	func TestRunAPIWireShapes(t *testing.T) {
79	type seen struct{ method, path, auth, body string }
80	var got []seen
81	var metaJSON, fileName, fileBody string
82	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
83	s := seen{method: r.Method, path: r.URL.RequestURI(), auth: r.Header.Get("Authorization")}
84	if strings.HasPrefix(r.Header.Get("Content-Type"), "multipart/") {
85	mr, err := r.MultipartReader()
86	if err != nil {
87	t.Fatal(err)
88	}
89	for {
90	part, err := mr.NextPart()
91	if err != nil {
92	break
93	}
94	raw, _ := io.ReadAll(part)
95	if part.FormName() == "meta" {
96	metaJSON = string(raw)
97	} else {
98	fileName, fileBody = part.FileName(), string(raw)
99	}
100	}
101	got = append(got, s)
102	w.WriteHeader(http.StatusAccepted)
103	json.NewEncoder(w).Encode(runs.Created{ID: "run_a", Host: "h.example.com", State: "queued"})
104	return
105	}
106	raw, _ := io.ReadAll(r.Body)
107	s.body = string(raw)
108	got = append(got, s)
109	switch {
110	case strings.Contains(s.path, "/expire"):
111	w.WriteHeader(http.StatusConflict)
112	json.NewEncoder(w).Encode(map[string]string{"error": "run run_a has already started"})
113	case r.Method == http.MethodGet:
114	json.NewEncoder(w).Encode(runs.Page{ID: "run_a", State: "picked"})
115	default:
116	json.NewEncoder(w).Encode(runs.StateAnswer{State: "cancel"})
117	}
118	}))
119	defer ts.Close()
120	dir := t.TempDir()
121	p := filepath.Join(dir, "x.csv")
122	os.WriteFile(p, []byte("1,2"), 0o600)
123	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
124	ctx := context.Background()
125	created, err := c.create(ctx, runs.Meta{Argv: []string{"import"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "x.csv", Size: 3}}}, []string{p})
126	if err != nil \|\| created.ID != "run_a" {
127	t.Fatalf("create %+v, %v", created, err)
128	}
129	if !strings.Contains(metaJSON, `"argv":["import"]`) \|\| fileName != "x.csv" \|\| fileBody != "1,2" {
130	t.Fatalf("meta %s file %s %q", metaJSON, fileName, fileBody)
131	}
132	if _, err := c.page(ctx, "run_a", 3); err != nil {
133	t.Fatal(err)
134	}
135	if st, err := c.cancel(ctx, "run_a"); err != nil \|\| st != "cancel" {
136	t.Fatalf("cancel %q, %v", st, err)
137	}
138	_, err = c.expire(ctx, "run_a")
139	var ce *runConsoleError
140	if !errors.As(err, &ce) \|\| ce.status != http.StatusConflict \|\| ce.msg != "run run_a has already started" {
141	t.Fatalf("expire %v", err)
142	}
143	if err := c.ack(ctx, "run_a", 7); err != nil {
144	t.Fatal(err)
145	}
146	base := "/api/cli/apps/acme/shop/instances/h.example.com/runs"
147	want := []seen{
148	{http.MethodPost, base, "Bearer cli-token", ""},
149	{http.MethodGet, base + "/run_a?after=3", "Bearer cli-token", ""},
150	{http.MethodPost, base + "/run_a/cancel", "Bearer cli-token", "{}"},
151	{http.MethodPost, base + "/run_a/expire", "Bearer cli-token", "{}"},
152	{http.MethodPost, base + "/run_a/ack", "Bearer cli-token", `{"last_seq":7}`},
153	}
154	if len(got) != len(want) {
155	t.Fatalf("requests %+v", got)
156	}
157	for i := range want {
158	if got[i] != want[i] {
159	t.Errorf("request %d = %+v, want %+v", i, got[i], want[i])
160	}
161	}
162	}
163
164	func TestRunMetaForRefusesWhatIsNotAFile(t *testing.T) {
165	dir := t.TempDir()
166	_, err := runMetaFor(runOpts{files: []string{dir}})
167	if !errors.Is(err, errRunLocalFile) \|\| !errors.Is(err, errRunNotRegular) \|\| !strings.HasPrefix(err.Error(), "could not read "+dir+": ") {
168	t.Fatalf("directory: %v", err)
169	}
170	}
171
172	// A part that disagrees with meta must never reach the console as a whole
173	// request: the file changed between the stat and the upload.
174	func TestRunAPICreateFailsWhenAFileChangesSize(t *testing.T) {
175	for name, tc := range map[string]struct {
176	content string
177	size int64
178	}{
179	"shrunk": {"12", 3},
180	"grown": {"12345", 3},
181	} {
182	t.Run(name, func(t *testing.T) {
183	var complete atomic.Bool
184	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
185	mr, err := r.MultipartReader()
186	if err != nil {
187	return
188	}
189	for {
190	part, err := mr.NextPart()
191	if err == io.EOF {
192	break
193	}
194	if err != nil {
195	return
196	}
197	if _, err := io.Copy(io.Discard, part); err != nil {
198	return
199	}
200	}
201	complete.Store(true)
202	w.WriteHeader(http.StatusAccepted)
203	json.NewEncoder(w).Encode(runs.Created{ID: "run_a"})
204	}))
205	p := filepath.Join(t.TempDir(), "x.csv")
206	if err := os.WriteFile(p, []byte(tc.content), 0o600); err != nil {
207	t.Fatal(err)
208	}
209	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
210	_, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "x.csv", Size: tc.size}}}, []string{p})
211	// Close waits for the handler, so complete is settled when read.
212	ts.Close()
213	var uerr *url.Error
214	if !errors.Is(err, errRunLocalFile) \|\| errors.As(err, &uerr) \|\| !strings.HasPrefix(err.Error(), "could not read "+p+": ") {
215	t.Fatalf("create: %v, want the bare could-not-read error", err)
216	}
217	if complete.Load() {
218	t.Fatal("the console received a whole request whose part disagrees with meta")
219	}
220	})
221	}
222	}
223
224	// fakeRunFile is an open file whose reader and reported sizes the test
225	// controls.
226	type fakeRunFile struct {
227	io.Reader
228	mode os.FileMode
229	sizes []int64 // what successive Stat calls report; the last repeats
230	}
231
232	type fakeInfo struct {
233	os.FileInfo
234	mode os.FileMode
235	size int64
236	}
237
238	func (i fakeInfo) Mode() os.FileMode { return i.mode }
239	func (i fakeInfo) Size() int64 { return i.size }
240
241	func (f *fakeRunFile) Stat() (os.FileInfo, error) {
242	size := f.sizes[0]
243	if len(f.sizes) > 1 {
244	f.sizes = f.sizes[1:]
245	}
246	return fakeInfo{mode: f.mode, size: size}, nil
247	}
248
249	func TestCopyRunFilePartBranches(t *testing.T) {
250	want := runs.MetaFile{Name: "x.csv", Size: 4}
251	for name, tc := range map[string]struct {
252	f *fakeRunFile
253	fail error // nil: the part is written
254	}{
255	"as promised": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{4, 4}}, nil},
256	"stat disagrees": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{5}}, errRunFileChanged},
257	"reader short": {&fakeRunFile{strings.NewReader("12"), 0, []int64{4, 4}}, errRunFileChanged},
258	"grown after the copy": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{4, 9}}, errRunFileChanged},
259	"not a regular file": {&fakeRunFile{strings.NewReader("1234"), os.ModeDir, []int64{4}}, errRunNotRegular},
260	"reader fails midway": {&fakeRunFile{io.MultiReader(strings.NewReader("12"), iotestErrReader{}), 0, []int64{4, 4}}, errRunReaderBroke},
261	} {
262	t.Run(name, func(t *testing.T) {
263	var buf bytes.Buffer
264	mw := multipart.NewWriter(&buf)
265	err := copyRunFilePart(mw, tc.f, "/data/x.csv", want)
266	if tc.fail == nil {
267	if err != nil \|\| !strings.Contains(buf.String(), "1234") {
268	t.Fatalf("err %v, body %q", err, buf.String())
269	}
270	return
271	}
272	if !errors.Is(err, tc.fail) \|\| !errors.Is(err, errRunLocalFile) \|\| !strings.HasPrefix(err.Error(), "could not read /data/x.csv: ") {
273	t.Fatalf("err %v, want %v as a local file error", err, tc.fail)
274	}
275	})
276	}
277	}
278
279	var errRunReaderBroke = errors.New("disk fell over")
280
281	type iotestErrReader struct{}
282
283	func (iotestErrReader) Read([]byte) (int, error) { return 0, errRunReaderBroke }
284
285	// The wire order the console's door requires: meta first, then one file
286	// part per meta file in meta's order, each named as meta names it.
287	func TestRunAPICreateSendsMetaThenFilesInOrder(t *testing.T) {
288	type part struct{ form, name, body string }
289	var got []part
290	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
291	mr, err := r.MultipartReader()
292	if err != nil {
293	t.Error(err)
294	return
295	}
296	for {
297	p, err := mr.NextPart()
298	if err != nil {
299	break
300	}
301	raw, _ := io.ReadAll(p)
302	got = append(got, part{p.FormName(), p.FileName(), string(raw)})
303	}
304	w.WriteHeader(http.StatusAccepted)
305	json.NewEncoder(w).Encode(runs.Created{ID: "run_a"})
306	}))
307	defer ts.Close()
308	dir := t.TempDir()
309	var paths []string
310	var files []runs.MetaFile
311	for _, f := range []struct{ name, body string }{{"b.csv", "bb"}, {"a.csv", "a"}} {
312	p := filepath.Join(dir, f.name)
313	if err := os.WriteFile(p, []byte(f.body), 0o600); err != nil {
314	t.Fatal(err)
315	}
316	paths = append(paths, p)
317	files = append(files, runs.MetaFile{Name: f.name, Size: int64(len(f.body))})
318	}
319	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
320	if _, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60, Files: files}, paths); err != nil {
321	t.Fatal(err)
322	}
323	if len(got) != 3 \|\| got[0].form != "meta" \|\| got[0].name != "" \|\|
324	got[1] != (part{"file", "b.csv", "bb"}) \|\| got[2] != (part{"file", "a.csv", "a"}) {
325	t.Fatalf("parts %+v", got)
326	}
327	}
328
329	func TestRunAPICreateRefusesFilesThatDisagreeWithMeta(t *testing.T) {
330	c := runAPI{s: newTestSession("http://127.0.0.1:1"), a: apiAcct{Seg: "acme"}, app: "a", host: "h.example.com"}
331	_, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60}, []string{"/nowhere"})
332	if err == nil {
333	t.Fatal("create accepted more paths than meta files")
334	}
335	}
336
337	// scriptedRun is a runClient that answers pages from a script.
338	type scriptedRun struct {
339	pages []runs.Page
340	pageErr error // every page call fails with it
341	pageErrs []error // the first page calls fail with these, in order
342	cancelErrs []error // one per cancel call, in order; cancelErr once spent
343	expireTo string
344	expireErr error
345	cancelErr error
346	createErr error
347	ackErrs []error // one per ack call, in order; nil once spent
348	acked []int64
349	cancels int
350	expires int
351	created runs.Meta
352
353	// mu guards cancels and pageCalls, which a test reads from its own
354	// goroutine while followRun's goroutine writes them.
355	mu sync.Mutex
356	pageCalls int
357	}
358
359	// cancelsMade and pagesRead are safe to call while followRun is running.
360	func (s *scriptedRun) cancelsMade() int {
361	s.mu.Lock()
362	defer s.mu.Unlock()
363	return s.cancels
364	}
365
366	func (s *scriptedRun) pagesRead() int {
367	s.mu.Lock()
368	defer s.mu.Unlock()
369	return s.pageCalls
370	}
371
372	func (s *scriptedRun) create(_ context.Context, m runs.Meta, _ []string) (runs.Created, error) {
373	s.created = m
374	if s.createErr != nil {
375	return runs.Created{}, s.createErr
376	}
377	return runs.Created{ID: "run_a", State: "queued"}, nil
378	}
379	func (s *scriptedRun) page(context.Context, string, int64) (runs.Page, error) {
380	s.mu.Lock()
381	s.pageCalls++
382	s.mu.Unlock()
383	if len(s.pageErrs) > 0 {
384	err := s.pageErrs[0]
385	s.pageErrs = s.pageErrs[1:]
386	if err != nil { // nil: this call is answered from pages
387	return runs.Page{}, err
388	}
389	}
390	if s.pageErr != nil {
391	return runs.Page{}, s.pageErr
392	}
393	p := s.pages[0]
394	if len(s.pages) > 1 {
395	s.pages = s.pages[1:]
396	}
397	return p, nil
398	}
399	func (s *scriptedRun) cancel(context.Context, string) (string, error) {
400	s.mu.Lock()
401	s.cancels++
402	s.mu.Unlock()
403	if len(s.cancelErrs) > 0 {
404	err := s.cancelErrs[0]
405	s.cancelErrs = s.cancelErrs[1:]
406	return "cancel", err
407	}
408	return "cancel", s.cancelErr
409	}
410	func (s *scriptedRun) expire(context.Context, string) (string, error) {
411	s.expires++
412	return s.expireTo, s.expireErr
413	}
414	func (s *scriptedRun) ack(_ context.Context, _ string, last int64) error {
415	s.acked = append(s.acked, last)
416	if len(s.ackErrs) == 0 {
417	return nil
418	}
419	err := s.ackErrs[0]
420	s.ackErrs = s.ackErrs[1:]
421	return err
422	}
423
424	type termFx struct {
425	out, err strings.Builder
426	sig chan os.Signal
427	now time.Time
428	}
429
430	func newTermFx() *termFx {
431	return &termFx{sig: make(chan os.Signal, 2), now: time.Unix(1_700_000_000, 0)}
432	}
433
434	func (f *termFx) term() runTerm {
435	return runTerm{stdout: &f.out, stderr: &f.err, interrupts: f.sig,
436	after: func(d time.Duration) <-chan time.Time {
437	f.now = f.now.Add(d)
438	c := make(chan time.Time, 1)
439	c <- f.now
440	return c
441	},
442	now: func() time.Time { return f.now }}
443	}
444
445	func chunk(seq int64, stream, data string) runs.Chunk {
446	return runs.Chunk{Seq: seq, Stream: stream, Data: []byte(data)}
447	}
448
449	func TestFollowRunInterleavesAndAcks(t *testing.T) {
450	f := newTermFx()
451	s := &scriptedRun{pages: []runs.Page{
452	{State: "picked", Chunks: []runs.Chunk{chunk(1, "o", "one\n"), chunk(2, "e", "warn\n")}},
453	{State: "picked", Chunks: []runs.Chunk{chunk(3, "o", "two\n")}, Result: &runs.Result{Exit: 0, LastSeq: 3}},
454	}}
455	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 0 {
456	t.Fatalf("exit %d", code)
457	}
458	if f.out.String() != "one\ntwo\n" \|\| f.err.String() != "warn\n" {
459	t.Fatalf("stdout %q stderr %q", f.out.String(), f.err.String())
460	}
461	if len(s.acked) != 1 \|\| s.acked[0] != 3 {
462	t.Fatalf("acked %v", s.acked)
463	}
464	}
465
466	func TestFollowRunPagesUntilTheResult(t *testing.T) {
467	f := newTermFx()
468	res := &runs.Result{Exit: 0, LastSeq: 3}
469	s := &scriptedRun{pages: []runs.Page{
470	{State: "done", Chunks: []runs.Chunk{chunk(1, "o", "a")}, Result: res, More: true},
471	{State: "done", Chunks: []runs.Chunk{chunk(2, "o", "b")}, Result: res, More: true},
472	{State: "done", Chunks: []runs.Chunk{chunk(3, "o", "c")}, Result: res},
473	}}
474	followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
475	if f.out.String() != "abc" \|\| len(s.acked) != 1 \|\| s.acked[0] != 3 {
476	t.Fatalf("stdout %q acked %v (every page before the ack)", f.out.String(), s.acked)
477	}
478	}
479
480	func TestFollowRunExitCodes(t *testing.T) {
481	started := time.Unix(1_700_000_000, 0)
482	for _, c := range []struct {
483	res runs.Result
484	code int
485	says string
486	}{
487	{runs.Result{Exit: 7}, 7, ""},
488	{runs.Result{Signal: "9"}, 137, ""},
489	{runs.Result{TimedOut: true, Started: started, Ended: started.Add(30 * time.Minute)}, 124, "shop.example.com: the command timed out after 30m0s"},
490	{runs.Result{Interrupted: "the instance was restarted"}, 125, "shop.example.com: the run was interrupted: the instance was restarted. it may have done part of its work"},
491	{runs.Result{Interrupted: runs.InterruptedLost}, 125, "shop.example.com: the run was interrupted: its box stopped reporting. it may have done part of its work"},
492	{runs.Result{Cancelled: true, Started: started}, 130, "shop.example.com: the command was cancelled"},
493	{runs.Result{Refused: runs.RefusedExpired}, 3, "shop.example.com: not picked up within 60 seconds. the run did not start. try again"},
494	{runs.Result{Refused: runs.RefusedFlat}, 3, "shop.example.com: the box is not on the nested layout yet, so carlos run cannot run there"},
495	{runs.Result{Refused: runs.RefusedClaim}, 3, "shop.example.com: the box has host claims armed. carlos run cannot run there yet"},
496	{runs.Result{Refused: runs.RefusedWake, Detail: "wake failed: restore"}, 3, "shop.example.com: the instance did not wake. wake failed: restore"},
497	{runs.Result{Refused: runs.RefusedIncarnation}, 3, "shop.example.com was deleted and created again after this run was requested. the run did not start"},
498	{runs.Result{Refused: runs.RefusedImport}, 3, "shop.example.com is an import that is not committed yet. the run did not start"},
499	{runs.Result{Exit: 0, Truncated: true}, 0, "shop.example.com: the output passed 64 MB and was cut off there. the command ran on to its end"},
500	{runs.Result{Exit: 5, Truncated: true}, 5, "shop.example.com: the output passed 64 MB and was cut off there. the command ran on to its end"},
501	} {
502	code, lines := runExit("shop.example.com", c.res)
503	if code != c.code \|\| (c.says != "" && !slices.Contains(lines, c.says)) \|\| (c.says == "" && len(lines) != 0) {
504	t.Errorf("%+v: %d %q, want %d %q", c.res, code, lines, c.code, c.says)
505	}
506	}
507	}
508
509	func TestFollowRunTellsCancelledBeforeStartFromCancelledAfter(t *testing.T) {
510	f := newTermFx()
511	s := &scriptedRun{pages: []runs.Page{{State: "cancelled", Result: &runs.Result{Cancelled: true, LastSeq: 0}}}}
512	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 {
513	t.Fatalf("exit %d", code)
514	}
515	if got := f.err.String(); got != "shop.example.com: run run_a was cancelled before it started\n" {
516	t.Fatalf("stderr %q", got)
517	}
518	if len(s.acked) != 1 {
519	t.Fatalf("acked %v", s.acked)
520	}
521	}
522
523	// A run whose deadline passed before any box launched it has no start
524	// time: it is told as timed out before it started, never with a duration
525	// measured from the zero time (re-review 8). Exit 124 either way.
526	func TestFollowRunTellsTimedOutBeforeStartFromTimedOutAfter(t *testing.T) {
527	started := time.Unix(1_700_000_000, 0)
528	for _, c := range []struct {
529	res runs.Result
530	says string
531	}{
532	{runs.Result{TimedOut: true, Ended: started}, "shop.example.com: run run_a timed out before it started\n"},
533	{runs.Result{TimedOut: true, Started: started, Ended: started.Add(30 * time.Minute)}, "shop.example.com: the command timed out after 30m0s\n"},
534	} {
535	f := newTermFx()
536	s := &scriptedRun{pages: []runs.Page{{State: "done", Result: &c.res}}}
537	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 124 {
538	t.Fatalf("%+v: exit %d", c.res, code)
539	}
540	if got := f.err.String(); got != c.says {
541	t.Fatalf("%+v: stderr %q, want %q", c.res, got, c.says)
542	}
543	}
544	}
545
546	func TestFollowRunExpiresAQueuedRunAfterSixtySeconds(t *testing.T) {
547	f := newTermFx()
548	s := &scriptedRun{pages: []runs.Page{{State: "queued"}}, expireTo: "expired"}
549	code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", f.now.Add(60*time.Second))
550	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") {
551	t.Fatalf("exit %d expires %d stderr %q", code, s.expires, f.err.String())
552	}
553	if len(s.acked) != 1 \|\| s.acked[0] != 0 {
554	t.Fatalf("acked %v (the uploaded files go now, not with the sweep)", s.acked)
555	}
556	if f.now.Sub(time.Unix(1_700_000_000, 0)) < 60*time.Second {
557	t.Fatal("expired before 60 seconds")
558	}
559	}
560
561	func TestFollowRunKeepsFollowingWhenExpiryLoses(t *testing.T) {
562	f := newTermFx()
563	queued := runs.Page{State: "queued"}
564	pages := make([]runs.Page, 0, 62)
565	for range 61 {
566	pages = append(pages, queued)
567	}
568	pages = append(pages, runs.Page{State: "done", Result: &runs.Result{Exit: 4}})
569	s := &scriptedRun{pages: pages, expireErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already started"}}
570	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", f.now.Add(60*time.Second)); code != 4 \|\| s.expires != 1 {
571	t.Fatalf("exit %d expires %d (a late pickup wins over the expiry)", code, s.expires)
572	}
573	}
574
575	func TestFollowRunAsksAgainWhenTheExpiryCallFailsToArrive(t *testing.T) {
576	f := newTermFx()
577	s := &scriptedRun{pages: []runs.Page{{State: "queued"}}, expireTo: "expired"}
578	calls := 0
579	c := &countingExpire{scriptedRun: s, fail: 2, calls: &calls}
580	if code := followRun(context.Background(), c, f.term(), "shop.example.com", "run_a", f.now.Add(time.Second)); code != 3 \|\| calls != 3 {
581	t.Fatalf("exit %d, expire calls %d", code, calls)
582	}
583	}
584
585	// countingExpire fails its first `fail` expire calls as a lost connection.
586	type countingExpire struct {
587	*scriptedRun
588	fail int
589	calls *int
590	}
591
592	func (c *countingExpire) expire(ctx context.Context, id string) (string, error) {
593	*c.calls++
594	if *c.calls <= c.fail {
595	return "", errors.New("connection reset")
596	}
597	return c.scriptedRun.expire(ctx, id)
598	}
599
600	func TestFollowRunStatesWithoutAResult(t *testing.T) {
601	for state, want := range map[string]struct {
602	code int
603	says string
604	}{
605	"expired": {3, "shop.example.com: not picked up within 60 seconds. the run did not start. try again"},
606	"cancelled": {130, "shop.example.com: run run_a was cancelled before it started"},
607	} {
608	f := newTermFx()
609	s := &scriptedRun{pages: []runs.Page{{State: state}}}
610	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != want.code \|\| !strings.Contains(f.err.String(), want.says) {
611	t.Errorf("%s: exit %d stderr %q", state, code, f.err.String())
612	}
613	}
614	}
615
616	// A page with no state and no result is a run the record no longer names.
617	// A result may still arrive, so the loop asks for a while, then gives up
618	// with RUN-C86 instead of polling until the sweep deletes the run.
619	func TestFollowRunGivesUpOnARunTheRecordLetGo(t *testing.T) {
620	t.Run("no result comes", func(t *testing.T) {
621	f := newTermFx()
622	start := f.now
623	s := &scriptedRun{pages: []runs.Page{{Chunks: []runs.Chunk{chunk(1, "o", "part\n")}}, {}}}
624	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 125 {
625	t.Fatalf("exit %d stderr %q", code, f.err.String())
626	}
627	if got := f.err.String(); got != "shop.example.com: the instance no longer holds run run_a, and no result came back. it may have done part of its work\n" {
628	t.Fatalf("stderr %q", got)
629	}
630	if waited := f.now.Sub(start); waited < runPageGiveUp \|\| waited > runPageGiveUp+2*time.Second {
631	t.Fatalf("gave up after %s, want %s", waited, runPageGiveUp)
632	}
633	if f.out.String() != "part\n" \|\| len(s.acked) != 0 {
634	t.Fatalf("stdout %q acked %v", f.out.String(), s.acked)
635	}
636	})
637	t.Run("a result arrives in time", func(t *testing.T) {
638	f := newTermFx()
639	pages := make([]runs.Page, 0, 91)
640	for range 90 {
641	pages = append(pages, runs.Page{})
642	}
643	pages = append(pages, runs.Page{Result: &runs.Result{Interrupted: runs.InterruptedLost}})
644	s := &scriptedRun{pages: pages}
645	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 125 \|\| !strings.Contains(f.err.String(), "its box stopped reporting") {
646	t.Fatalf("exit %d stderr %q", code, f.err.String())
647	}
648	if len(s.acked) != 1 {
649	t.Fatalf("acked %v", s.acked)
650	}
651	})
652	t.Run("a state in between starts the wait again", func(t *testing.T) {
653	f := newTermFx()
654	var pages []runs.Page
655	for range 100 {
656	pages = append(pages, runs.Page{})
657	}
658	pages = append(pages, runs.Page{State: "picked"})
659	for range 100 {
660	pages = append(pages, runs.Page{})
661	}
662	pages = append(pages, runs.Page{State: "done", Result: &runs.Result{Exit: 4}})
663	s := &scriptedRun{pages: pages}
664	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 4 {
665	t.Fatalf("exit %d stderr %q", code, f.err.String())
666	}
667	})
668	}
669
670	func TestFollowRunPageErrors(t *testing.T) {
671	for name, c := range map[string]struct {
672	err error
673	code int
674	says string
675	}{
676	"no run": {&runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"}, 2, "shop.example.com has no run run_a, or this console has no carlos run yet"},
677	"console 4xx": {&runConsoleError{status: http.StatusConflict, msg: "carlos run is not available here"}, 2, "carlos run: carlos run is not available here"},
678	"network": {errors.New("connection refused"), 1, "carlos run: connection refused"},
679	} {
680	f := newTermFx()
681	s := &scriptedRun{pageErr: c.err}
682	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != c.code \|\| f.err.String() != c.says+"\n" {
683	t.Errorf("%s: exit %d stderr %q", name, code, f.err.String())
684	}
685	}
686	}
687
688	func TestFollowRunRetriesAPageThatMayPass(t *testing.T) {
689	done := runs.Page{State: "done", Chunks: []runs.Chunk{chunk(1, "o", "out\n")}, Result: &runs.Result{Exit: 9, LastSeq: 1}}
690	t.Run("a 503 then success keeps following", func(t *testing.T) {
691	f := newTermFx()
692	start := f.now
693	s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{errors.New("could not read the run. try again"), errors.New("connection reset")}}
694	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 {
695	t.Fatalf("exit %d stderr %q", code, f.err.String())
696	}
697	if f.out.String() != "out\n" \|\| f.err.String() != "" \|\| len(s.acked) != 1 \|\| f.now.Sub(start) != 2*time.Second {
698	t.Fatalf("stdout %q stderr %q acked %v waited %s", f.out.String(), f.err.String(), s.acked, f.now.Sub(start))
699	}
700	})
701	t.Run("a 429 waits as long as Retry-After says", func(t *testing.T) {
702	f := newTermFx()
703	start := f.now
704	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}}}
705	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 \|\| f.now.Sub(start) != 17*time.Second {
706	t.Fatalf("exit %d waited %s", code, f.now.Sub(start))
707	}
708	})
709	t.Run("a 429 without Retry-After waits longer than a second, and no longer than a minute", func(t *testing.T) {
710	f := newTermFx()
711	start := f.now
712	s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{
713	&runConsoleError{status: http.StatusTooManyRequests, msg: "slow down"},
714	&runConsoleError{status: http.StatusTooManyRequests, msg: "slow down", retryAfter: time.Hour},
715	}}
716	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 \|\| f.now.Sub(start) != runRateWait+runRateWaitMax {
717	t.Fatalf("exit %d waited %s", code, f.now.Sub(start))
718	}
719	})
720	t.Run("a page that keeps failing is given up on after two minutes with exit 1", func(t *testing.T) {
721	f := newTermFx()
722	start := f.now
723	s := &scriptedRun{pageErr: errors.New("could not read the run. try again")}
724	code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
725	if waited := f.now.Sub(start); code != 1 \|\| waited < 2time.Minute \|\| waited > 2time.Minute+5*time.Second \|\| f.err.String() != "carlos run: could not read the run. try again\n" {
726	t.Fatalf("exit %d waited %s stderr %q", code, waited, f.err.String())
727	}
728	})
729	t.Run("the bound counts failures in a row", func(t *testing.T) {
730	f := newTermFx()
731	var errs []error
732	for range 100 {
733	errs = append(errs, errors.New("blip"))
734	}
735	// 100s of failures, a page that works, 100s more, then the result:
736	// 200s in all, but never two minutes in a row.
737	all := append(append(append([]error{}, errs...), nil), errs...)
738	s := &scriptedRun{pageErrs: append(all, nil), pages: []runs.Page{{State: "picked"}, done}}
739	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 {
740	t.Fatalf("exit %d stderr %q", code, f.err.String())
741	}
742	})
743	t.Run("a refusal that cannot pass is not retried", func(t *testing.T) {
744	f := newTermFx()
745	s := &scriptedRun{pageErrs: []error{&runConsoleError{status: http.StatusForbidden, msg: "no"}}, pages: []runs.Page{done}}
746	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 2 \|\| f.err.String() != "carlos run: no\n" {
747	t.Fatalf("exit %d stderr %q", code, f.err.String())
748	}
749	})
750	t.Run("a failing page does not stop Ctrl-C", func(t *testing.T) {
751	f := newTermFx()
752	f.sig <- os.Interrupt
753	f.sig <- os.Interrupt
754	s := &scriptedRun{pageErr: errors.New("connection refused")}
755	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 \|\| s.cancels < 1 {
756	t.Fatalf("exit %d cancels %d", code, s.cancels)
757	}
758	})
759	}
760
761	func TestFollowRunPacesAPageCutShort(t *testing.T) {
762	f := newTermFx()
763	start := f.now
764	res := &runs.Result{LastSeq: 3}
765	s := &scriptedRun{pages: []runs.Page{
766	{Chunks: []runs.Chunk{chunk(1, "o", "a")}, Result: res, More: true},
767	{Chunks: []runs.Chunk{chunk(2, "o", "b")}, Result: res, More: true},
768	{Chunks: []runs.Chunk{chunk(3, "o", "c")}, Result: res},
769	}}
770	followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
771	if f.out.String() != "abc" \|\| f.now.Sub(start) != 2*runMorePause {
772	t.Fatalf("stdout %q paused %s", f.out.String(), f.now.Sub(start))
773	}
774	}
775
776	func TestRunAPIPageReadsRetryAfter(t *testing.T) {
777	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
778	w.Header().Set("Retry-After", "7")
779	w.WriteHeader(http.StatusTooManyRequests)
780	json.NewEncoder(w).Encode(map[string]string{"error": "too many run requests. wait a minute and try again"})
781	}))
782	defer ts.Close()
783	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "shop.example.com"}
784	_, err := c.page(context.Background(), "run_a", 0)
785	var ce *runConsoleError
786	if !errors.As(err, &ce) \|\| ce.status != http.StatusTooManyRequests \|\| ce.retryAfter != 7*time.Second {
787	t.Fatalf("%#v", err)
788	}
789	}
790
791	func TestFollowRunAck(t *testing.T) {
792	notDone := &runConsoleError{status: http.StatusConflict, msg: "run run_a has not finished yet"}
793	gone := &runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"}
794	page := runs.Page{State: "done", Result: &runs.Result{Exit: 6, LastSeq: 2}}
795	t.Run("a lost 200 retried finds the objects gone and is success", func(t *testing.T) {
796	f := newTermFx()
797	s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{gone}}
798	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 \|\| f.err.String() != "" \|\| len(s.acked) != 1 {
799	t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked)
800	}
801	})
802	t.Run("an ack before the box's done is retried", func(t *testing.T) {
803	f := newTermFx()
804	s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{notDone, notDone}}
805	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 \|\| f.err.String() != "" \|\| len(s.acked) != 3 {
806	t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked)
807	}
808	})
809	t.Run("a box that never finishes is given up on, the exit code stands", func(t *testing.T) {
810	f := newTermFx()
811	errs := make([]error, 100)
812	for i := range errs {
813	errs[i] = notDone
814	}
815	s := &scriptedRun{pages: []runs.Page{page}, ackErrs: errs}
816	code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
817	if code != 6 \|\| len(s.acked) < 2 \|\| len(s.acked) > 20 \|\| !strings.Contains(f.err.String(), "carlos run: run run_a has not finished yet") {
818	t.Fatalf("exit %d acked %d stderr %q", code, len(s.acked), f.err.String())
819	}
820	})
821	t.Run("a refused ack is said and the exit code stands", func(t *testing.T) {
822	f := newTermFx()
823	s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{&runConsoleError{status: http.StatusConflict, msg: "run run_a ended at chunk 2, not 3"}}}
824	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 \|\| len(s.acked) != 1 \|\| f.err.String() != "carlos run: run run_a ended at chunk 2, not 3\n" {
825	t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked)
826	}
827	})
828	}
829
830	func TestFollowRunInterrupts(t *testing.T) {
831	queued := runs.Page{State: "picked"}
832	done := runs.Page{State: "done", Result: &runs.Result{Cancelled: true, Started: time.Unix(1_700_000_000, 0)}}
833	t.Run("the first cancels and keeps following", func(t *testing.T) {
834	f := newTermFx()
835	f.sig <- os.Interrupt
836	s := &scriptedRun{pages: []runs.Page{queued, done}}
837	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 \|\| s.cancels != 1 {
838	t.Fatalf("exit %d cancels %d", code, s.cancels)
839	}
840	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\nshop.example.com: the command was cancelled\n" {
841	t.Fatalf("stderr %q", got)
842	}
843	})
844	t.Run("the second stops waiting", func(t *testing.T) {
845	f := newTermFx()
846	f.sig <- os.Interrupt
847	f.sig <- os.Interrupt
848	s := &scriptedRun{pages: []runs.Page{queued}}
849	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 \|\| s.cancels != 1 \|\| len(s.acked) != 0 {
850	t.Fatalf("exit %d cancels %d acked %v", code, s.cancels, s.acked)
851	}
852	})
853	t.Run("a cancel that finds the run already ended is silent success", func(t *testing.T) {
854	f := newTermFx()
855	f.sig <- os.Interrupt
856	s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already ended"}}
857	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 {
858	t.Fatalf("exit %d", code)
859	}
860	if got := f.err.String(); strings.Contains(got, "already ended") {
861	t.Fatalf("stderr %q", got)
862	}
863	})
864	t.Run("a cancel that did not arrive is sent again on the next poll, and said once", func(t *testing.T) {
865	f := newTermFx()
866	f.sig <- os.Interrupt
867	s := &scriptedRun{pages: []runs.Page{queued, queued, queued, done}, cancelErrs: []error{errors.New("connection reset"), errors.New("connection reset"), nil}}
868	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 \|\| s.cancels != 3 {
869	t.Fatalf("exit %d cancels %d", code, s.cancels)
870	}
871	if got := strings.Count(f.err.String(), "carlos run: connection reset"); got != 1 {
872	t.Fatalf("said %d times: %q", got, f.err.String())
873	}
874	})
875	t.Run("the second Ctrl-C tries an unanswered cancel once more", func(t *testing.T) {
876	f := newTermFx()
877	f.sig <- os.Interrupt
878	f.sig <- os.Interrupt
879	s := &scriptedRun{pages: []runs.Page{queued}, cancelErrs: []error{errors.New("connection reset"), nil}}
880	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 \|\| s.cancels != 2 {
881	t.Fatalf("exit %d cancels %d", code, s.cancels)
882	}
883	})
884	t.Run("a cancel that failed is said", func(t *testing.T) {
885	f := newTermFx()
886	f.sig <- os.Interrupt
887	s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: errors.New("connection reset")}
888	followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
889	if !strings.Contains(f.err.String(), "carlos run: connection reset") {
890	t.Fatalf("stderr %q", f.err.String())
891	}
892	})
893	}
894
895	// gatedRun holds one request open, as a request in flight: entered is closed
896	// when the held call arrives, and the call returns when release is closed or
897	// its context is cancelled, as a real request does.
898	type gatedRun struct {
899	*scriptedRun
900	holdPage int // the page call (1-based) to hold, 0 for none
901	holdCancel int // the cancel call (1-based) to hold, 0 for none
902	holdAck bool // hold the first ack call
903	entered chan struct{} // the held page or ack call has arrived
904	cancelIn chan struct{} // the held cancel call has arrived
905	release chan struct{}
906	ctxErr error // the held page's context error when it returned
907	cancelCtx error // the held cancel's context error as it arrived
908	acksHeld int
909	}
910
911	func (g *gatedRun) hold(ctx context.Context, entered chan struct{}) error {
912	if entered == g.cancelIn {
913	g.cancelCtx = ctx.Err()
914	}
915	close(entered)
916	select {
917	case <-g.release:
918	case <-ctx.Done():
919	}
920	g.ctxErr = ctx.Err()
921	return g.ctxErr
922	}
923
924	func (g *gatedRun) page(ctx context.Context, id string, after int64) (runs.Page, error) {
925	if g.pagesRead()+1 == g.holdPage {
926	if err := g.hold(ctx, g.entered); err != nil {
927	g.scriptedRun.mu.Lock()
928	g.scriptedRun.pageCalls++
929	g.scriptedRun.mu.Unlock()
930	return runs.Page{}, err
931	}
932	}
933	return g.scriptedRun.page(ctx, id, after)
934	}
935
936	func (g *gatedRun) cancel(ctx context.Context, id string) (string, error) {
937	if g.holdCancel != 0 && g.cancelsMade()+1 == g.holdCancel {
938	g.scriptedRun.mu.Lock()
939	g.scriptedRun.cancels++
940	g.scriptedRun.mu.Unlock()
941	return "", g.hold(ctx, g.cancelIn)
942	}
943	return g.scriptedRun.cancel(ctx, id)
944	}
945
946	func (g *gatedRun) ack(ctx context.Context, id string, last int64) error {
947	if g.holdAck && g.acksHeld == 0 {
948	g.acksHeld++
949	g.scriptedRun.acked = append(g.scriptedRun.acked, last)
950	return g.hold(ctx, g.entered)
951	}
952	return g.scriptedRun.ack(ctx, id, last)
953	}
954
955	func newGatedRun(s scriptedRun) gatedRun {
956	return &gatedRun{scriptedRun: s, entered: make(chan struct{}), cancelIn: make(chan struct{}), release: make(chan struct{})}
957	}
958
959	func awaitFollowRun(t *testing.T, done <-chan int, what string) int {
960	t.Helper()
961	select {
962	case code := <-done:
963	return code
964	case <-time.After(2 * time.Second):
965	t.Fatal(what)
966	return 0
967	}
968	}
969
970	func awaitSignal(t *testing.T, ch <-chan struct{}, what string) {
971	t.Helper()
972	select {
973	case <-ch:
974	case <-time.After(2 * time.Second):
975	t.Fatal(what)
976	}
977	}
978
979	func TestFollowRunCtrlCOnceCancelsTwiceExits(t *testing.T) {
980	f := newTermFx()
981	s := &scriptedRun{pages: []runs.Page{{State: "picked"}}}
982	done := make(chan int, 1)
983	go func() { done <- followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}) }()
984	f.sig <- os.Interrupt
985	deadline := time.Now().Add(2 * time.Second)
986	for s.cancelsMade() == 0 && time.Now().Before(deadline) {
987	time.Sleep(5 * time.Millisecond)
988	}
989	if got := s.cancelsMade(); got != 1 {
990	t.Fatalf("the first Ctrl-C made %d cancels, want 1", got)
991	}
992	// The first Ctrl-C does not end the wait: the run is still being polled.
993	seen := s.pagesRead()
994	for s.pagesRead() < seen+3 && time.Now().Before(deadline) {
995	time.Sleep(5 * time.Millisecond)
996	}
997	if s.pagesRead() < seen+3 {
998	t.Fatal("followRun stopped polling after the first Ctrl-C")
999	}
1000	select {
1001	case code := <-done:
1002	t.Fatalf("followRun returned %d after one Ctrl-C", code)
1003	default:
1004	}
1005	f.sig <- os.Interrupt
1006	if code := awaitFollowRun(t, done, "the second Ctrl-C did not stop waiting"); code != 130 {
1007	t.Fatalf("exit %d, want 130", code)
1008	}
1009	if got := s.cancelsMade(); got != 1 {
1010	t.Fatalf("%d cancels, want 1: the second Ctrl-C does not cancel again", got)
1011	}
1012	if len(s.acked) != 0 {
1013	t.Fatalf("acked %v: nothing was read to the end", s.acked)
1014	}
1015	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
1016	t.Fatalf("stderr %q", got)
1017	}
1018	}
1019
1020	func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
1021	picked := runs.Page{State: "picked"}
1022	done := runs.Page{State: "done", Result: &runs.Result{Cancelled: true, Started: time.Unix(1_700_000_000, 0)}}
1023	start := func(g gatedRun, f termFx) <-chan int {
1024	codes := make(chan int, 1)
1025	go func() {
1026	codes <- followRun(context.Background(), g, f.term(), "shop.example.com", "run_a", time.Time{})
1027	}()
1028	return codes
1029	}
1030	// promptly is how long a second Ctrl-C may take to end a request that
1031	// never answers; the held request would otherwise wait for ever.
1032	const promptly = 500 * time.Millisecond
1033	quick := func(t *testing.T, codes <-chan int) int {
1034	t.Helper()
1035	select {
1036	case code := <-codes:
1037	return code
1038	case <-time.After(promptly):
1039	t.Fatal("the second Ctrl-C did not end the request in flight")
1040	return 0
1041	}
1042	}
1043	t.Run("the first is taken when the request returns, before the next poll", func(t *testing.T) {
1044	f := newTermFx()
1045	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked, done}})
1046	g.holdPage = 1
1047	codes := start(g, f)
1048	awaitSignal(t, g.entered, "the first page request was never made")
1049	f.sig <- os.Interrupt
1050	// Wait until the watcher has taken the Ctrl-C, then give the loop a
1051	// moment to act on it: it must not, while the request is open.
1052	for end := time.Now().Add(2 * time.Second); len(f.sig) > 0 && time.Now().Before(end); {
1053	time.Sleep(time.Millisecond)
1054	}
1055	time.Sleep(50 * time.Millisecond)
1056	// The request is not abandoned, and nothing is cancelled until it is back.
1057	if got := g.cancelsMade(); got != 0 {
1058	t.Fatalf("%d cancels while the page request was in flight", got)
1059	}
1060	close(g.release)
1061	if code := awaitFollowRun(t, codes, "the run was not followed to its result"); code != 130 {
1062	t.Fatalf("exit %d, want 130", code)
1063	}
1064	if g.ctxErr != nil {
1065	t.Fatalf("the request in flight was cut off: %v", g.ctxErr)
1066	}
1067	if got := g.cancelsMade(); got != 1 {
1068	t.Fatalf("%d cancels, want 1", got)
1069	}
1070	if got := g.pagesRead(); got != 2 {
1071	t.Fatalf("%d page requests, want 2: the one in flight and the one that carries the result", got)
1072	}
1073	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\nshop.example.com: the command was cancelled\n" {
1074	t.Fatalf("stderr %q", got)
1075	}
1076	})
1077	t.Run("the second ends a page request in flight at once", func(t *testing.T) {
1078	f := newTermFx()
1079	f.sig <- os.Interrupt // the first, handled after the first page
1080	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
1081	g.holdPage = 2
1082	codes := start(g, f)
1083	awaitSignal(t, g.entered, "the second page request was never made")
1084	if got := g.cancelsMade(); got != 1 {
1085	t.Fatalf("%d cancels before the second page request, want 1", got)
1086	}
1087	f.sig <- os.Interrupt // the request is never released
1088	if code := quick(t, codes); code != 130 {
1089	t.Fatalf("exit %d, want 130", code)
1090	}
1091	if !errors.Is(g.ctxErr, context.Canceled) {
1092	t.Fatalf("the request in flight ended with %v, want it cut off", g.ctxErr)
1093	}
1094	if got := g.pagesRead(); got != 2 {
1095	t.Fatalf("%d page requests, want 2: no poll after the second Ctrl-C", got)
1096	}
1097	if got := g.cancelsMade(); got != 1 {
1098	t.Fatalf("%d cancels, want 1: the one already sent stays sent and is not repeated", got)
1099	}
1100	if len(g.acked) != 0 {
1101	t.Fatalf("acked %v", g.acked)
1102	}
1103	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
1104	t.Fatalf("stderr %q", got)
1105	}
1106	})
1107	t.Run("both arriving during one page request still cancel, then stop", func(t *testing.T) {
1108	f := newTermFx()
1109	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
1110	g.holdPage = 1
1111	codes := start(g, f)
1112	awaitSignal(t, g.entered, "the first page request was never made")
1113	f.sig <- os.Interrupt
1114	f.sig <- os.Interrupt
1115	if code := quick(t, codes); code != 130 {
1116	t.Fatalf("exit %d, want 130", code)
1117	}
1118	if got := g.cancelsMade(); got != 1 {
1119	t.Fatalf("%d cancels, want 1: the first Ctrl-C still asks for the cancel", got)
1120	}
1121	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
1122	t.Fatalf("stderr %q", got)
1123	}
1124	})
1125	t.Run("the second ends a cancel request in flight and does not send it again", func(t *testing.T) {
1126	f := newTermFx()
1127	f.sig <- os.Interrupt
1128	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
1129	g.holdCancel = 1
1130	codes := start(g, f)
1131	awaitSignal(t, g.cancelIn, "the cancel request was never made")
1132	f.sig <- os.Interrupt
1133	if code := quick(t, codes); code != 130 {
1134	t.Fatalf("exit %d, want 130", code)
1135	}
1136	if got := g.cancelsMade(); got != 1 {
1137	t.Fatalf("%d cancels, want 1", got)
1138	}
1139	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
1140	t.Fatalf("stderr %q: a cut-off cancel is not an error to report", got)
1141	}
1142	})
1143	t.Run("both during the ack end the ack, not the exit code, and cancel nothing", func(t *testing.T) {
1144	f := newTermFx()
1145	g := newGatedRun(&scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{LastSeq: 3}}}})
1146	g.holdAck = true
1147	codes := start(g, f)
1148	awaitSignal(t, g.entered, "the ack request was never made")
1149	f.sig <- os.Interrupt
1150	f.sig <- os.Interrupt
1151	if code := quick(t, codes); code != 0 {
1152	t.Fatalf("exit %d, want the command's own 0", code)
1153	}
1154	if got := g.cancelsMade(); got != 0 {
1155	t.Fatalf("%d cancels sent to a finished run", got)
1156	}
1157	if got := f.err.String(); got != "" {
1158	t.Fatalf("stderr %q", got)
1159	}
1160	})
1161	t.Run("a second Ctrl-C during the cancel resend ends the wait before another request", func(t *testing.T) {
1162	f := newTermFx()
1163	f.sig <- os.Interrupt
1164	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}, cancelErrs: []error{errors.New("connection reset")}})
1165	g.holdCancel = 2 // the first cancel fails fast; its resend never answers
1166	g.holdPage = 3 // any page request after that would hang too
1167	codes := start(g, f)
1168	awaitSignal(t, g.cancelIn, "the cancel was never sent again")
1169	f.sig <- os.Interrupt
1170	if code := quick(t, codes); code != 130 {
1171	t.Fatalf("exit %d, want 130", code)
1172	}
1173	if got := g.pagesRead(); got != 2 {
1174	t.Fatalf("%d page requests, want 2: none after the second Ctrl-C", got)
1175	}
1176	})
1177	late := func(t *testing.T) {
1178	old := runLateCancelWait
1179	runLateCancelWait = 100 * time.Millisecond
1180	t.Cleanup(func() { runLateCancelWait = old })
1181	}
1182	t.Run("a cancel first sent after the second Ctrl-C is attempted, but only briefly", func(t *testing.T) {
1183	late(t)
1184	f := newTermFx()
1185	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
1186	g.holdPage = 1
1187	g.holdCancel = 1 // never answers
1188	codes := start(g, f)
1189	awaitSignal(t, g.entered, "the page request was never made")
1190	f.sig <- os.Interrupt
1191	f.sig <- os.Interrupt
1192	select {
1193	case code := <-codes:
1194	if code != 130 {
1195	t.Fatalf("exit %d, want 130", code)
1196	}
1197	case <-time.After(time.Second):
1198	t.Fatal("a cancel sent after the second Ctrl-C held the exit")
1199	}
1200	if got := g.cancelsMade(); got != 1 {
1201	t.Fatalf("%d cancels, want 1: still attempted", got)
1202	}
1203	if g.cancelCtx != nil {
1204	t.Fatalf("the cancel was sent on a context already %v: it was not attempted", g.cancelCtx)
1205	}
1206	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
1207	t.Fatalf("stderr %q", got)
1208	}
1209	})
1210	t.Run("a cancel resent on the second Ctrl-C is attempted, but only briefly", func(t *testing.T) {
1211	late(t)
1212	f := newTermFx()
1213	f.sig <- os.Interrupt
1214	g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}, cancelErrs: []error{errors.New("connection reset")}})
1215	g.holdPage = 2 // the second Ctrl-C arrives during this page request
1216	g.holdCancel = 2 // the resend never answers
1217	codes := start(g, f)
1218	awaitSignal(t, g.entered, "the second page request was never made")
1219	f.sig <- os.Interrupt
1220	select {
1221	case code := <-codes:
1222	if code != 130 {
1223	t.Fatalf("exit %d, want 130", code)
1224	}
1225	case <-time.After(time.Second):
1226	t.Fatal("a cancel resent after the second Ctrl-C held the exit")
1227	}
1228	if got := g.cancelsMade(); got != 2 {
1229	t.Fatalf("%d cancels, want 2: the resend is still attempted", got)
1230	}
1231	if g.cancelCtx != nil {
1232	t.Fatalf("the resend was sent on a context already %v: it was not attempted", g.cancelCtx)
1233	}
1234	})
1235	}
1236
1237	func TestStartRunPrintsTheFollowCommand(t *testing.T) {
1238	f := newTermFx()
1239	s := &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}}
1240	o := runOpts{instance: "shop.example.com", argv: []string{"import"}, timeout: time.Minute}
1241	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 {
1242	t.Fatalf("exit %d", code)
1243	}
1244	if !strings.Contains(f.err.String(), "run run_a on shop.example.com. to follow it again: carlos run --app shop --instance shop.example.com --follow run_a") {
1245	t.Fatalf("stderr %q", f.err.String())
1246	}
1247	if s.created.TimeoutS != 60 {
1248	t.Fatalf("meta %+v", s.created)
1249	}
1250	}
1251
1252	func TestStartRunFailures(t *testing.T) {
1253	missing := filepath.Join(t.TempDir(), "absent.csv")
1254	for name, c := range map[string]struct {
1255	o runOpts
1256	createErr error
1257	code int
1258	says string
1259	}{
1260	"a file that cannot be read": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute, files: []string{missing}}, nil, 2, "could not read " + missing},
1261	"a console refusal": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute}, &runConsoleError{status: http.StatusConflict, msg: "an instance can run one command at a time"}, 2, "carlos run: an instance can run one command at a time"},
1262	"a lost connection": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute}, errors.New("connection refused"), 1, "carlos run: connection refused"},
1263	} {
1264	f := newTermFx()
1265	s := &scriptedRun{createErr: c.createErr}
1266	if code := startRun(context.Background(), s, f.term(), c.o, "shop"); code != c.code \|\| !strings.Contains(f.err.String(), c.says) \|\| strings.Contains(f.err.String(), "to follow it again") {
1267	t.Errorf("%s: exit %d stderr %q", name, code, f.err.String())
1268	}
1269	}
1270	}
1271
1272	// A keyed run sends the key as the Idempotency-Key header, and declares each
1273	// file's sha256 in the meta so the console can tell a retry of the same
1274	// request from a different one under the same key.
1275	func TestRunIdempotencyKeyFlag(t *testing.T) {
1276	o, err := parseRunArgs([]string{"--instance", "h.example.com", "--idempotency-key", "import-42:a.b_c", "--", "import"}, io.Discard)
1277	if err != nil \|\| o.idempotencyKey != "import-42:a.b_c" {
1278	t.Fatalf("opts %+v, %v", o, err)
1279	}
1280	dir := t.TempDir()
1281	p := filepath.Join(dir, "x.csv")
1282	os.WriteFile(p, []byte("1,2"), 0o600)
1283	sum := sha256.Sum256([]byte("1,2"))
1284	want := hex.EncodeToString(sum[:])
1285
1286	m, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: time.Minute, files: []string{p}, idempotencyKey: "k1"})
1287	if err != nil \|\| len(m.Files) != 1 \|\| m.Files[0].SHA256 != want {
1288	t.Fatalf("keyed meta %+v, %v, want sha256 %s", m, err, want)
1289	}
1290	m, err = runMetaFor(runOpts{argv: []string{"import"}, timeout: time.Minute, files: []string{p}})
1291	if err != nil \|\| m.Files[0].SHA256 != "" {
1292	t.Fatalf("an unkeyed run declares no hash: %+v, %v", m, err)
1293	}
1294
1295	var header, metaJSON string
1296	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
1297	header = r.Header.Get("Idempotency-Key")
1298	mr, err := r.MultipartReader()
1299	if err != nil {
1300	t.Error(err)
1301	return
1302	}
1303	for {
1304	part, err := mr.NextPart()
1305	if err != nil {
1306	break
1307	}
1308	raw, _ := io.ReadAll(part)
1309	if part.FormName() == "meta" {
1310	metaJSON = string(raw)
1311	}
1312	}
1313	w.WriteHeader(http.StatusAccepted)
1314	json.NewEncoder(w).Encode(runs.Created{ID: "run_a", Host: "h.example.com", State: "queued"})
1315	}))
1316	defer ts.Close()
1317	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com", key: "k1"}
1318	meta, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: time.Minute, files: []string{p}, idempotencyKey: "k1"})
1319	if err != nil {
1320	t.Fatal(err)
1321	}
1322	if _, err := c.create(context.Background(), meta, []string{p}); err != nil {
1323	t.Fatal(err)
1324	}
1325	if header != "k1" \|\| !strings.Contains(metaJSON, `"sha256":"`+want+`"`) {
1326	t.Fatalf("header %q, meta %s", header, metaJSON)
1327	}
1328	c.key = ""
1329	header = "unset"
1330	if _, err := c.create(context.Background(), meta, []string{p}); err != nil \|\| header != "" {
1331	t.Fatalf("an unkeyed create sent the header %q, %v", header, err)
1332	}
1333	}
1334
1335	// An invalid key is refused by the CLI, before any network call: runRun
1336	// parses its arguments before it resolves a console.
1337	func TestRunIdempotencyKeyIsCheckedBeforeAnyCall(t *testing.T) {
1338	for name, key := range map[string]string{
1339	"a space": "two words",
1340	"a slash": "a/b",
1341	"too long": strings.Repeat("k", 129),
1342	"non-ascii": "klé",
1343	} {
1344	_, err := parseRunArgs([]string{"--instance", "h.example.com", "--idempotency-key", key, "--", "x"}, io.Discard)
1345	if err == nil \|\| !strings.Contains(err.Error(), "--idempotency-key") {
1346	t.Errorf("%s: %v, want a refusal naming --idempotency-key", name, err)
1347	}
1348	}
1349	// An empty value is the same as not passing the flag.
1350	if o, err := parseRunArgs([]string{"--instance", "h.example.com", "--idempotency-key", "", "--", "x"}, io.Discard); err != nil \|\| o.idempotencyKey != "" {
1351	t.Errorf("empty key: %+v, %v", o, err)
1352	}
1353	// A key belongs to starting a run; --follow starts none.
1354	if _, err := parseRunArgs([]string{"--instance", "h.example.com", "--follow", "run_x", "--idempotency-key", "k"}, io.Discard); err == nil {
1355	t.Error("--follow with a key accepted")
1356	}
1357	}
1358
1359	// keyedConsole is a console that admits a keyed create once: the second
1360	// create under the same key answers 202 with the same run and replayed.
1361	type keyedConsole struct {
1362	mu sync.Mutex
1363	creates int // create requests made
1364	started int // runs started
1365	admitted map[string]string
1366	pages []string // run ids paged
1367	}
1368
1369	func (k keyedConsole) ServeHTTP(w http.ResponseWriter, r http.Request) {
1370	k.mu.Lock()
1371	defer k.mu.Unlock()
1372	if r.Method == http.MethodPost && strings.HasPrefix(r.Header.Get("Content-Type"), "multipart/") {
1373	k.creates++
1374	key := r.Header.Get("Idempotency-Key")
1375	if id, ok := k.admitted[key]; ok {
1376	w.WriteHeader(http.StatusAccepted)
1377	json.NewEncoder(w).Encode(runs.Created{ID: id, Host: "h.example.com", State: "done", Replayed: true})
1378	return
1379	}
1380	k.started++
1381	id := "run_" + strconv.Itoa(k.started)
1382	k.admitted[key] = id
1383	w.WriteHeader(http.StatusAccepted)
1384	json.NewEncoder(w).Encode(runs.Created{ID: id, Host: "h.example.com", State: "queued"})
1385	return
1386	}
1387	if r.Method == http.MethodGet {
1388	k.pages = append(k.pages, r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:])
1389	json.NewEncoder(w).Encode(runs.Page{State: "done", Result: &runs.Result{Exit: 0}})
1390	return
1391	}
1392	json.NewEncoder(w).Encode(runs.StateAnswer{State: "done"}) // ack
1393	}
1394
1395	func TestRunCLIRetryFollowsTheReplayedRun(t *testing.T) {
1396	k := &keyedConsole{admitted: map[string]string{}}
1397	ts := httptest.NewServer(k)
1398	defer ts.Close()
1399	o := runOpts{instance: "h.example.com", argv: []string{"import"}, timeout: time.Minute, idempotencyKey: "import-42"}
1400	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com", key: o.idempotencyKey}
1401
1402	first := newTermFx()
1403	if code := startRun(context.Background(), c, first.term(), o, "shop"); code != 0 {
1404	t.Fatalf("first: exit %d, stderr %q", code, first.err.String())
1405	}
1406	if strings.Contains(first.err.String(), "already started") \|\| !strings.Contains(first.err.String(), "run run_1 on h.example.com. to follow it again") {
1407	t.Fatalf("a fresh run says %q", first.err.String())
1408	}
1409
1410	second := newTermFx()
1411	if code := startRun(context.Background(), c, second.term(), o, "shop"); code != 0 {
1412	t.Fatalf("retry: exit %d, stderr %q", code, second.err.String())
1413	}
1414	if !strings.Contains(second.err.String(), "already started: run run_1 on h.example.com. to follow it again: carlos run --app shop --instance h.example.com --follow run_1") {
1415	t.Fatalf("the retry says %q", second.err.String())
1416	}
1417	if k.creates != 2 \|\| k.started != 1 {
1418	t.Fatalf("%d creates, %d runs started: want 2 and 1", k.creates, k.started)
1419	}
1420	if want := []string{"run_1", "run_1"}; !slices.Equal(k.pages, want) {
1421	t.Fatalf("followed %v, want %v: the retry follows the same run", k.pages, want)
1422	}
1423	}
1424
1425	// replayedRun is a console whose create always answers a replay of run_a
1426	// with the given state.
1427	type replayedRun struct {
1428	*scriptedRun
1429	state string
1430	}
1431
1432	func (r replayedRun) create(context.Context, runs.Meta, []string) (runs.Created, error) {
1433	return runs.Created{ID: "run_a", Host: "shop.example.com", State: r.state, Replayed: true}, nil
1434	}
1435
1436	// A retry whose key started a run that was since acknowledged or swept has
1437	// nothing to follow: it says so, rather than the generic "has no run".
1438	func TestRunCLIRetryOfARunWhoseOutputIsGone(t *testing.T) {
1439	o := runOpts{instance: "shop.example.com", argv: []string{"import"}, timeout: time.Minute, idempotencyKey: "k"}
1440	f := newTermFx()
1441	s := replayedRun{scriptedRun: &scriptedRun{pageErr: &runConsoleError{status: http.StatusNotFound, msg: "no such run"}}}
1442	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 2 {
1443	t.Fatalf("exit %d, stderr %q", code, f.err.String())
1444	}
1445	got := f.err.String()
1446	if !strings.Contains(got, "run run_a was already started with this key, and its output is no longer kept") \|\|
1447	strings.Contains(got, "has no run") \|\| strings.Contains(got, "to follow it again") {
1448	t.Fatalf("stderr %q", got)
1449	}
1450
1451	// A replay the record still names is followed, with no probe.
1452	f = newTermFx()
1453	s = replayedRun{scriptedRun: &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}}, state: "done"}
1454	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 \|\| s.pagesRead() != 1 {
1455	t.Fatalf("named replay: exit %d, %d pages, stderr %q", code, s.pagesRead(), f.err.String())
1456	}
1457	// A replay the record no longer names, whose output is still there, is
1458	// followed too.
1459	f = newTermFx()
1460	s = replayedRun{scriptedRun: &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}}}
1461	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 \|\| !strings.Contains(f.err.String(), "already started: run run_a") {
1462	t.Fatalf("unnamed replay: exit %d, stderr %q", code, f.err.String())
1463	}
1464	}
1465