amadan

carlos / platform Public

console: a run whose admission answer was lost keeps its objects; more door tests (review round 1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 76acbe7b1026bdfe1ea1c049b4d889a56b9f0948 parent f04a058
2 files changed, +194 −9
  • internal/console/apiruns.go +50 −3
  • internal/console/apiruns_test.go +144 −6

CI passed — run details

diff --git a/internal/console/apiruns.go b/internal/console/apiruns.go
index 98eb80505..25b24afb0 100644
--- a/internal/console/apiruns.go
+++ b/internal/console/apiruns.go
@@ -266,8 +266,30 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
}
// The pickup window opens at admission, not at the first byte: a slow
// upload must not eat the box's 60 seconds.
- got, err := provision.SetRecordRun(ctx, s.cfg.Releases, acct.Sqid, app, host,
- provision.RunAdmit(provision.RunAdmission{ID: id, By: actor, Deadline: time.Now().UTC().Add(runs.PickupWindow)}))
+ //
+ // The admission runs to its end even if the caller has gone: a conditional
+ // Put that is cancelled may or may not have landed, and deleting the
+ // objects of a run the record holds would leave a queued run no box can
+ // read. The outcome is then settled from the record, not from the error.
+ wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+ defer cancel()
+ state, err := s.admitRun(wctx, acct.Sqid, app, host, id, actor)
+ if err != nil && !runAdmitRefused(err) {
+ // The outcome is unknown (a lost response, a cancelled request, a
+ // transport error after the write): ask the record whose run it holds.
+ cur, rerr := provision.GetRecord(wctx, s.cfg.Releases, acct.Sqid, app, host)
+ switch {
+ case rerr != nil:
+ // Still unknown. Keep the objects: if the run was admitted a box
+ // needs them, and if not, the orphan sweep ages them out by
+ // request.json (Task 5).
+ log.Printf("console: api run %s/%s/%s: admit %s: %v; record unreadable: %v", acct.Sqid, app, host, id, err, rerr)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ case cur.ActiveRun != nil && cur.ActiveRun.ID == id:
+ state, err = cur.ActiveRun.State, nil
+ }
+ }
if err != nil {
_ = runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id)
var open *provision.RunOpenError
@@ -289,7 +311,32 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
s.bumpSerial(ctx)
s.auditActivity(acct, app, actor, Activity{Action: "run-start", TargetType: "instance", Target: host,
Detail: runStartDetail(req, s.runVersion(ctx, acct.Sqid, app, rec.Channel))})
- writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: got.ActiveRun.State})
+ writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: state})
+}
+
+// admitRun is the admission CAS. A RunOpenError naming id itself means an
+// earlier attempt of this same call committed (updateRecord retries after a
+// lost-response conflict, and finds its own run queued): that is admitted.
+func (s *Server) admitRun(ctx context.Context, sqid, app, host, id, by string) (string, error) {
+ got, err := provision.SetRecordRun(ctx, s.cfg.Releases, sqid, app, host,
+ provision.RunAdmit(provision.RunAdmission{ID: id, By: by, Deadline: time.Now().UTC().Add(runs.PickupWindow)}))
+ var open *provision.RunOpenError
+ switch {
+ case err == nil:
+ return got.ActiveRun.State, nil
+ case errors.As(err, &open) && open.Run.ID == id:
+ return open.Run.State, nil
+ }
+ return "", err
+}
+
+// runAdmitRefused reports whether err is the record saying no (another run,
+// a move, an uncommitted import, no record) rather than a failure whose
+// outcome the record has to settle.
+func runAdmitRefused(err error) bool {
+ var open *provision.RunOpenError
+ return errors.As(err, &open) || errors.Is(err, provision.ErrRunMoving) ||
+ errors.Is(err, provision.ErrRunImport) || errors.Is(err, os.ErrNotExist)
}
// writeRunObjects writes request.json first (the orphan sweep ages a run
diff --git a/internal/console/apiruns_test.go b/internal/console/apiruns_test.go
index d5025b6f1..d7c4e9394 100644
--- a/internal/console/apiruns_test.go
+++ b/internal/console/apiruns_test.go
@@ -4,9 +4,12 @@ import (
"bytes"
"context"
"encoding/json"
+ "errors"
+ "fmt"
"io"
"mime/multipart"
"net/http"
+ "os"
"strings"
"sync"
"testing"
@@ -39,6 +42,13 @@ func (fx *runFx) runsURL(host string) string {
// postRun sends the multipart create; parts lets a test send parts that
// disagree with meta.
func postRun(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ name, body string }) (int, string) {
+ t.Helper()
+ status, _, body := postRunH(t, u, bearer, meta, parts)
+ return status, body
+}
+
+// postRunH is postRun that also returns the response headers.
+func postRunH(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ name, body string }) (int, http.Header, string) {
t.Helper()
var buf bytes.Buffer
mw := multipart.NewWriter(&buf)
@@ -71,7 +81,7 @@ func postRun(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ nam
}
defer res.Body.Close()
out, _ := io.ReadAll(res.Body)
- return res.StatusCode, string(out)
+ return res.StatusCode, res.Header, string(out)
}
func okMeta() runs.Meta { return runs.Meta{Argv: []string{"import", "x.csv"}, TimeoutS: 60} }
@@ -85,14 +95,32 @@ func errorOf(t *testing.T, body string) string {
return e.Error
}
+// spoolDir points the door's spool files at a directory of the test's own
+// and returns it, so a test can see that none is left behind.
+func spoolDir(t *testing.T) string {
+ t.Helper()
+ dir := t.TempDir()
+ t.Setenv("TMPDIR", dir)
+ return dir
+}
+
+func assertNoSpool(t *testing.T, dir, when string) {
+ t.Helper()
+ if left, _ := os.ReadDir(dir); len(left) != 0 {
+ t.Errorf("%s left %d spool file(s) behind", when, len(left))
+ }
+}
+
func TestAPIRunCreateAdmitsAndRecords(t *testing.T) {
fx := runFixture(t)
+ spool := spoolDir(t)
meta := okMeta()
meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
if status != http.StatusAccepted {
t.Fatalf("status %d %s", status, body)
}
+ assertNoSpool(t, spool, "an admitted run")
var c runs.Created
if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) || c.State != provision.RunQueued {
t.Fatalf("created %+v, %v", c, err)
@@ -136,11 +164,8 @@ func TestAPIRunCreateRefusesInOrder(t *testing.T) {
if status, _ := postRun(t, fx.runsURL(runTestHost), svc, okMeta(), nil); status != http.StatusForbidden {
t.Errorf("service credential: %d, want 403", status)
}
- status, body := postRun(t, fx.runsURL(runTestHost), stranger, okMeta(), nil)
- want := canonicalNotFound(t)
- if status != want.Status || body != want.Body {
- t.Errorf("stranger: %d %q, want the byte-identical 404", status, body)
- }
+ status, h, body := postRunH(t, fx.runsURL(runTestHost), stranger, okMeta(), nil)
+ assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body)
if status, body := postRun(t, fx.runsURL("nobody.go.tito.io"), fx.member, okMeta(), nil); status != http.StatusNotFound || errorOf(t, body) != "nobody.go.tito.io is not an instance of titogo" {
t.Errorf("no record: %d %s", status, body)
}
@@ -218,6 +243,7 @@ func TestAPIRunCreateSpendsTheInstanceBudget(t *testing.T) {
func TestAPIRunCreateRefusesAPartThatDisagreesWithMeta(t *testing.T) {
fx := runFixture(t)
+ spool := spoolDir(t)
meta := okMeta()
meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
for name, parts := range map[string][]struct{ name, body string }{
@@ -225,11 +251,13 @@ func TestAPIRunCreateRefusesAPartThatDisagreesWithMeta(t *testing.T) {
"long": {{"x.csv", "abcdefg"}},
"wrong name": {{"y.csv", "abcde"}},
"missing": nil,
+ "extra": {{"x.csv", "abcde"}, {"y.csv", "abcde"}},
} {
status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, parts)
if status != http.StatusBadRequest || errorOf(t, body) != "the run request is malformed" {
t.Errorf("%s: %d %s", name, status, body)
}
+ assertNoSpool(t, spool, name)
}
if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "titogo")); len(left) != 0 {
t.Fatalf("a refused upload left %q", left)
@@ -326,4 +354,114 @@ func TestAPIRunCreateOnAHostClaimConsole(t *testing.T) {
if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg {
t.Fatalf("%d %s", status, body)
}
+ // RUN-C42 comes right after membership: a stranger still gets the
+ // byte-identical 404, and a member with no budget left gets RUN-C42.
+ stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-stranger")
+ status, h, body := postRunH(t, fx.runsURL(runTestHost), stranger, okMeta(), nil)
+ assertNotFound(t, "stranger on a claim console", canonicalNotFound(t), status, h, body)
+ fx.srv.cli.instances = newRateLimiter(1, cliRateWindow)
+ fx.srv.cli.instances.allow(hashToken(fx.member), time.Now())
+ if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg {
+ t.Fatalf("over budget on a claim console: %d %s", status, body)
+ }
+ fx.srv.cfg.HostClaim = false
+ if status, _ := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusTooManyRequests {
+ t.Fatalf("the same member off a claim console: %d, want the 429 the claim console skipped", status)
+ }
+}
+
+// The body cap (file bytes plus the meta allowance) ends a stream that grows
+// past it with the 64 MB refusal: here a full-size declared file behind a
+// meta part padded past the allowance.
+func TestAPIRunCreateRefusesABodyPastTheCap(t *testing.T) {
+ fx := runFixture(t)
+ spool := spoolDir(t)
+ meta := okMeta()
+ meta.Files = []runs.MetaFile{{Name: "big", Size: runs.MaxFileBytes}}
+ var mb bytes.Buffer
+ json.NewEncoder(&mb).Encode(meta)
+ pad := strings.Repeat(" ", runs.MetaAllowance) // trailing white space: valid JSON, counted by the cap
+ var buf bytes.Buffer
+ mw := multipart.NewWriter(&buf)
+ mp, _ := mw.CreateFormField("meta")
+ mp.Write(mb.Bytes())
+ mp.Write([]byte(pad))
+ fp, _ := mw.CreateFormFile("file", "big")
+ fp.Write(make([]byte, runs.MaxFileBytes))
+ mw.Close()
+ if buf.Len() <= runs.MaxFileBytes+runs.MetaAllowance {
+ t.Fatalf("body is %d bytes, under the cap; the test needs a longer pad", buf.Len())
+ }
+ req, _ := http.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &buf)
+ req.Header.Set("Content-Type", mw.FormDataContentType())
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out, _ := io.ReadAll(res.Body)
+ res.Body.Close()
+ if res.StatusCode != http.StatusRequestEntityTooLarge || errorOf(t, string(out)) != "the files come to more than 64 MB" {
+ t.Fatalf("%d %s", res.StatusCode, out)
+ }
+ assertNoSpool(t, spool, "a body past the cap")
+ if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "titogo")); len(left) != 0 {
+ t.Fatalf("left %q", left)
+ }
+}
+
+// commitThenFailStore commits the first write to the instance record and then
+// answers err, as a lost response or a cancelled request would: the record
+// holds the run, the caller was told nothing.
+type commitThenFailStore struct {
+ release.Store
+ recordKey string
+ err error
+ once sync.Once
+}
+
+func (c *commitThenFailStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ etag, err := c.Store.Put(ctx, key, body, ifMatch)
+ if err == nil && key == c.recordKey {
+ c.once.Do(func() { err = c.err })
+ }
+ return etag, err
+}
+
+// An admission whose answer was lost is still an admission: the door says
+// 202 and keeps the run's objects, because the record holds the run and a
+// box will read them.
+func TestAPIRunCreateAdmissionWhoseAnswerWasLost(t *testing.T) {
+ for name, failure := range map[string]error{
+ // The SDK re-sent a committed write: 412, then updateRecord's retry
+ // finds this very run queued.
+ "retried write": fmt.Errorf("s3store: record: %w", release.ErrCASConflict),
+ // A transport error, or the request's context cancelled, after the
+ // write landed: the door asks the record.
+ "transport error": errors.New("connection reset by peer"),
+ } {
+ t.Run(name, func(t *testing.T) {
+ fx := runFixture(t)
+ fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "titogo", runTestHost), err: failure}
+ meta := okMeta()
+ meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
+ status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
+ if status != http.StatusAccepted {
+ t.Fatalf("%d %s", status, body)
+ }
+ var c runs.Created
+ if err := json.Unmarshal([]byte(body), &c); err != nil || c.State != provision.RunQueued {
+ t.Fatalf("created %+v, %v", c, err)
+ }
+ ctx := context.Background()
+ rec, err := provision.GetRecord(ctx, fx.rel, release.OpsAccount, "titogo", runTestHost)
+ if err != nil || rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID {
+ t.Fatalf("record run %+v, %v", rec.ActiveRun, err)
+ }
+ keys, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", c.ID))
+ if len(keys) != 2 {
+ t.Fatalf("the admitted run's objects are %q, want request.json and in/x.csv", keys)
+ }
+ })
+ }
}