amadan
console: the run sweep: expiry, lost runs, a day's retention, orphans (run spec §5.8, codex R3-3)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9 files changed,
+606
−0
cmd/console/main.go+3 −0internal/console/claim_invariant_fixtures_test.go+1 −0internal/console/claim_invariant_probes_worker_test.go+1 −0internal/console/claim_workers.go+1 −0internal/console/runsweep.go+243 −0internal/console/runsweep_test.go+354 −0internal/console/server.go+1 −0internal/console/testdata/claim-authority-sites.txt+1 −0internal/console/testdata/claim-routes.txt+1 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go| index 9ef54f0e2..46635fbf4 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -446,6 +446,9 @@ func serve() { |
| // The temporary-app expiry pass (internal/console/expirypass.go). Same |
| // placement rule as every pass above: started here, never in NewServer. |
| go srv.RunClaimWorker(context.Background(), "StartExpiryPass") |
| + // The run sweep (internal/console/runsweep.go): expires queued runs, |
| + // ends lost ones, deletes a day-old run's objects. Same placement rule. |
| + go srv.RunClaimWorker(context.Background(), "StartRunSweep") |
| // The relay's writer lease and its converging reconcile |
| // (internal/console/relayops.go). Same placement rule as every pass |
| // above, and it returns immediately when CARLOS_RELAY_REGIONS is empty — |
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go| index fad8d15f4..b782dd227 100644 |
| --- a/internal/console/claim_invariant_fixtures_test.go |
| +++ b/internal/console/claim_invariant_fixtures_test.go |
| @@ -1423,6 +1423,7 @@ var claimFixtures = map[string]claimFixture{ |
| "worker LogPlacementDiscrepancies": claimRead("logs region drift between the registry and instance records; writes nothing"), |
| "worker PublishMovesConsole": claimNonApp("the moves console record (control/moves-console.json): the platform's moves protocol and key fingerprint; no app, and nothing with the moves switch off"), |
| "worker ReportApexSquats": claimRead("logs apex squats found in the registry; writes nothing"), |
| + "worker StartRunSweep": claimRead("off on a host-claim console, where every run route refuses: it logs that and writes nothing"), |
| "worker StartAlertWatcher": claimNonApp("box alert state under alerts/boxes and operator mail"), |
| "worker StartEconomicsPass": claimNonApp("daily usage census, statements and bills: deployment and account economics"), |
| "worker StartGeoipJobs": claimNonApp("GeoIP orphan sweep, D1 expiry and feed status under control/geoip/: the platform database, no app"), |
diff --git a/internal/console/claim_invariant_probes_worker_test.go b/internal/console/claim_invariant_probes_worker_test.go| index 5c3eb43aa..25e0c6b0c 100644 |
| --- a/internal/console/claim_invariant_probes_worker_test.go |
| +++ b/internal/console/claim_invariant_probes_worker_test.go |
| @@ -74,6 +74,7 @@ var claimWorkerProbes = map[string]claimProbe{ |
| cx.addRoute(t, registry.Route{Host: "titogo.bab.oncarlos.com", Kind: "instance", Status: "ready", |
| App: "kass", Account: "bab", Addr: "127.0.0.1:9301"}) |
| }, "apex squat", "titogo.bab.oncarlos.com"), |
| + "worker StartRunSweep": claimLogWorkerProbe("StartRunSweep", nil, "run sweep", "host-claim console"), |
| "worker LogPlacementDiscrepancies": claimLogWorkerProbe("LogPlacementDiscrepancies", claimRegionalApp, |
| msgPlacementDiscrepancy("drift.go.tito.io", "blog", "uk")), |
diff --git a/internal/console/claim_workers.go b/internal/console/claim_workers.go| index 01b781308..89bedaca3 100644 |
| --- a/internal/console/claim_workers.go |
| +++ b/internal/console/claim_workers.go |
| @@ -13,6 +13,7 @@ func (s *Server) registerClaimWorkers() { |
| "StartInfraPass": s.StartInfraPass, |
| "StartScanPass": s.StartScanPass, |
| "StartExpiryPass": s.StartExpiryPass, |
| + "StartRunSweep": s.StartRunSweep, |
| "StartRelay": s.StartRelay, |
| "StartMailEventPush": s.StartMailEventPush, |
| "StartDoorKeyResign": s.StartDoorKeyResign, |
diff --git a/internal/console/runsweep.go b/internal/console/runsweep.go| new file mode 100644 |
| index 000000000..1237255ff |
| --- /dev/null |
| +++ b/internal/console/runsweep.go |
| @@ -0,0 +1,243 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "log" |
| + "os" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/runs" |
| +) |
| + |
| +const runSweepInterval = 30 * time.Second |
| + |
| +// 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 |
| +// claim worker of its own, started from cmd/console like the others. |
| +func (s *Server) StartRunSweep(ctx context.Context) { |
| + if s.cfg.HostClaim { |
| + log.Printf("console: run sweep: off on a host-claim console, where carlos run is refused") |
| + return |
| + } |
| + t := time.NewTicker(runSweepInterval) |
| + defer t.Stop() |
| + for { |
| + s.runSweep(ctx, time.Now()) |
| + select { |
| + case <-ctx.Done(): |
| + return |
| + case <-t.C: |
| + } |
| + } |
| +} |
| + |
| +// runSweep 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) { |
| + if s.cfg.Releases == nil { |
| + return |
| + } |
| + recs, err := provision.ListAllRecords(ctx, s.cfg.Releases) |
| + if err != nil { |
| + log.Printf("console: run sweep: list records: %v", err) |
| + return |
| + } |
| + named := map[string]bool{} |
| + apps := map[[2]string]bool{} |
| + for _, rec := range recs { |
| + apps[[2]string{rec.Account, rec.App}] = true |
| + r := rec.ActiveRun |
| + if r == nil { |
| + continue |
| + } |
| + named[runs.Prefix(rec.Account, rec.App, r.ID)] = true |
| + acct, err := s.store.AccountBySqid(rec.Account) |
| + if err != nil || acct == nil { |
| + continue |
| + } |
| + c := runCall{acct: acct, app: rec.App, host: rec.Host, id: r.ID} |
| + switch { |
| + case r.State == provision.RunQueued && now.After(r.Deadline): |
| + 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(): |
| + s.sweepTerminalRun(ctx, c, r, now) |
| + } |
| + } |
| + strays := map[string]bool{} |
| + for app := range apps { |
| + s.sweepOrphanRuns(ctx, app[0], app[1], named, strays, now) |
| + } |
| + s.runSweepStrays = strays |
| +} |
| + |
| +// sweepExpireQueued expires a queued run past its pickup deadline. The |
| +// deadline is the record's, stamped at admission after the upload, never |
| +// request.json's requested_at. A box that picked it first wins the CAS, and |
| +// then nothing is written. The result.json and the run-end row are the ones |
| +// the expire route writes. |
| +func (s *Server) sweepExpireQueued(ctx context.Context, c runCall) { |
| + ctx, cancel := runWriteCtx(ctx) |
| + defer cancel() |
| + if _, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunExpire(c.id)); err != nil { |
| + if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| + log.Printf("console: run sweep: %s %s: expire: %v", c.host, c.id, err) |
| + } |
| + return |
| + } |
| + if err := s.endUnpicked(ctx, c, provision.RunExpired); err != nil { |
| + // The record says expired; the next sweep finds the missing result. |
| + log.Printf("console: run sweep: %s %s: end after expire: %v", c.host, c.id, err) |
| + } |
| +} |
| + |
| +// sweepLostRun ends a picked or cancelling run whose box has not reported an |
| +// hour past its deadline. The lost result is create-only, so a result the |
| +// box wrote first stands; the run-end row then says what the result says. |
| +func (s *Server) sweepLostRun(ctx context.Context, c runCall, r *provision.RunState, now time.Time) { |
| + ctx, cancel := runWriteCtx(ctx) |
| + defer cancel() |
| + rel := s.cfg.Releases |
| + // The ack refuses a last_seq that is not the result's, so a lost result |
| + // that guessed 0 would let the CLI acknowledge a run with chunks left. |
| + last, err := runs.LastChunkSeq(ctx, rel, c.acct.Sqid, c.app, c.id) |
| + if err != nil { |
| + log.Printf("console: run sweep: %s %s: list chunks: %v", c.host, c.id, err) |
| + return |
| + } |
| + lost := runs.Result{Interrupted: runs.InterruptedLost, Box: r.Box, LastSeq: last, Ended: now.UTC()} |
| + if err := runs.WriteResult(ctx, rel, c.acct.Sqid, c.app, c.id, lost); err != nil { |
| + log.Printf("console: run sweep: %s %s: lost result: %v", c.host, c.id, err) |
| + return |
| + } |
| + if _, err := provision.SetRecordRun(ctx, rel, c.acct.Sqid, c.app, c.host, provision.RunFinish(c.id)); err != nil { |
| + if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| + log.Printf("console: run sweep: %s %s: finish: %v", c.host, c.id, err) |
| + } |
| + return |
| + } |
| + 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 |
| + } |
| + outcome := "result" |
| + if res.Interrupted == runs.InterruptedLost { |
| + outcome = "lost" |
| + } |
| + s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, outcome, res) |
| +} |
| + |
| +// sweepTerminalRun clears a terminal run and deletes its objects once its |
| +// result is a day old, or, with no result (a run the record has as done), |
| +// once its deadline is. The record is cleared first and the objects deleted |
| +// only if that landed, as the ack does: a run the record still names keeps |
| +// its objects. A delete cut short leaves request.json, which the orphan pass |
| +// finishes from. |
| +func (s *Server) sweepTerminalRun(ctx context.Context, c runCall, r *provision.RunState, now time.Time) { |
| + ctx, cancel := runWriteCtx(ctx) |
| + defer cancel() |
| + rel := s.cfg.Releases |
| + res, err := runs.ReadResult(ctx, rel, c.acct.Sqid, c.app, c.id) |
| + if err != nil { |
| + return |
| + } |
| + ended := r.Deadline |
| + switch { |
| + case res != nil && !res.Ended.IsZero(): |
| + ended = res.Ended |
| + case res == nil && (r.State == provision.RunCancelled || r.State == provision.RunExpired): |
| + // The console that ended this run died before it wrote the result |
| + // and the run-end row. Write them now (both create-only); the day |
| + // starts from here. |
| + if _, err := runs.ReadRequest(ctx, rel, c.acct.Sqid, c.app, c.id); err == nil { |
| + if err := s.endUnpicked(ctx, c, r.State); err != nil { |
| + log.Printf("console: run sweep: %s %s: end %s run: %v", c.host, c.id, r.State, err) |
| + } |
| + return |
| + } |
| + } |
| + if now.Sub(ended) <= runs.KeepFor { |
| + return |
| + } |
| + if _, err := provision.SetRecordRun(ctx, rel, c.acct.Sqid, c.app, c.host, provision.RunClear(c.id)); err != nil { |
| + if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| + log.Printf("console: run sweep: %s %s: clear: %v", c.host, c.id, err) |
| + } |
| + return |
| + } |
| + if err := runs.DeleteRun(ctx, rel, c.acct.Sqid, c.app, c.id); err != nil { |
| + log.Printf("console: run sweep: %s %s: delete: %v", c.host, c.id, err) |
| + } |
| +} |
| + |
| +// sweepOrphanRuns deletes run directories no record names: acknowledged |
| +// runs whose delete was cut short, runs replaced by a later admission, |
| +// admissions that lost their CAS or whose outcome was unknown, runs of |
| +// deleted hosts. With a request.json the directory goes once the request is |
| +// 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) { |
| + rel := s.cfg.Releases |
| + dirs, err := rel.ListDirs(ctx, runs.RunsPrefix(acct, app)) |
| + if err != nil { |
| + return |
| + } |
| + for _, dir := range dirs { |
| + if named[dir] { |
| + continue |
| + } |
| + id := strings.TrimSuffix(strings.TrimPrefix(dir, runs.RunsPrefix(acct, app)), "/") |
| + if !runs.ValidID(id) { |
| + continue |
| + } |
| + req, err := runs.ReadRequest(ctx, rel, acct, app, id) |
| + switch { |
| + case err == nil: |
| + if now.Sub(req.RequestedAt) <= runs.KeepFor { |
| + continue |
| + } |
| + case errors.Is(err, os.ErrNotExist): |
| + if !s.strayIsOld(ctx, dir, strays, now) { |
| + continue |
| + } |
| + default: |
| + continue |
| + } |
| + if err := runs.DeleteRun(ctx, rel, acct, app, id); err != nil { |
| + log.Printf("console: run sweep: %s/%s %s: delete orphan: %v", acct, app, id, err) |
| + } |
| + } |
| +} |
| + |
| +// 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 |
| + } |
| + if s.runSweepStrays[dir] { |
| + return true |
| + } |
| + strays[dir] = true |
| + return false |
| +} |
diff --git a/internal/console/runsweep_test.go b/internal/console/runsweep_test.go| new file mode 100644 |
| index 000000000..8bbdf1bd6 |
| --- /dev/null |
| +++ b/internal/console/runsweep_test.go |
| @@ -0,0 +1,354 @@ |
| +package console |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/runs" |
| +) |
| + |
| +func (fx *runFx) record(t *testing.T) *provision.Record { |
| + t.Helper() |
| + rec, err := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "titogo", runTestHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return rec |
| +} |
| + |
| +// plainStore hides every optional extension of the store under it (ListMod |
| +// among them), as the test fakes and the fleet store door do. |
| +type plainStore struct{ release.Store } |
| + |
| +func (fx *runFx) putRunJSON(t *testing.T, key string, v any) { |
| + t.Helper() |
| + raw, _ := json.Marshal(v) |
| + if _, err := fx.rel.Put(context.Background(), key, raw, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func TestRunSweepExpiresAQueuedRunWhoseCallerLeft(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.startRun(t) |
| + fx.srv.runSweep(context.Background(), time.Now().Add(30*time.Second)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunQueued { |
| + t.Fatalf("expired before its deadline: %+v", r) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunExpired { |
| + t.Fatalf("not expired 60 s past the request (codex R3-3): %+v", r) |
| + } |
| + events, _ := fx.st.Events(OpsAccountID, "titogo", 20) |
| + found := false |
| + for _, e := range events { |
| + found = found || (e.Action == "run-end" && strings.Contains(e.Detail, id) && strings.Contains(e.Detail, `"expired"`)) |
| + } |
| + if !found { |
| + t.Error("no run-end expired row") |
| + } |
| + // The expire route's own result: a follower can read why it ended. |
| + res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id) |
| + if err != nil || res == nil || res.Refused != runs.RefusedExpired { |
| + t.Fatalf("expired result %+v, %v", res, err) |
| + } |
| + if fx.startRun(t) == "" { |
| + t.Fatal("an expired run blocked the next") |
| + } |
| +} |
| + |
| +// The deadline that counts is the record's, stamped at admission after the |
| +// upload. request.json's requested_at, which an upload can leave minutes |
| +// old, must not expire a run whose window has only just opened. |
| +func TestRunSweepExpiresByTheRecordsDeadlineNotTheRequests(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.startRun(t) |
| + req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "titogo", id) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + req.RequestedAt = time.Now().Add(-10 * time.Minute) |
| + raw, _ := json.Marshal(req) |
| + _, etag, err := fx.rel.Get(context.Background(), runs.RequestKey(release.OpsAccount, "titogo", id)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := fx.rel.Put(context.Background(), runs.RequestKey(release.OpsAccount, "titogo", id), raw, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(10*time.Second)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunQueued { |
| + t.Fatalf("expired by requested_at: %+v", r) |
| + } |
| +} |
| + |
| +func TestRunSweepExpiryLosesToALatePickup(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))) |
| + fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunPicked { |
| + t.Fatalf("the sweep overrode a pickup: %+v", r) |
| + } |
| + if res, _ := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id); res != nil { |
| + t.Fatalf("the losing sweep wrote a result: %+v", res) |
| + } |
| + if ends := fx.runEnds(t, id); len(ends) != 0 { |
| + t.Fatalf("the losing sweep wrote run-end rows: %q", ends) |
| + } |
| +} |
| + |
| +// The other order: the sweep's CAS lands first, and the box's pickup is the |
| +// one that changes nothing. |
| +func TestRunSweepExpiryBeatsALatePickupTheOtherWay(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.startRun(t) |
| + fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute)) |
| + _, err := provision.SetRecordRun(context.Background(), fx.rel, release.OpsAccount, "titogo", runTestHost, |
| + provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour))) |
| + if !errors.Is(err, provision.ErrRunConflict) { |
| + t.Fatalf("a pickup after the sweep expired the run: %v", err) |
| + } |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunExpired { |
| + t.Fatalf("%+v", r) |
| + } |
| +} |
| + |
| +func TestRunSweepMarksALostRunDone(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))) |
| + fx.chunk(t, id, 1, runs.StreamOut, "partial") |
| + fx.chunk(t, id, 2, runs.StreamErr, "more") |
| + fx.srv.runSweep(context.Background(), time.Now().Add(30*time.Minute)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunPicked { |
| + t.Fatalf("lost before an hour past its deadline: %+v", r) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunDone { |
| + t.Fatalf("lost run not done: %+v", r) |
| + } |
| + res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id) |
| + if err != nil || res == nil || res.Interrupted != runs.InterruptedLost { |
| + t.Fatalf("lost result %+v, %v", res, err) |
| + } |
| + if res.LastSeq != 2 || res.Box != "box-1" { |
| + t.Fatalf("lost result does not name the last chunk and box: %+v", res) |
| + } |
| + if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) { |
| + t.Fatalf("run-end rows %q", ends) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(3*time.Hour)) |
| + if ends := fx.runEnds(t, id); len(ends) != 1 { |
| + t.Fatalf("a second sweep wrote another row: %q", ends) |
| + } |
| +} |
| + |
| +func TestRunSweepMarksACancellingRunLost(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))) |
| + fx.transition(t, provision.RunCancelTransition(id)) |
| + fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour)) |
| + if r := fx.record(t).ActiveRun; r.State != provision.RunDone { |
| + t.Fatalf("a cancelling run whose box is gone is not done: %+v", r) |
| + } |
| +} |
| + |
| +// A box that wrote its own result and died before moving the run to done: |
| +// its result stands, and the row says result, not lost. |
| +func TestRunSweepLostRunKeepsABoxResultThatWonTheWrite(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))) |
| + own := runs.Result{Exit: 3, Started: time.Now().Add(-time.Hour), Ended: time.Now().Add(-59 * time.Minute)} |
| + if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, own); err != nil { |
| + t.Fatal(err) |
| + } |
| + 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) |
| + } |
| + res, _ := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id) |
| + if res == nil || res.Exit != 3 || res.Interrupted != "" { |
| + t.Fatalf("the box's result was overwritten: %+v", res) |
| + } |
| + if ends := fx.runEnds(t, id); len(ends) != 1 || strings.Contains(ends[0], `"lost"`) { |
| + t.Fatalf("run-end rows %q", ends) |
| + } |
| +} |
| + |
| +func TestRunSweepClearsTerminalRunsAfterADay(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{Ended: ended}); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.transition(t, provision.RunFinish(id)) |
| + fx.srv.runSweep(context.Background(), ended.Add(23*time.Hour)) |
| + if fx.record(t).ActiveRun == nil { |
| + t.Fatal("cleared before 24 hours") |
| + } |
| + fx.srv.runSweep(context.Background(), ended.Add(25*time.Hour)) |
| + if r := fx.record(t).ActiveRun; r != nil { |
| + t.Fatalf("not cleared after 24 hours: %+v", r) |
| + } |
| + if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id)); len(left) != 0 { |
| + t.Fatalf("objects left %q", left) |
| + } |
| +} |
| + |
| +// A cancelled or expired run is ended by the console, which wrote its result |
| +// at that moment: the day counts from there. |
| +func TestRunSweepClearsACancelledRunADayAfterItsResult(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.startRun(t) |
| + if status, body := fx.post(t, id, "cancel", ""); status != 200 { |
| + t.Fatalf("cancel: %d %s", status, body) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(23*time.Hour)) |
| + if fx.record(t).ActiveRun == nil { |
| + t.Fatal("cleared before 24 hours") |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(25*time.Hour)) |
| + if r := fx.record(t).ActiveRun; r != nil { |
| + t.Fatalf("not cleared after 24 hours: %+v", r) |
| + } |
| + if left := fx.left(id); len(left) != 0 { |
| + t.Fatalf("objects left %q", left) |
| + } |
| +} |
| + |
| +// A console that died between the record's CAS and the result and run-end |
| +// row leaves a terminal run with neither. The sweep writes them, once. |
| +func TestRunSweepFinishesAnExpiredRunThatHasNoResult(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.startRun(t) |
| + fx.transition(t, provision.RunExpire(id)) |
| + fx.srv.runSweep(context.Background(), time.Now()) |
| + res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id) |
| + if err != nil || res == nil || res.Refused != runs.RefusedExpired { |
| + t.Fatalf("result %+v, %v", res, err) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now()) |
| + if ends := fx.runEnds(t, id); len(ends) != 1 { |
| + t.Fatalf("run-end rows %q", ends) |
| + } |
| +} |
| + |
| +func TestRunSweepDeletesOrphanRunDirectories(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) |
| + young := old |
| + young.ID, young.RequestedAt = "run_dddddddddddddddddddddddddd", time.Now() |
| + fx.putRunJSON(t, runs.RequestKey(young.Account, young.App, young.ID), young) |
| + fx.srv.runSweep(ctx, time.Now()) |
| + if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) != 0 { |
| + t.Errorf("a day-old orphan survived: %q", left) |
| + } |
| + if left, _ := fx.rel.List(ctx, runs.Prefix(young.Account, young.App, young.ID)); len(left) == 0 { |
| + t.Error("a fresh orphan (a create in flight) was deleted") |
| + } |
| +} |
| + |
| +// The run the record names is never an orphan, however old its request. |
| +func TestRunSweepKeepsTheRunTheRecordNames(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.startRun(t) |
| + fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(100*time.Hour))) |
| + fx.srv.runSweep(context.Background(), time.Now().Add(48*time.Hour)) |
| + if left := fx.left(id); len(left) == 0 { |
| + t.Fatal("the picked run's objects were deleted") |
| + } |
| +} |
| + |
| +// A directory with no request.json (a stray result.json a stale retry can |
| +// leave, or an audited marker left by a cut-short delete) is reaped once |
| +// its newest object is a day old, when the store lists modification times. |
| +func TestRunSweepReapsAStrayRunDirectoryByItsNewestObject(t *testing.T) { |
| + fx := runFixture(t) |
| + ctx := context.Background() |
| + id := "run_eeeeeeeeeeeeeeeeeeeeeeeeee" |
| + if _, err := fx.rel.Put(ctx, runs.ResultKey(release.OpsAccount, "titogo", id), []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.runSweep(ctx, time.Now()) |
| + fx.srv.runSweep(ctx, time.Now()) |
| + if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 { |
| + t.Fatal("a stray written a moment ago was deleted") |
| + } |
| + fx.srv.runSweep(ctx, time.Now().Add(25*time.Hour)) |
| + if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "titogo", id)); len(left) != 0 { |
| + t.Fatalf("a day-old stray survived: %q", left) |
| + } |
| +} |
| + |
| +// Without modification times the stray goes on the second sweep that sees it. |
| +func TestRunSweepReapsAStrayRunDirectoryOnTheSecondSweepWithoutModTimes(t *testing.T) { |
| + fx := runFixture(t) |
| + ctx := context.Background() |
| + fx.srv.cfg.Releases = plainStore{fx.rel} |
| + if _, ok := fx.srv.cfg.Releases.(release.ModLister); ok { |
| + t.Fatal("the wrapper still lists modification times") |
| + } |
| + id := "run_ffffffffffffffffffffffffff" |
| + 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") |
| + } |
| + 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) |
| + } |
| +} |
| + |
| +// A stray that is gone and comes back is a new stray: the count starts again. |
| +func TestRunSweepForgetsAStrayThatWentAway(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 { |
| + t.Fatal(err) |
| + } |
| + fx.srv.runSweep(ctx, time.Now()) |
| + if err := fx.rel.Delete(ctx, key); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.runSweep(ctx, time.Now()) |
| + if _, err := fx.rel.Put(ctx, key, []byte(`{}`), ""); 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) |
| + } |
| +} |
| + |
| +func TestRunSweepLeavesAnUnreadableRequestAlone(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 { |
| + 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") |
| + } |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index a0827041b..64d47049b 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -827,6 +827,7 @@ 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 8313a1e89..55830c730 100644 |
| --- a/internal/console/testdata/claim-authority-sites.txt |
| +++ b/internal/console/testdata/claim-authority-sites.txt |
| @@ -83,6 +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() |
| 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) |
diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt| index c0a5abf5e..af2a76894 100644 |
| --- a/internal/console/testdata/claim-routes.txt |
| +++ b/internal/console/testdata/claim-routes.txt |
| @@ -258,4 +258,5 @@ worker StartHostReports |
| worker StartInfraPass |
| worker StartMailEventPush |
| worker StartRelay |
| +worker StartRunSweep |
| worker StartScanPass |