amadan

carlos / platform Public

console: every terminal run has a result; the run-end row survives a partial ack delete (review round 1)

DeleteRun removes request.json and the audited marker last; auditRunEnd takes
back a marker written after the run was deleted; a run cancelled or expired
before pickup gets a console-written result.json.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 36732307e3b80521f5acd51c794d532829f07bca parent bec0204
5 files changed, +374 −13
  • internal/console/apirunsfollow.go +49 −8
  • internal/console/apirunsfollow_test.go +253 −0
  • internal/runs/runs.go +4 −0
  • internal/runs/runs_test.go +40 −0
  • internal/runs/store.go +28 −5

CI passed — run details

diff --git a/internal/console/apirunsfollow.go b/internal/console/apirunsfollow.go
index bf28905af..e5bc8eba0 100644
--- a/internal/console/apirunsfollow.go
+++ b/internal/console/apirunsfollow.go
@@ -196,7 +196,11 @@ func (s *Server) handleAPIRunCancel(w http.ResponseWriter, r *http.Request) {
}
switch got.ActiveRun.State {
case provision.RunCancelled:
- s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "cancelled", nil)
+ if err := s.endUnpicked(ctx, c, provision.RunCancelled); err != nil {
+ log.Printf("console: api run %s/%s/%s %s: end after cancel: %v", c.acct.Sqid, c.app, c.host, c.id, err)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
case provision.RunCancel:
s.bumpSerial(ctx)
}
@@ -221,16 +225,39 @@ func (s *Server) handleAPIRunExpire(w http.ResponseWriter, r *http.Request) {
s.runTransitionRefused(w, ctx, c, err)
return
}
- s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "expired", nil)
+ if err := s.endUnpicked(ctx, c, provision.RunExpired); err != nil {
+ log.Printf("console: api run %s/%s/%s %s: end after expire: %v", c.acct.Sqid, c.app, c.host, c.id, err)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State})
}
+// endUnpicked finishes a run that ended before any box picked it: cancelled
+// or expired is terminal, so no box will ever write its result.json. The
+// console writes one (create-only, so a repeat changes nothing), so that every
+// terminal run has a result a follower can read after the record has moved
+// on, and then the run-end row. state is the record's terminal state.
+func (s *Server) endUnpicked(ctx context.Context, c runCall, state string) error {
+ res := runs.Result{Ended: time.Now().UTC()}
+ if state == provision.RunCancelled {
+ res.Cancelled = true
+ } else {
+ res.Refused = runs.RefusedExpired
+ }
+ if err := runs.WriteResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id, res); err != nil {
+ return err
+ }
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, state, &res)
+ return nil
+}
+
// runTransitionRefused names why a cancel or expire did not apply, from a
// fresh read of the record. A run the record shows cancelled or expired
-// gets its run-end row here too (written once, so a no-op when the call
-// that ended it already wrote it): a cancel whose response was lost and was
-// retried finds its own terminal state, and the row must not depend on the
-// first attempt having got that far.
+// gets its result and run-end row here too (both written once, so a no-op
+// when the call that ended it already wrote them): a cancel whose response
+// was lost and was retried finds its own terminal state, and these must not
+// depend on the first attempt having got that far.
func (s *Server) runTransitionRefused(w http.ResponseWriter, ctx context.Context, c runCall, err error) {
if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
log.Printf("console: api run %s/%s/%s %s: %v", c.acct.Sqid, c.app, c.host, c.id, err)
@@ -247,7 +274,11 @@ func (s *Server) runTransitionRefused(w http.ResponseWriter, ctx context.Context
jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has already started", c.id))
default:
if run.State == provision.RunCancelled || run.State == provision.RunExpired {
- s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, run.State, nil)
+ if err := s.endUnpicked(ctx, c, run.State); err != nil {
+ log.Printf("console: api run %s/%s/%s %s: end after retry: %v", c.acct.Sqid, c.app, c.host, c.id, err)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
}
jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has already ended", c.id))
}
@@ -335,8 +366,18 @@ func (s *Server) auditRunEnd(ctx context.Context, acct *Account, app, host, id,
}
return // written already, or the bucket refused: either way not twice
}
+ // The marker is the last object DeleteRun removes, after request.json. A
+ // marker Put that succeeds for a run whose request.json is gone therefore
+ // raced a finished ack: the row was written already. Take the marker back
+ // and write nothing.
by := ""
- if req, err := runs.ReadRequest(ctx, s.cfg.Releases, acct.Sqid, app, id); err == nil {
+ switch req, err := runs.ReadRequest(ctx, s.cfg.Releases, acct.Sqid, app, id); {
+ case errors.Is(err, os.ErrNotExist):
+ if derr := s.cfg.Releases.Delete(ctx, runs.AuditedKey(acct.Sqid, app, id)); derr != nil {
+ log.Printf("console: api run %s/%s/%s %s: remove stray audit marker: %v", acct.Sqid, app, host, id, derr)
+ }
+ return
+ case err == nil:
by = req.By
}
d := struct {
diff --git a/internal/console/apirunsfollow_test.go b/internal/console/apirunsfollow_test.go
index 921fded45..c4480f2b2 100644
--- a/internal/console/apirunsfollow_test.go
+++ b/internal/console/apirunsfollow_test.go
@@ -3,8 +3,10 @@ package console
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"net/http"
+ "slices"
"strings"
"testing"
"time"
@@ -331,3 +333,254 @@ func TestAPIRunRoutesRefuseAServiceCredentialAndABadID(t *testing.T) {
t.Errorf("bad id: %d %s", status, body)
}
}
+
+// faultStore injects the failures a bucket has: a delete that fails, a Put
+// whose response is lost (landed or not), a Get that fails.
+type faultStore struct {
+ release.Store
+ delete func(key string) error
+ put func(key string) (landed bool, err error)
+ get func(key string) error
+}
+
+func (f *faultStore) Delete(ctx context.Context, key string) error {
+ if f.delete != nil {
+ if err := f.delete(key); err != nil {
+ return err
+ }
+ }
+ return f.Store.Delete(ctx, key)
+}
+
+func (f *faultStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if f.put != nil {
+ if landed, err := f.put(key); err != nil {
+ if landed {
+ _, _ = f.Store.Put(ctx, key, body, ifMatch)
+ }
+ return "", err
+ }
+ }
+ return f.Store.Put(ctx, key, body, ifMatch)
+}
+
+func (f *faultStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if f.get != nil {
+ if err := f.get(key); err != nil {
+ return nil, "", err
+ }
+ }
+ return f.Store.Get(ctx, key)
+}
+
+// runEnds is the details of the run-end rows written for id.
+func (fx *runFx) runEnds(t *testing.T, id string) []string {
+ t.Helper()
+ events, err := fx.st.Events(OpsAccountID, "titogo", 50)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out []string
+ for _, e := range events {
+ if e.Action == "run-end" && strings.Contains(e.Detail, id) {
+ out = append(out, e.Detail)
+ }
+ }
+ return out
+}
+
+func (fx *runFx) left(id string) []string {
+ left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id))
+ return left
+}
+
+// finishedRun is a picked run with one chunk and a result, finished on the record.
+func (fx *runFx) finishedRun(t *testing.T) string {
+ t.Helper()
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ fx.chunk(t, id, 1, runs.StreamOut, "a")
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, runs.Result{Exit: 0, LastSeq: 1}); err != nil {
+ t.Fatal(err)
+ }
+ fx.transition(t, provision.RunFinish(id))
+ return id
+}
+
+func TestAPIRunAckThatCannotDeleteKeepsTheMarkerSoTheRowIsWrittenOnce(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.finishedRun(t)
+ failing := true
+ fx.srv.cfg.Releases = &faultStore{Store: fx.rel, delete: func(key string) error {
+ if failing && key == runs.ResultKey(release.OpsAccount, "titogo", id) {
+ return errors.New("delete refused")
+ }
+ return nil
+ }}
+ if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
+ t.Fatalf("ack: %d %s", status, body)
+ }
+ left := fx.left(id)
+ for _, want := range []string{runs.ResultKey(release.OpsAccount, "titogo", id), runs.RequestKey(release.OpsAccount, "titogo", id), runs.AuditedKey(release.OpsAccount, "titogo", id)} {
+ if !slices.Contains(left, want) {
+ t.Fatalf("after a failed delete %q is gone; left %q", want, left)
+ }
+ }
+ // A follower and the retried ack both find the result; neither writes a second row.
+ failing = false
+ if p := fx.page(t, id, 1); p.Result == nil {
+ t.Fatalf("page %+v", p)
+ }
+ if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
+ t.Fatalf("retried ack: %d %s", status, body)
+ }
+ if left := fx.left(id); len(left) != 0 {
+ t.Fatalf("retried ack left %q", left)
+ }
+ if rows := fx.runEnds(t, id); len(rows) != 1 {
+ t.Fatalf("run-end rows %q, want 1", rows)
+ }
+}
+
+func TestAPIRunEndIsNotWrittenForARunAnAckHasDeleted(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.finishedRun(t)
+ if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
+ t.Fatalf("ack: %d %s", status, body)
+ }
+ // A follower that read the result before the ack's delete reaches the
+ // audit after it.
+ fx.srv.auditRunEnd(context.Background(), &Account{ID: OpsAccountID, Sqid: release.OpsAccount}, "titogo", runTestHost, id, "result", &runs.Result{LastSeq: 1})
+ if left := fx.left(id); len(left) != 0 {
+ t.Fatalf("a late audit left %q", left)
+ }
+ if rows := fx.runEnds(t, id); len(rows) != 1 {
+ t.Fatalf("run-end rows %q, want 1", rows)
+ }
+}
+
+func TestAPIRunCancelAndExpireBeforePickupLeaveAResult(t *testing.T) {
+ fx := runFixture(t)
+ cancelled := fx.startRun(t)
+ if status, body := fx.post(t, cancelled, "cancel", ""); status != http.StatusOK {
+ t.Fatalf("cancel: %d %s", status, body)
+ }
+ fx.transition(t, provision.RunClear(cancelled))
+ expiredID := fx.startRun(t)
+ if status, body := fx.post(t, expiredID, "expire", ""); status != http.StatusOK {
+ t.Fatalf("expire: %d %s", status, body)
+ }
+ fx.transition(t, provision.RunClear(expiredID))
+ // The record no longer names either run; a follower still learns how it ended.
+ if p := fx.page(t, cancelled, 0); p.State != "" || p.Result == nil || !p.Result.Cancelled || p.More || len(p.Chunks) != 0 {
+ t.Errorf("cancelled run page %+v", p)
+ }
+ if p := fx.page(t, expiredID, 0); p.State != "" || p.Result == nil || p.Result.Refused != runs.RefusedExpired {
+ t.Errorf("expired run page %+v", p)
+ }
+ if status, body := fx.post(t, cancelled, "ack", `{"last_seq":0}`); status != http.StatusOK {
+ t.Errorf("ack of a cancelled run: %d %s", status, body)
+ }
+ if left := fx.left(cancelled); len(left) != 0 {
+ t.Errorf("left %q", left)
+ }
+}
+
+func TestAPIRunExpireWritesTheRunEndRow(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ if status, body := fx.post(t, id, "expire", ""); status != http.StatusOK {
+ t.Fatalf("expire: %d %s", status, body)
+ }
+ rows := fx.runEnds(t, id)
+ if len(rows) != 1 || !strings.Contains(rows[0], `"outcome":"expired"`) {
+ t.Fatalf("run-end rows %q", rows)
+ }
+}
+
+// A cancel or expire whose CAS landed but whose response was lost is retried
+// and finds its own terminal state; the result and the row are still written.
+func TestAPIRunRetriedCancelAndExpireStillEndTheRun(t *testing.T) {
+ fx := runFixture(t)
+ for _, c := range []struct {
+ verb, outcome string
+ mut func(string) func(*provision.Record) error
+ }{
+ {"cancel", "cancelled", provision.RunCancelTransition},
+ {"expire", "expired", provision.RunExpire},
+ } {
+ id := fx.startRun(t)
+ fx.transition(t, c.mut(id))
+ if status, body := fx.post(t, id, c.verb, ""); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has already ended" {
+ t.Fatalf("%s retry: %d %s", c.verb, status, body)
+ }
+ if res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id); err != nil || res == nil {
+ t.Errorf("%s retry left no result: %+v, %v", c.verb, res, err)
+ }
+ if rows := fx.runEnds(t, id); len(rows) != 1 || !strings.Contains(rows[0], `"outcome":"`+c.outcome+`"`) {
+ t.Errorf("%s retry rows %q", c.verb, rows)
+ }
+ fx.post(t, id, c.verb, "")
+ if rows := fx.runEnds(t, id); len(rows) != 1 {
+ t.Errorf("%s retried twice: rows %q", c.verb, rows)
+ }
+ fx.transition(t, provision.RunClear(id))
+ }
+}
+
+func TestAPIRunAckKeepsTheObjectsWhenTheClearIsNotSettled(t *testing.T) {
+ recordKey := provision.RecordKey(release.OpsAccount, "titogo", runTestHost)
+ for _, c := range []struct {
+ name string
+ setup func(*faultStore, *bool)
+ wantStatus int
+ wantKept bool
+ }{
+ {"clear refused", func(f *faultStore, _ *bool) {
+ f.put = func(key string) (bool, error) {
+ if key == recordKey {
+ return false, errors.New("put refused")
+ }
+ return false, nil
+ }
+ }, http.StatusBadGateway, true},
+ {"clear landed, response lost", func(f *faultStore, _ *bool) {
+ f.put = func(key string) (bool, error) {
+ if key == recordKey {
+ return true, errors.New("response lost")
+ }
+ return false, nil
+ }
+ }, http.StatusOK, false},
+ {"clear landed, re-read fails", func(f *faultStore, afterPut *bool) {
+ f.put = func(key string) (bool, error) {
+ if key == recordKey {
+ *afterPut = true
+ }
+ return false, nil
+ }
+ f.get = func(key string) error {
+ if key == recordKey && *afterPut {
+ return errors.New("get refused")
+ }
+ return nil
+ }
+ }, http.StatusBadGateway, true},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.finishedRun(t)
+ f := &faultStore{Store: fx.rel}
+ var afterPut bool
+ c.setup(f, &afterPut)
+ fx.srv.cfg.Releases = f
+ status, body := fx.post(t, id, "ack", `{"last_seq":1}`)
+ if status != c.wantStatus {
+ t.Fatalf("ack: %d %s, want %d", status, body, c.wantStatus)
+ }
+ if kept := len(fx.left(id)) != 0; kept != c.wantKept {
+ t.Fatalf("objects kept = %v, want %v (left %q)", kept, c.wantKept, fx.left(id))
+ }
+ })
+ }
+}
diff --git a/internal/runs/runs.go b/internal/runs/runs.go
index 03f29b45a..158fdfdbc 100644
--- a/internal/runs/runs.go
+++ b/internal/runs/runs.go
@@ -49,6 +49,10 @@ const (
RefusedWake = "wake"
RefusedIncarnation = "incarnation"
RefusedImport = "import"
+ // RefusedExpired is written by the console, not a box: the run expired
+ // before any box picked it up (run spec §5.2), so no box ever wrote a
+ // result. The CLI reports it as "not picked up".
+ RefusedExpired = "expired"
)
// Why a run was interrupted, beside activator's Stop* reasons (RUN-C44 to
diff --git a/internal/runs/runs_test.go b/internal/runs/runs_test.go
index 20b82e926..012d682bd 100644
--- a/internal/runs/runs_test.go
+++ b/internal/runs/runs_test.go
@@ -256,3 +256,43 @@ func TestStoreHelpersRefuseAnAccountOrAppThatIsNotOne(t *testing.T) {
t.Fatalf("a refused call changed the store: %q -> %q", before, after)
}
}
+
+// DeleteRun removes the marker last and request.json just before it, so a
+// run with any object left still has the marker that keeps its row single.
+func TestDeleteRunKeepsTheMarkerAndRequestUntilTheRestIsGone(t *testing.T) {
+ ctx := context.Background()
+ inner := release.NewDirStore(t.TempDir())
+ id, _ := MintID()
+ for _, k := range []string{AuditedKey("bab", "titogo", id), RequestKey("bab", "titogo", id), ResultKey("bab", "titogo", id), ChunkKey("bab", "titogo", id, 1, StreamOut)} {
+ if _, err := inner.Put(ctx, k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ s := &failDelete{Store: inner, fail: ResultKey("bab", "titogo", id)}
+ if err := DeleteRun(ctx, s, "bab", "titogo", id); err == nil {
+ t.Fatal("a failed delete must be returned")
+ }
+ left, _ := inner.List(ctx, Prefix("bab", "titogo", id))
+ if len(left) != 3 { // result, request, audited
+ t.Fatalf("left %q, want result.json, request.json and audited", left)
+ }
+ s.fail = ""
+ if err := DeleteRun(ctx, s, "bab", "titogo", id); err != nil {
+ t.Fatal(err)
+ }
+ if left, _ := inner.List(ctx, Prefix("bab", "titogo", id)); len(left) != 0 {
+ t.Fatalf("left %q", left)
+ }
+}
+
+type failDelete struct {
+ release.Store
+ fail string
+}
+
+func (f *failDelete) Delete(ctx context.Context, key string) error {
+ if key == f.fail {
+ return errors.New("delete refused")
+ }
+ return f.Store.Delete(ctx, key)
+}
diff --git a/internal/runs/store.go b/internal/runs/store.go
index 139837bab..2911bfa58 100644
--- a/internal/runs/store.go
+++ b/internal/runs/store.go
@@ -125,7 +125,13 @@ func LastChunkSeq(ctx context.Context, s release.Store, acct, app, id string) (i
return last, nil
}
-// DeleteRun deletes every object of the run.
+// DeleteRun deletes every object of the run, in an order that keeps a
+// half-deleted run findable and audited once. The input and output objects
+// and result.json go first; request.json, which the orphan sweep ages a run
+// by and which says whose run it was, goes only after they are all gone; the
+// audited marker goes last, so a run that still has any object still has its
+// marker, and nothing can write the run-end row a second time. A tier that
+// fails stops the later ones, and the call can be repeated.
func DeleteRun(ctx context.Context, s release.Store, acct, app, id string) error {
if err := checkRun(acct, app, id); err != nil {
return err
@@ -134,11 +140,28 @@ func DeleteRun(ctx context.Context, s release.Store, acct, app, id string) error
if err != nil {
return err
}
- var errs []error
+ request, audited := RequestKey(acct, app, id), AuditedKey(acct, app, id)
+ tiers := [3][]string{}
for _, k := range keys {
- if err := s.Delete(ctx, k); err != nil {
- errs = append(errs, err)
+ switch k {
+ case audited:
+ tiers[2] = append(tiers[2], k)
+ case request:
+ tiers[1] = append(tiers[1], k)
+ default:
+ tiers[0] = append(tiers[0], k)
}
}
- return errors.Join(errs...)
+ for _, tier := range tiers {
+ var errs []error
+ for _, k := range tier {
+ if err := s.Delete(ctx, k); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ if err := errors.Join(errs...); err != nil {
+ return err
+ }
+ }
+ return nil
}