amadan

carlos / platform Public

cli: carlos run reports a local file error bare and typed; race-free file-change test (review fix 1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 3313a530c6bd538aceab1a30a14abdc38a2ce54a parent 84de64b
2 files changed, +179 −28
  • run.go +46 −3
  • run_test.go +133 −25

CI failed — run details

diff --git a/run.go b/run.go
index af9f06b43..ccbe3515a 100644
--- a/run.go
+++ b/run.go
@@ -14,6 +14,7 @@ import (
"path/filepath"
"slices"
"strings"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/runs"
@@ -86,10 +87,22 @@ 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")
+// errRunLocalFile is what errors.Is finds in a --file that cannot be
+// uploaded: a failure on this machine (exit 2), not the console's answer.
+var errRunLocalFile = errors.New("a --file could not be read")
+
+// runFileError is a --file failure carrying the RUN-C53 text.
+type runFileError struct{ err error }
+
+func (e *runFileError) Error() string { return e.err.Error() }
+func (e *runFileError) Unwrap() error { return e.err }
+func (e *runFileError) Is(target error) bool { return target == errRunLocalFile }
+
// runReadError is a --file that cannot be uploaded (RUN-C53). It is a local
-// failure: nothing has been sent.
+// failure, and either nothing has been sent or the upload is aborted before
+// its last byte.
func runReadError(path string, err error) error {
- return fmt.Errorf("could not read %s: %w", path, err)
+ return &runFileError{fmt.Errorf("could not read %s: %w", path, err)}
}
// runMetaFor is the create request's meta: argv, the timeout, and each
@@ -167,7 +180,14 @@ func (c runAPI) create(ctx context.Context, meta runs.Meta, paths []string) (run
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 {
+ var mu sync.Mutex
+ var writeErr error
+ status, body, err := c.s.c.doMultipart(ctx, c.path(""), func(mw *multipart.Writer) (err error) {
+ defer func() {
+ mu.Lock()
+ writeErr = err
+ mu.Unlock()
+ }()
mp, err := mw.CreateFormField("meta")
if err != nil {
return err
@@ -182,10 +202,28 @@ func (c runAPI) create(ctx context.Context, meta runs.Meta, paths []string) (run
}
return nil
})
+ if err != nil {
+ // The http client wraps a failed body in a *url.Error. When the
+ // failure was a file the writer could not read, the user is told that,
+ // not which URL the request was for.
+ mu.Lock()
+ we := writeErr
+ mu.Unlock()
+ if errors.Is(we, errRunLocalFile) {
+ return runs.Created{}, we
+ }
+ }
var out runs.Created
return out, runAnswer(status, body, err, http.StatusAccepted, &out)
}
+// runFileHandle is the open file writeRunFilePart reads: the reader and the
+// file's current size.
+type runFileHandle interface {
+ io.Reader
+ Stat() (os.FileInfo, error)
+}
+
// 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
@@ -196,6 +234,11 @@ func writeRunFilePart(mw *multipart.Writer, path string, want runs.MetaFile) err
return runReadError(path, err)
}
defer f.Close()
+ return copyRunFilePart(mw, f, path, want)
+}
+
+// copyRunFilePart is writeRunFilePart over an open file.
+func copyRunFilePart(mw *multipart.Writer, f runFileHandle, path string, want runs.MetaFile) error {
st, err := f.Stat()
if err != nil {
return runReadError(path, err)
diff --git a/run_test.go b/run_test.go
index 7b40cdfb2..b6b4479a1 100644
--- a/run_test.go
+++ b/run_test.go
@@ -1,15 +1,19 @@
package main
import (
+ "bytes"
"context"
"encoding/json"
"errors"
"io"
+ "mime/multipart"
"net/http"
"net/http/httptest"
+ "net/url"
"os"
"path/filepath"
"strings"
+ "sync/atomic"
"testing"
"time"
@@ -17,11 +21,11 @@ import (
)
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)
+ ok, err := parseRunArgs([]string{"--app", "shop", "--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" {
+ 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" {
t.Fatalf("opts %+v", ok)
}
def, err := parseRunArgs([]string{"--instance", "h.example.com", "--", "x"}, io.Discard)
@@ -29,12 +33,12 @@ func TestParseRunArgs(t *testing.T) {
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 --",
+ "--app shop --instance h.example.com import --dry-run": "",
+ "--app shop -- 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 {
@@ -61,7 +65,7 @@ func TestRunMetaForStatsEveryFile(t *testing.T) {
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+": ") {
+ if _, err := runMetaFor(runOpts{files: []string{missing}}); !errors.Is(err, errRunLocalFile) || !strings.HasPrefix(err.Error(), "could not read "+missing+": ") {
t.Fatalf("missing file: %v", err)
}
}
@@ -111,7 +115,7 @@ func TestRunAPIWireShapes(t *testing.T) {
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"}
+ c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", 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" {
@@ -134,7 +138,7 @@ func TestRunAPIWireShapes(t *testing.T) {
if err := c.ack(ctx, "run_a", 7); err != nil {
t.Fatal(err)
}
- base := "/api/cli/apps/tito/titogo/instances/h.example.com/runs"
+ base := "/api/cli/apps/acme/shop/instances/h.example.com/runs"
want := []seen{
{http.MethodPost, base, "Bearer cli-token", ""},
{http.MethodGet, base + "/run_a?after=3", "Bearer cli-token", ""},
@@ -155,7 +159,7 @@ func TestRunAPIWireShapes(t *testing.T) {
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+": ") {
+ if !errors.Is(err, errRunLocalFile) || !errors.Is(err, errRunNotRegular) || !strings.HasPrefix(err.Error(), "could not read "+dir+": ") {
t.Fatalf("directory: %v", err)
}
}
@@ -171,50 +175,154 @@ func TestRunAPICreateFailsWhenAFileChangesSize(t *testing.T) {
"grown": {"12345", 3},
} {
t.Run(name, func(t *testing.T) {
- var complete bool
+ var complete atomic.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
+ return
}
if _, err := io.Copy(io.Discard, part); err != nil {
- ok = false
- break
+ return
}
}
- complete = ok
+ complete.Store(true)
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"}
+ c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", 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)
+ // Close waits for the handler, so complete is settled when read.
+ ts.Close()
+ var uerr *url.Error
+ if !errors.Is(err, errRunLocalFile) || errors.As(err, &uerr) || !strings.HasPrefix(err.Error(), "could not read "+p+": ") {
+ t.Fatalf("create: %v, want the bare could-not-read error", err)
}
- if complete {
+ if complete.Load() {
t.Fatal("the console received a whole request whose part disagrees with meta")
}
})
}
}
+// fakeRunFile is an open file whose reader and reported sizes the test
+// controls.
+type fakeRunFile struct {
+ io.Reader
+ mode os.FileMode
+ sizes []int64 // what successive Stat calls report; the last repeats
+}
+
+type fakeInfo struct {
+ os.FileInfo
+ mode os.FileMode
+ size int64
+}
+
+func (i fakeInfo) Mode() os.FileMode { return i.mode }
+func (i fakeInfo) Size() int64 { return i.size }
+
+func (f *fakeRunFile) Stat() (os.FileInfo, error) {
+ size := f.sizes[0]
+ if len(f.sizes) > 1 {
+ f.sizes = f.sizes[1:]
+ }
+ return fakeInfo{mode: f.mode, size: size}, nil
+}
+
+func TestCopyRunFilePartBranches(t *testing.T) {
+ want := runs.MetaFile{Name: "x.csv", Size: 4}
+ for name, tc := range map[string]struct {
+ f *fakeRunFile
+ fail error // nil: the part is written
+ }{
+ "as promised": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{4, 4}}, nil},
+ "stat disagrees": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{5}}, errRunFileChanged},
+ "reader short": {&fakeRunFile{strings.NewReader("12"), 0, []int64{4, 4}}, errRunFileChanged},
+ "grown after the copy": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{4, 9}}, errRunFileChanged},
+ "not a regular file": {&fakeRunFile{strings.NewReader("1234"), os.ModeDir, []int64{4}}, errRunNotRegular},
+ "reader fails midway": {&fakeRunFile{io.MultiReader(strings.NewReader("12"), iotestErrReader{}), 0, []int64{4, 4}}, errRunReaderBroke},
+ } {
+ t.Run(name, func(t *testing.T) {
+ var buf bytes.Buffer
+ mw := multipart.NewWriter(&buf)
+ err := copyRunFilePart(mw, tc.f, "/data/x.csv", want)
+ if tc.fail == nil {
+ if err != nil || !strings.Contains(buf.String(), "1234") {
+ t.Fatalf("err %v, body %q", err, buf.String())
+ }
+ return
+ }
+ if !errors.Is(err, tc.fail) || !errors.Is(err, errRunLocalFile) || !strings.HasPrefix(err.Error(), "could not read /data/x.csv: ") {
+ t.Fatalf("err %v, want %v as a local file error", err, tc.fail)
+ }
+ })
+ }
+}
+
+var errRunReaderBroke = errors.New("disk fell over")
+
+type iotestErrReader struct{}
+
+func (iotestErrReader) Read([]byte) (int, error) { return 0, errRunReaderBroke }
+
+// The wire order the console's door requires: meta first, then one file
+// part per meta file in meta's order, each named as meta names it.
+func TestRunAPICreateSendsMetaThenFilesInOrder(t *testing.T) {
+ type part struct{ form, name, body string }
+ var got []part
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ mr, err := r.MultipartReader()
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ for {
+ p, err := mr.NextPart()
+ if err != nil {
+ break
+ }
+ raw, _ := io.ReadAll(p)
+ got = append(got, part{p.FormName(), p.FileName(), string(raw)})
+ }
+ w.WriteHeader(http.StatusAccepted)
+ json.NewEncoder(w).Encode(runs.Created{ID: "run_a"})
+ }))
+ defer ts.Close()
+ dir := t.TempDir()
+ var paths []string
+ var files []runs.MetaFile
+ for _, f := range []struct{ name, body string }{{"b.csv", "bb"}, {"a.csv", "a"}} {
+ p := filepath.Join(dir, f.name)
+ if err := os.WriteFile(p, []byte(f.body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ paths = append(paths, p)
+ files = append(files, runs.MetaFile{Name: f.name, Size: int64(len(f.body))})
+ }
+ c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
+ if _, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60, Files: files}, paths); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 3 || got[0].form != "meta" || got[0].name != "" ||
+ got[1] != (part{"file", "b.csv", "bb"}) || got[2] != (part{"file", "a.csv", "a"}) {
+ t.Fatalf("parts %+v", got)
+ }
+}
+
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"}
+ c := runAPI{s: newTestSession("http://127.0.0.1:1"), a: apiAcct{Seg: "acme"}, 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")