carlos / platform Public

console: the run sweep writes the row from the result, runs retention every 20th pass, bounds a pass (review round 1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev 9497518cab3942491472be4ebaf2a30de7dc268a parent 5cab843
5 files changed, +212 −73
  • internal/console/apirunsfollow.go +18 −3
  • internal/console/runsweep.go +69 −40
  • internal/console/runsweep_test.go +124 −28
  • internal/console/server.go +0 −1
  • internal/console/testdata/claim-authority-sites.txt +1 −1

CI passed — run details

diff --git a/internal/console/apirunsfollow.go b/internal/console/apirunsfollow.go
index e5bc8eba0..74620c72c 100644
--- a/internal/console/apirunsfollow.go
+++ b/internal/console/apirunsfollow.go
@@ -174,7 +174,7 @@ func (s *Server) handleAPIRunGet(w http.ResponseWriter, r *http.Request) {
p.Chunks = append(p.Chunks, runs.Chunk{Seq: seq, Stream: stream, Data: raw})
}
if res != nil {
- s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "result", res)
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
}
writeJSON(w, http.StatusOK, p)
}
@@ -248,7 +248,7 @@ func (s *Server) endUnpicked(ctx context.Context, c runCall, state string) error
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)
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(&res), &res)
return nil
}
@@ -332,7 +332,7 @@ func (s *Server) handleAPIRunAck(w http.ResponseWriter, r *http.Request) {
jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has not finished yet", c.id))
return
}
- s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "result", res)
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
if run != nil {
// Clear the record first. If the CAS fails, or its outcome is unknown,
// the record is asked again before anything is deleted: a run the
@@ -352,6 +352,21 @@ func (s *Server) handleAPIRunAck(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, runs.StateAnswer{State: "acknowledged"})
}
+// runEndOutcome is the run-end row's outcome, derived from the result alone so
+// that whoever writes the row first (a follower's poll, an ack, a cancel, an
+// expiry, the sweep) writes the same word: lost, cancelled, expired, or result.
+func runEndOutcome(res *runs.Result) string {
+ switch {
+ case res.Interrupted == runs.InterruptedLost:
+ return "lost"
+ case res.Cancelled:
+ return provision.RunCancelled
+ case res.Refused == runs.RefusedExpired:
+ return provision.RunExpired
+ }
+ return "result"
+}
+
// auditRunEnd writes the run-end row exactly once per run (run spec §6):
// whoever creates runs/<id>/audited first writes it, from the CLI's poll,
// a cancel, an expiry or the sweep. The actor is the run's caller. It runs
diff --git a/internal/console/runsweep.go b/internal/console/runsweep.go
index 1237255ff..55c351be2 100644
--- a/internal/console/runsweep.go
+++ b/internal/console/runsweep.go
@@ -13,7 +13,17 @@ import (
"github.com/carlosframework/platform/internal/runs"
)
-const runSweepInterval = 30 * time.Second
+const (
+ runSweepInterval = 30 * time.Second
+ // runSweepHousekeepingEvery: expiry and lost-run closing run on every
+ // pass, because a caller waits on them; retention and orphans have 24
+ // hour thresholds, so they run on every 20th pass (about 10 minutes).
+ runSweepHousekeepingEvery = 20
+)
+
+// runSweepTimeout bounds one pass, so a hung list or read cannot stall the
+// sweep for good. A variable so a test can shorten it.
+var runSweepTimeout = 5 * time.Minute
// StartRunSweep runs the run sweep every 30 seconds (run spec §5.8). The
// console has no deadline-sweep loop to join (index item 1), so this is a
@@ -25,8 +35,8 @@ func (s *Server) StartRunSweep(ctx context.Context) {
}
t := time.NewTicker(runSweepInterval)
defer t.Stop()
- for {
- s.runSweep(ctx, time.Now())
+ for pass := 0; ; pass++ {
+ s.runSweepPass(ctx, time.Now(), pass%runSweepHousekeepingEvery == 0)
select {
case <-ctx.Done():
return
@@ -35,14 +45,20 @@ func (s *Server) StartRunSweep(ctx context.Context) {
}
}
-// runSweep is one pass. Every transition is a CAS on the record, so a box
+// runSweep is one full pass: every phase.
+func (s *Server) runSweep(ctx context.Context, now time.Time) { s.runSweepPass(ctx, now, true) }
+
+// runSweepPass is one pass. Every transition is a CAS on the record, so a box
// or a CLI that acts first wins and the sweep changes nothing: a queued run
// a box picked in the meantime stays picked, a run a box finished stays
-// finished. It is called from one goroutine.
-func (s *Server) runSweep(ctx context.Context, now time.Time) {
+// finished. housekeeping adds the terminal-retention and orphan phases to the
+// expiry and lost-run phases every pass runs. It is called from one goroutine.
+func (s *Server) runSweepPass(ctx context.Context, now time.Time, housekeeping bool) {
if s.cfg.Releases == nil {
return
}
+ ctx, cancel := context.WithTimeout(ctx, runSweepTimeout)
+ defer cancel()
recs, err := provision.ListAllRecords(ctx, s.cfg.Releases)
if err != nil {
log.Printf("console: run sweep: list records: %v", err)
@@ -51,6 +67,10 @@ func (s *Server) runSweep(ctx context.Context, now time.Time) {
named := map[string]bool{}
apps := map[[2]string]bool{}
for _, rec := range recs {
+ if ctx.Err() != nil {
+ log.Printf("console: run sweep: pass cut short: %v", ctx.Err())
+ return
+ }
apps[[2]string{rec.Account, rec.App}] = true
r := rec.ActiveRun
if r == nil {
@@ -58,7 +78,11 @@ func (s *Server) runSweep(ctx context.Context, now time.Time) {
}
named[runs.Prefix(rec.Account, rec.App, r.ID)] = true
acct, err := s.store.AccountBySqid(rec.Account)
- if err != nil || acct == nil {
+ if err != nil {
+ log.Printf("console: run sweep: account %s: %v", rec.Account, err)
+ continue
+ }
+ if acct == nil {
continue
}
c := runCall{acct: acct, app: rec.App, host: rec.Host, id: r.ID}
@@ -67,15 +91,20 @@ func (s *Server) runSweep(ctx context.Context, now time.Time) {
s.sweepExpireQueued(ctx, c)
case (r.State == provision.RunPicked || r.State == provision.RunCancel) && now.After(r.Deadline.Add(runs.LostAfter)):
s.sweepLostRun(ctx, c, r, now)
- case !r.Open():
+ case !r.Open() && housekeeping:
s.sweepTerminalRun(ctx, c, r, now)
}
}
- strays := map[string]bool{}
+ if !housekeeping {
+ return
+ }
for app := range apps {
- s.sweepOrphanRuns(ctx, app[0], app[1], named, strays, now)
+ if ctx.Err() != nil {
+ log.Printf("console: run sweep: pass cut short: %v", ctx.Err())
+ return
+ }
+ s.sweepOrphanRuns(ctx, app[0], app[1], named, now)
}
- s.runSweepStrays = strays
}
// sweepExpireQueued expires a queued run past its pickup deadline. The
@@ -125,13 +154,9 @@ func (s *Server) sweepLostRun(ctx context.Context, c runCall, r *provision.RunSt
}
res, err := runs.ReadResult(ctx, rel, c.acct.Sqid, c.app, c.id)
if err != nil || res == nil {
- return // the run is done; a follower's poll writes the row
+ return // the run is done; the terminal phase writes the row
}
- outcome := "result"
- if res.Interrupted == runs.InterruptedLost {
- outcome = "lost"
- }
- s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, outcome, res)
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
}
// sweepTerminalRun clears a terminal run and deletes its objects once its
@@ -163,6 +188,13 @@ func (s *Server) sweepTerminalRun(ctx context.Context, c runCall, r *provision.R
return
}
}
+ if res != nil {
+ // The run-end row for any terminal run with a result, before anything
+ // is cleared: a run nobody followed or acked (a box's own finish), or
+ // one whose sweep died after the done CAS, has no row yet. The marker
+ // makes this a no-op when the row exists.
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
+ }
if now.Sub(ended) <= runs.KeepFor {
return
}
@@ -184,9 +216,8 @@ func (s *Server) sweepTerminalRun(ctx context.Context, c runCall, r *provision.R
// a day old. Without one it is a stray (request.json is written first and
// deleted last but for the audited marker, so it is neither a create in
// flight nor a run anyone can read): it goes once its newest object is a day
-// old when the store lists modification times, and otherwise on the second
-// sweep that sees it. strays collects the ones seen and kept this pass.
-func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named, strays map[string]bool, now time.Time) {
+// old, and stays when the store cannot say.
+func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named map[string]bool, now time.Time) {
rel := s.cfg.Releases
dirs, err := rel.ListDirs(ctx, runs.RunsPrefix(acct, app))
if err != nil {
@@ -207,7 +238,7 @@ func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named, s
continue
}
case errors.Is(err, os.ErrNotExist):
- if !s.strayIsOld(ctx, dir, strays, now) {
+ if !s.strayIsOld(ctx, dir, now) {
continue
}
default:
@@ -219,25 +250,23 @@ func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named, s
}
}
-// strayIsOld reports whether a run directory with no request.json may be
-// deleted now, and records it in strays when not.
-func (s *Server) strayIsOld(ctx context.Context, dir string, strays map[string]bool, now time.Time) bool {
- if ml, ok := s.cfg.Releases.(release.ModLister); ok {
- objs, err := ml.ListMod(ctx, dir)
- if err != nil || len(objs) == 0 {
- return false
- }
- newest := objs[0].Mod
- for _, o := range objs[1:] {
- if o.Mod.After(newest) {
- newest = o.Mod
- }
- }
- return now.Sub(newest) > runs.KeepFor
+// strayIsOld reports whether a run directory with no request.json has had
+// nothing written to it for a day. A store that cannot list modification
+// times (both real stores can) leaves strays alone.
+func (s *Server) strayIsOld(ctx context.Context, dir string, now time.Time) bool {
+ ml, ok := s.cfg.Releases.(release.ModLister)
+ if !ok {
+ return false
}
- if s.runSweepStrays[dir] {
- return true
+ objs, err := ml.ListMod(ctx, dir)
+ if err != nil || len(objs) == 0 {
+ return false
+ }
+ newest := objs[0].Mod
+ for _, o := range objs[1:] {
+ if o.Mod.After(newest) {
+ newest = o.Mod
+ }
}
- strays[dir] = true
- return false
+ return now.Sub(newest) > runs.KeepFor
}
diff --git a/internal/console/runsweep_test.go b/internal/console/runsweep_test.go
index 8bbdf1bd6..860c3cdc7 100644
--- a/internal/console/runsweep_test.go
+++ b/internal/console/runsweep_test.go
@@ -294,8 +294,8 @@ func TestRunSweepReapsAStrayRunDirectoryByItsNewestObject(t *testing.T) {
}
}
-// Without modification times the stray goes on the second sweep that sees it.
-func TestRunSweepReapsAStrayRunDirectoryOnTheSecondSweepWithoutModTimes(t *testing.T) {
+// A store that cannot list modification times leaves strays alone.
+func TestRunSweepLeavesAStrayAloneWithoutModTimes(t *testing.T) {
fx := runFixture(t)
ctx := context.Background()
fx.srv.cfg.Releases = plainStore{fx.rel}
@@ -306,49 +306,145 @@ func TestRunSweepReapsAStrayRunDirectoryOnTheSecondSweepWithoutModTimes(t *testi
if _, err := fx.rel.Put(ctx, runs.ResultKey(release.OpsAccount, "titogo", id), []byte(`{}`), ""); err != nil {
t.Fatal(err)
}
- fx.srv.runSweep(ctx, time.Now())
- if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
- t.Fatal("deleted on the first sweep that saw it")
+ for i := 0; i < 3; i++ {
+ fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
}
- fx.srv.runSweep(ctx, time.Now())
- if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) != 0 {
- t.Fatalf("survived the second sweep: %q", left)
+ if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
+ t.Fatal("a stray was deleted by a store that cannot say how old it is")
}
}
-// A stray that is gone and comes back is a new stray: the count starts again.
-func TestRunSweepForgetsAStrayThatWentAway(t *testing.T) {
+func TestRunSweepLeavesAnUnreadableRequestAlone(t *testing.T) {
fx := runFixture(t)
ctx := context.Background()
- fx.srv.cfg.Releases = plainStore{fx.rel}
- id := "run_gggggggggggggggggggggggggg"
- key := runs.ResultKey(release.OpsAccount, "titogo", id)
- if _, err := fx.rel.Put(ctx, key, []byte(`{}`), ""); err != nil {
+ id := "run_hhhhhhhhhhhhhhhhhhhhhhhhhh"
+ if _, err := fx.rel.Put(ctx, runs.RequestKey(release.OpsAccount, "titogo", id), bytes.Repeat([]byte("x"), 8), ""); err != nil {
t.Fatal(err)
}
- fx.srv.runSweep(ctx, time.Now())
- if err := fx.rel.Delete(ctx, key); err != nil {
+ fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
+ if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
+ t.Fatal("a directory whose request could not be read was deleted")
+ }
+}
+
+// A follower's poll that lands between the sweep's lost result and its done
+// CAS wins the audit marker. The row must still say lost.
+func TestRunSweepLostRunRowSaysLostWhenAPollGotThereFirst(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
+ lost := runs.Result{Interrupted: runs.InterruptedLost, Box: "box-1", Ended: time.Now().Add(2 * time.Hour)}
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, lost); err != nil {
t.Fatal(err)
}
- fx.srv.runSweep(ctx, time.Now())
- if _, err := fx.rel.Put(ctx, key, []byte(`{}`), ""); err != nil {
+ if p := fx.page(t, id, 0); p.Result == nil {
+ t.Fatal("the poll did not see the result")
+ }
+ fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
+ if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
+ t.Fatalf("%+v", r)
+ }
+ if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
+ t.Fatalf("run-end rows %q", ends)
+ }
+}
+
+// A run a box finished that nobody followed or acked gets its one row from
+// the sweep, before the sweep clears it.
+func TestRunSweepWritesTheRowOfAFinishedRunNobodyFollowed(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ ended := time.Now()
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, runs.Result{Exit: 2, Ended: ended}); err != nil {
t.Fatal(err)
}
- fx.srv.runSweep(ctx, time.Now())
- if _, _, err := fx.rel.Get(ctx, key); err != nil {
- t.Fatalf("a new stray was deleted on its first sweep: %v", err)
+ fx.transition(t, provision.RunFinish(id))
+ fx.srv.runSweep(context.Background(), ended.Add(time.Hour))
+ if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"result"`) {
+ t.Fatalf("run-end rows %q", ends)
+ }
+ fx.srv.runSweep(context.Background(), ended.Add(25*time.Hour))
+ if r := fx.record(t).ActiveRun; r != nil {
+ t.Fatalf("not cleared: %+v", r)
+ }
+ if ends := fx.runEnds(t, id); len(ends) != 1 {
+ t.Fatalf("run-end rows after the clear %q", ends)
}
}
-func TestRunSweepLeavesAnUnreadableRequestAlone(t *testing.T) {
+// A run the sweep marked done and then died: no row. A later pass writes it,
+// as lost.
+func TestRunSweepWritesTheRowOfALostRunItsSweepDiedOn(t *testing.T) {
fx := runFixture(t)
- ctx := context.Background()
- id := "run_hhhhhhhhhhhhhhhhhhhhhhhhhh"
- if _, err := fx.rel.Put(ctx, runs.RequestKey(release.OpsAccount, "titogo", id), bytes.Repeat([]byte("x"), 8), ""); err != nil {
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
+ lost := runs.Result{Interrupted: runs.InterruptedLost, Box: "box-1", Ended: time.Now()}
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, lost); err != nil {
t.Fatal(err)
}
- fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
- if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
- t.Fatal("a directory whose request could not be read was deleted")
+ fx.transition(t, provision.RunFinish(id))
+ fx.srv.runSweep(context.Background(), time.Now().Add(time.Hour))
+ if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
+ t.Fatalf("run-end rows %q", ends)
+ }
+}
+
+// Expiry and lost-run closing run on every pass; retention and orphans only
+// on housekeeping passes.
+func TestRunSweepHousekeepingPhasesSkipOffPasses(t *testing.T) {
+ fx := runFixture(t)
+ ctx := context.Background()
+ old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "titogo", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
+ fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
+ fx.startRun(t)
+ fx.srv.runSweepPass(ctx, time.Now().Add(2*time.Minute), false)
+ if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
+ t.Fatalf("expiry did not run on an off pass: %+v", r)
+ }
+ if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) == 0 {
+ t.Fatal("the orphan phase ran on an off pass")
+ }
+ fx.srv.runSweepPass(ctx, time.Now().Add(26*time.Hour), false)
+ if fx.record(t).ActiveRun == nil {
+ t.Fatal("the retention phase ran on an off pass")
+ }
+ fx.srv.runSweepPass(ctx, time.Now().Add(26*time.Hour), true)
+ if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) != 0 {
+ t.Fatalf("the orphan phase did not run on a housekeeping pass: %q", left)
+ }
+ if r := fx.record(t).ActiveRun; r != nil {
+ t.Fatalf("the retention phase did not run on a housekeeping pass: %+v", r)
+ }
+}
+
+// hangingStore blocks every call until its context is done.
+type hangingStore struct{ release.Store }
+
+func (hangingStore) Get(ctx context.Context, _ string) ([]byte, string, error) {
+ <-ctx.Done()
+ return nil, "", ctx.Err()
+}
+func (hangingStore) List(ctx context.Context, _ string) ([]string, error) {
+ <-ctx.Done()
+ return nil, ctx.Err()
+}
+func (hangingStore) ListDirs(ctx context.Context, _ string) ([]string, error) {
+ <-ctx.Done()
+ return nil, ctx.Err()
+}
+
+func TestRunSweepPassIsBoundedByATimeout(t *testing.T) {
+ fx := runFixture(t)
+ fx.srv.cfg.Releases = hangingStore{fx.rel}
+ old := runSweepTimeout
+ runSweepTimeout = 50 * time.Millisecond
+ defer func() { runSweepTimeout = old }()
+ done := make(chan struct{})
+ go func() { fx.srv.runSweep(context.Background(), time.Now()); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(10 * time.Second):
+ t.Fatal("a hung store stalled the pass")
}
}
diff --git a/internal/console/server.go b/internal/console/server.go
index 64d47049b..a0827041b 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -827,7 +827,6 @@ type Server struct {
claimBackendObserver func(backendEvent) // fixture recorder only
claimEntries claimInventory
claimWorkers map[string]func(context.Context)
- runSweepStrays map[string]bool // run directories with no request.json the last sweep saw (runsweep.go)
dbHandleKey []byte
cfg Config
store *Store
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt
index 55830c730..071c68898 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -83,7 +83,7 @@ economicspass.go|sizeProvisionedStore|objstore.S3ClientFor(ctx, p.Region, p.Endp
geoippublish.go|Commit|os.Open(st.path)
hostwakereset.go|registerWakeRetryReset|a.Control.RegisterStart(ctx, h.Account, h.App, reg.AppID, id, reg, registeredAt)
hostwakereset.go|registerWakeRetryReset|start.FinishWakeRetryReset(cleanup)
-runsweep.go|runSweep|r.Open()
+runsweep.go|runSweepPass|r.Open()
scanpass.go|storeLister|objstore.S3ClientFor(ctx, p.Region, p.Endpoint)
store.go|Open|sql.Open("sqlite", path)
store_gate.go|Connect|k.g.d.Open(k.g.dsn)