amadan

carlos / platform Public

cli: carlos run's flags and the five console routes (run spec §3, §5.1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 84de64b4793a9e9379ad0dc31da036d3ad725dc0 parent 809329f
4 files changed, +488 −1
  • approved_run_copy_test.go +1 −1
  • main.go +3 −0
  • run.go +262 −0
  • run_test.go +222 −0
diff --git a/approved_run_copy_test.go b/approved_run_copy_test.go
index 6cdae40a3..ead810535 100644
--- a/approved_run_copy_test.go
+++ b/approved_run_copy_test.go
@@ -33,7 +33,7 @@ var runCopyIDs = func() []string {
// runCopyBuilt is every deck id whose text the code or docs now carry. The
// task that writes a row's text adds its id here in the same commit.
-var runCopyBuilt = []string{"RUN-C1", "RUN-C2", "RUN-C3", "RUN-C4", "RUN-C5", "RUN-C6", "RUN-C7", "RUN-C8", "RUN-C9", "RUN-C10", "RUN-C11", "RUN-C12", "RUN-C13", "RUN-C14", "RUN-C15", "RUN-C16", "RUN-C17", "RUN-C18", "RUN-C19", "RUN-C20", "RUN-C21", "RUN-C23", "RUN-C24", "RUN-C25", "RUN-C26", "RUN-C27", "RUN-C28", "RUN-C29", "RUN-C30", "RUN-C31", "RUN-C32", "RUN-C33", "RUN-C34", "RUN-C35", "RUN-C36", "RUN-C37", "RUN-C38", "RUN-C39", "RUN-C40", "RUN-C41", "RUN-C42", "RUN-C43", "RUN-C44", "RUN-C45", "RUN-C46", "RUN-C82"}
+var runCopyBuilt = []string{"RUN-C1", "RUN-C2", "RUN-C3", "RUN-C4", "RUN-C5", "RUN-C6", "RUN-C7", "RUN-C8", "RUN-C9", "RUN-C10", "RUN-C11", "RUN-C12", "RUN-C13", "RUN-C14", "RUN-C15", "RUN-C16", "RUN-C17", "RUN-C18", "RUN-C19", "RUN-C20", "RUN-C21", "RUN-C23", "RUN-C24", "RUN-C25", "RUN-C26", "RUN-C27", "RUN-C28", "RUN-C29", "RUN-C30", "RUN-C31", "RUN-C32", "RUN-C33", "RUN-C34", "RUN-C35", "RUN-C36", "RUN-C37", "RUN-C38", "RUN-C39", "RUN-C40", "RUN-C41", "RUN-C42", "RUN-C43", "RUN-C44", "RUN-C45", "RUN-C46", "RUN-C47", "RUN-C48", "RUN-C49", "RUN-C50", "RUN-C51", "RUN-C52", "RUN-C53", "RUN-C82", "RUN-C83"}
func mustParseRunRecord(t *testing.T) []copyRow {
t.Helper()
diff --git a/main.go b/main.go
index 15078b40f..500ee001f 100644
--- a/main.go
+++ b/main.go
@@ -73,6 +73,8 @@ func main() {
runRestart(args)
case "db":
runDB(args)
+ case "run":
+ runRun(args)
case "steering":
runSteering(args)
case "schedule":
@@ -203,6 +205,7 @@ Your apps — ship and run software on a CARLOS deployment
--host or --environment narrows it, --dry-run names the set
db read a copy of an instance's database, restored from its
replica: pull (save it) | query (one read-only query)
+ run run one of your app's commands against one instance, as that instance
schedule recurring or one-shot work per app: ls | set | rm | run
store object storage: create | status | apply | approvals | approve | decline | grant | rotate
relay the platform STUN/TURN relay: serve (what the unit runs)
diff --git a/run.go b/run.go
new file mode 100644
index 000000000..af9f06b43
--- /dev/null
+++ b/run.go
@@ -0,0 +1,262 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "io"
+ "mime/multipart"
+ "net/http"
+ "net/url"
+ "os"
+ "path/filepath"
+ "slices"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+const runUsageRun = "usage: carlos run --app <app> --instance <host> [--file <path>]... [--timeout 30m] -- <arg>..."
+
+const runUsageFollow = "carlos run --app <app> --instance <host> --follow <id>"
+
+func runUsageText() string { return runUsageRun + "\n " + runUsageFollow + "\n" }
+
+const runDirectModeMsg = "carlos run works through a console, and this shell is in direct bucket mode. unset CARLOS_DEPLOYMENT_BUCKET and CARLOS_DEPLOYMENT_DIR, then run carlos auth login --console <url>"
+
+// errRunUsage is a command line that does not parse: the usage text says
+// what was wrong.
+var errRunUsage = errors.New("carlos run: usage")
+
+// runOpts is `carlos run` parsed.
+type runOpts struct {
+ app, instance, follow, console string
+ acctRaw, acctIDRaw *string
+ files []string
+ timeout time.Duration
+ argv []string
+}
+
+// parseRunArgs splits at the first "--": carlos's flags before it, the
+// command after it. Nothing may sit between the flags and the "--", so a
+// forgotten "--" is a usage error rather than a command whose flags carlos
+// swallowed.
+func parseRunArgs(args []string, stderr io.Writer) (runOpts, error) {
+ flagArgs, argv := args, []string(nil)
+ if i := slices.Index(args, "--"); i >= 0 {
+ flagArgs, argv = args[:i], args[i+1:]
+ }
+ fs := flag.NewFlagSet("run", flag.ContinueOnError)
+ fs.SetOutput(stderr)
+ app := appFlag(fs, "the app whose command to run")
+ instance := fs.String("instance", "", "the instance to run it against")
+ var files multiFlag
+ fs.Var(&files, "file", "a file to upload with the run (repeatable; 64 MB in total)")
+ timeout := fs.Duration("timeout", runs.DefaultTimeout, "how long the command may run (1s to 6h)")
+ follow := fs.String("follow", "", "attach to a run already started, by its id")
+ acctRaw, acctIDRaw, _ := accountFlags(fs)
+ console := consoleOption(fs, "console to run through (default: the only logged-in one)")
+ fs.Usage = flagUsage(fs, runUsageText())
+ if err := fs.Parse(flagArgs); err != nil {
+ return runOpts{}, err
+ }
+ if fs.NArg() > 0 {
+ return runOpts{}, errRunUsage
+ }
+ o := runOpts{app: strings.TrimSpace(*app), instance: strings.TrimSpace(*instance), follow: strings.TrimSpace(*follow),
+ console: *console, acctRaw: acctRaw, acctIDRaw: acctIDRaw, files: files, timeout: *timeout, argv: argv}
+ switch {
+ case o.instance == "":
+ return o, errors.New("--app and --instance name the instance to run against")
+ case o.follow != "" && (len(o.argv) > 0 || len(o.files) > 0):
+ return o, errors.New("--follow takes a run id and no command after --")
+ case runs.CheckTimeout(o.timeout) != nil:
+ return o, errors.New("--timeout must be between 1s and 6h")
+ }
+ return o, nil
+}
+
+// errRunNotRegular is a --file that is a directory or a device: only a
+// regular file has a size to promise the console.
+var errRunNotRegular = errors.New("not a regular file")
+
+// errRunFileChanged is a --file whose size is no longer what meta promised.
+var errRunFileChanged = errors.New("the file changed while carlos was reading it")
+
+// runReadError is a --file that cannot be uploaded (RUN-C53). It is a local
+// failure: nothing has been sent.
+func runReadError(path string, err error) error {
+ return fmt.Errorf("could not read %s: %w", path, err)
+}
+
+// runMetaFor is the create request's meta: argv, the timeout, and each
+// file's base name and size. The console checks all of it again.
+func runMetaFor(o runOpts) (runs.Meta, error) {
+ m := runs.Meta{Argv: o.argv, TimeoutS: int64(o.timeout / time.Second)}
+ for _, p := range o.files {
+ st, err := os.Stat(p)
+ if err != nil {
+ return m, runReadError(p, err)
+ }
+ if !st.Mode().IsRegular() {
+ return m, runReadError(p, errRunNotRegular)
+ }
+ m.Files = append(m.Files, runs.MetaFile{Name: filepath.Base(p), Size: st.Size()})
+ }
+ return m, nil
+}
+
+// runClient is the console's five run routes (run spec §5.1, §5.6, §5.7).
+type runClient interface {
+ create(ctx context.Context, meta runs.Meta, paths []string) (runs.Created, error)
+ page(ctx context.Context, id string, after int64) (runs.Page, error)
+ cancel(ctx context.Context, id string) (string, error)
+ expire(ctx context.Context, id string) (string, error)
+ ack(ctx context.Context, id string, lastSeq int64) error
+}
+
+// runConsoleError is a 4xx the console answered, with its message: a
+// refusal by the console, exit 2.
+type runConsoleError struct {
+ status int
+ msg string
+}
+
+func (e *runConsoleError) Error() string { return e.msg }
+
+// runAnswer decodes a want-status answer into out, or turns any other
+// answer into an error: a 4xx into a runConsoleError.
+func runAnswer(status int, body []byte, err error, want int, out any) error {
+ if err != nil {
+ return err
+ }
+ if status != want {
+ e := apiErrorFromBody(status, body)
+ if status >= 400 && status < 500 {
+ return &runConsoleError{status: status, msg: e.Error()}
+ }
+ return e
+ }
+ if out == nil {
+ return nil
+ }
+ return json.Unmarshal(body, out)
+}
+
+// runAPI is runClient over the signed-in console.
+type runAPI struct {
+ s *apiSession
+ a apiAcct
+ app string
+ host string
+}
+
+func (c runAPI) path(tail string) string {
+ return apiPath(c.a, c.app, "/instances/"+url.PathEscape(c.host)+"/runs"+tail)
+}
+
+// create sends meta, then one file part per meta file in meta's order, each
+// named as meta names it and exactly as long as meta says. A file whose size
+// is no longer meta's fails the call before its part is written, or aborts
+// the upload part-way: the console never gets a whole request whose parts
+// disagree with its meta.
+func (c runAPI) create(ctx context.Context, meta runs.Meta, paths []string) (runs.Created, error) {
+ if len(paths) != len(meta.Files) {
+ return runs.Created{}, errors.New("carlos run: the files to upload and the files in meta differ")
+ }
+ status, body, err := c.s.c.doMultipart(ctx, c.path(""), func(mw *multipart.Writer) error {
+ mp, err := mw.CreateFormField("meta")
+ if err != nil {
+ return err
+ }
+ if err := json.NewEncoder(mp).Encode(meta); err != nil {
+ return err
+ }
+ for i, p := range paths {
+ if err := writeRunFilePart(mw, p, meta.Files[i]); err != nil {
+ return err
+ }
+ }
+ return nil
+ })
+ var out runs.Created
+ return out, runAnswer(status, body, err, http.StatusAccepted, &out)
+}
+
+// writeRunFilePart writes one file part: exactly want.Size bytes of the file
+// at path, under want.Name. The size is checked on the open file, before and
+// after the copy, so a file that grew or shrank since the stat is an error
+// and not a part that disagrees with meta.
+func writeRunFilePart(mw *multipart.Writer, path string, want runs.MetaFile) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return runReadError(path, err)
+ }
+ defer f.Close()
+ st, err := f.Stat()
+ if err != nil {
+ return runReadError(path, err)
+ }
+ if !st.Mode().IsRegular() {
+ return runReadError(path, errRunNotRegular)
+ }
+ if st.Size() != want.Size {
+ return runReadError(path, errRunFileChanged)
+ }
+ fp, err := mw.CreateFormFile("file", want.Name)
+ if err != nil {
+ return err
+ }
+ if n, err := io.CopyN(fp, f, want.Size); err != nil {
+ if n < want.Size && errors.Is(err, io.EOF) {
+ err = errRunFileChanged
+ }
+ return runReadError(path, err)
+ }
+ if st, err := f.Stat(); err != nil || st.Size() != want.Size {
+ return runReadError(path, errRunFileChanged)
+ }
+ return nil
+}
+
+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)))
+ var out runs.Page
+ return out, runAnswer(status, body, err, http.StatusOK, &out)
+}
+
+func (c runAPI) cancel(ctx context.Context, id string) (string, error) {
+ status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/cancel"), struct{}{})
+ var out runs.StateAnswer
+ return out.State, runAnswer(status, body, err, http.StatusOK, &out)
+}
+
+func (c runAPI) expire(ctx context.Context, id string) (string, error) {
+ status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/expire"), struct{}{})
+ var out runs.StateAnswer
+ return out.State, runAnswer(status, body, err, http.StatusOK, &out)
+}
+
+func (c runAPI) ack(ctx context.Context, id string, lastSeq int64) error {
+ status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/ack"), runs.Ack{LastSeq: lastSeq})
+ return runAnswer(status, body, err, http.StatusOK, nil)
+}
+
+// runRun is `carlos run` (run spec §3).
+func runRun(args []string) {
+ o, err := parseRunArgs(args, os.Stderr)
+ if err != nil {
+ if errors.Is(err, flag.ErrHelp) {
+ os.Exit(0)
+ }
+ if !errors.Is(err, errRunUsage) {
+ fmt.Fprintln(os.Stderr, err.Error())
+ }
+ fmt.Fprint(os.Stderr, runUsageText())
+ os.Exit(2)
+ }
+ _ = o
+}
diff --git a/run_test.go b/run_test.go
new file mode 100644
index 000000000..7b40cdfb2
--- /dev/null
+++ b/run_test.go
@@ -0,0 +1,222 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+func TestParseRunArgs(t *testing.T) {
+ ok, err := parseRunArgs([]string{"--app", "titogo", "--instance", "h.example.com", "--file", "a.csv", "--file", "b.csv", "--timeout", "2m", "--", "import", "--dry-run"}, io.Discard)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ok.app != "titogo" || ok.instance != "h.example.com" || len(ok.files) != 2 || ok.timeout != 2*time.Minute || strings.Join(ok.argv, " ") != "import --dry-run" {
+ t.Fatalf("opts %+v", ok)
+ }
+ def, err := parseRunArgs([]string{"--instance", "h.example.com", "--", "x"}, io.Discard)
+ if err != nil || def.timeout != 30*time.Minute {
+ t.Fatalf("default timeout %s, %v", def.timeout, err)
+ }
+ for args, want := range map[string]string{
+ "--app titogo --instance h.example.com import --dry-run": "",
+ "--app titogo -- import": "--app and --instance name the instance to run against",
+ "--instance h.example.com --timeout 7h -- x": "--timeout must be between 1s and 6h",
+ "--instance h.example.com --timeout 0s -- x": "--timeout must be between 1s and 6h",
+ "--instance h.example.com --follow run_x -- x": "--follow takes a run id and no command after --",
+ "--instance h.example.com --follow run_x --file a": "--follow takes a run id and no command after --",
+ } {
+ _, err := parseRunArgs(strings.Fields(args), io.Discard)
+ if err == nil {
+ t.Errorf("%q accepted", args)
+ continue
+ }
+ if want == "" && !errors.Is(err, errRunUsage) {
+ t.Errorf("%q: %v, want the usage error", args, err)
+ }
+ if want != "" && err.Error() != want {
+ t.Errorf("%q: %q, want %q", args, err, want)
+ }
+ }
+}
+
+func TestRunMetaForStatsEveryFile(t *testing.T) {
+ dir := t.TempDir()
+ p := filepath.Join(dir, "checkins.csv")
+ if err := os.WriteFile(p, []byte("a,b\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ m, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: 90 * time.Second, files: []string{p}})
+ if err != nil || m.TimeoutS != 90 || len(m.Files) != 1 || m.Files[0].Name != "checkins.csv" || m.Files[0].Size != 4 {
+ t.Fatalf("meta %+v, %v", m, err)
+ }
+ missing := filepath.Join(dir, "nope.csv")
+ if _, err := runMetaFor(runOpts{files: []string{missing}}); err == nil || !strings.HasPrefix(err.Error(), "could not read "+missing+": ") {
+ t.Fatalf("missing file: %v", err)
+ }
+}
+
+func TestRunAPIWireShapes(t *testing.T) {
+ type seen struct{ method, path, auth, body string }
+ var got []seen
+ var metaJSON, fileName, fileBody string
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ s := seen{method: r.Method, path: r.URL.RequestURI(), auth: r.Header.Get("Authorization")}
+ if strings.HasPrefix(r.Header.Get("Content-Type"), "multipart/") {
+ mr, err := r.MultipartReader()
+ if err != nil {
+ t.Fatal(err)
+ }
+ for {
+ part, err := mr.NextPart()
+ if err != nil {
+ break
+ }
+ raw, _ := io.ReadAll(part)
+ if part.FormName() == "meta" {
+ metaJSON = string(raw)
+ } else {
+ fileName, fileBody = part.FileName(), string(raw)
+ }
+ }
+ got = append(got, s)
+ w.WriteHeader(http.StatusAccepted)
+ json.NewEncoder(w).Encode(runs.Created{ID: "run_a", Host: "h.example.com", State: "queued"})
+ return
+ }
+ raw, _ := io.ReadAll(r.Body)
+ s.body = string(raw)
+ got = append(got, s)
+ switch {
+ case strings.Contains(s.path, "/expire"):
+ w.WriteHeader(http.StatusConflict)
+ json.NewEncoder(w).Encode(map[string]string{"error": "run run_a has already started"})
+ case r.Method == http.MethodGet:
+ json.NewEncoder(w).Encode(runs.Page{ID: "run_a", State: "picked"})
+ default:
+ json.NewEncoder(w).Encode(runs.StateAnswer{State: "cancel"})
+ }
+ }))
+ defer ts.Close()
+ dir := t.TempDir()
+ p := filepath.Join(dir, "x.csv")
+ os.WriteFile(p, []byte("1,2"), 0o600)
+ c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "tito"}, app: "titogo", host: "h.example.com"}
+ ctx := context.Background()
+ created, err := c.create(ctx, runs.Meta{Argv: []string{"import"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "x.csv", Size: 3}}}, []string{p})
+ if err != nil || created.ID != "run_a" {
+ t.Fatalf("create %+v, %v", created, err)
+ }
+ if !strings.Contains(metaJSON, `"argv":["import"]`) || fileName != "x.csv" || fileBody != "1,2" {
+ t.Fatalf("meta %s file %s %q", metaJSON, fileName, fileBody)
+ }
+ if _, err := c.page(ctx, "run_a", 3); err != nil {
+ t.Fatal(err)
+ }
+ if st, err := c.cancel(ctx, "run_a"); err != nil || st != "cancel" {
+ t.Fatalf("cancel %q, %v", st, err)
+ }
+ _, err = c.expire(ctx, "run_a")
+ var ce *runConsoleError
+ if !errors.As(err, &ce) || ce.status != http.StatusConflict || ce.msg != "run run_a has already started" {
+ t.Fatalf("expire %v", err)
+ }
+ if err := c.ack(ctx, "run_a", 7); err != nil {
+ t.Fatal(err)
+ }
+ base := "/api/cli/apps/tito/titogo/instances/h.example.com/runs"
+ want := []seen{
+ {http.MethodPost, base, "Bearer cli-token", ""},
+ {http.MethodGet, base + "/run_a?after=3", "Bearer cli-token", ""},
+ {http.MethodPost, base + "/run_a/cancel", "Bearer cli-token", "{}"},
+ {http.MethodPost, base + "/run_a/expire", "Bearer cli-token", "{}"},
+ {http.MethodPost, base + "/run_a/ack", "Bearer cli-token", `{"last_seq":7}`},
+ }
+ if len(got) != len(want) {
+ t.Fatalf("requests %+v", got)
+ }
+ for i := range want {
+ if got[i] != want[i] {
+ t.Errorf("request %d = %+v, want %+v", i, got[i], want[i])
+ }
+ }
+}
+
+func TestRunMetaForRefusesWhatIsNotAFile(t *testing.T) {
+ dir := t.TempDir()
+ _, err := runMetaFor(runOpts{files: []string{dir}})
+ if err == nil || !strings.HasPrefix(err.Error(), "could not read "+dir+": ") {
+ t.Fatalf("directory: %v", err)
+ }
+}
+
+// A part that disagrees with meta must never reach the console as a whole
+// request: the file changed between the stat and the upload.
+func TestRunAPICreateFailsWhenAFileChangesSize(t *testing.T) {
+ for name, tc := range map[string]struct {
+ content string
+ size int64
+ }{
+ "shrunk": {"12", 3},
+ "grown": {"12345", 3},
+ } {
+ t.Run(name, func(t *testing.T) {
+ var complete bool
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ mr, err := r.MultipartReader()
+ if err != nil {
+ return
+ }
+ ok := true
+ for {
+ part, err := mr.NextPart()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ ok = false
+ break
+ }
+ if _, err := io.Copy(io.Discard, part); err != nil {
+ ok = false
+ break
+ }
+ }
+ complete = ok
+ w.WriteHeader(http.StatusAccepted)
+ json.NewEncoder(w).Encode(runs.Created{ID: "run_a"})
+ }))
+ defer ts.Close()
+ p := filepath.Join(t.TempDir(), "x.csv")
+ if err := os.WriteFile(p, []byte(tc.content), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "tito"}, app: "titogo", host: "h.example.com"}
+ _, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "x.csv", Size: tc.size}}}, []string{p})
+ if err == nil || !strings.Contains(err.Error(), "could not read "+p+": ") {
+ t.Fatalf("create: %v, want a could-not-read error", err)
+ }
+ if complete {
+ t.Fatal("the console received a whole request whose part disagrees with meta")
+ }
+ })
+ }
+}
+
+func TestRunAPICreateRefusesFilesThatDisagreeWithMeta(t *testing.T) {
+ c := runAPI{s: newTestSession("http://127.0.0.1:1"), a: apiAcct{Seg: "tito"}, app: "a", host: "h.example.com"}
+ _, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60}, []string{"/nowhere"})
+ if err == nil {
+ t.Fatal("create accepted more paths than meta files")
+ }
+}