| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log" |
| 7 | "os" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/provision" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | "github.com/carlosframework/platform/internal/runs" |
| 14 | ) |
| 15 | |
| 16 | const ( |
| 17 | runSweepInterval = 30 * time.Second |
| 18 | // runSweepHousekeepingEvery: expiry and lost-run closing run on every |
| 19 | // pass, because a caller waits on them; retention and orphans have 24 |
| 20 | // hour thresholds, so they run on every 20th pass (about 10 minutes). |
| 21 | runSweepHousekeepingEvery = 20 |
| 22 | // orphanCandidateWindow: with the record view armed, orphan |
| 23 | // housekeeping lists runs/ only for apps with a run-start row this |
| 24 | // recent, the window in which a run's objects can still be live or |
| 25 | // awaiting cleanup (S3 change-index design §4.6, codex R2-1). |
| 26 | orphanCandidateWindow = 48 * time.Hour |
| 27 | // orphanFullSweepEvery: and once this often it lists every app with |
| 28 | // records, as it always did, as the backstop. |
| 29 | orphanFullSweepEvery = 24 * time.Hour |
| 30 | ) |
| 31 | |
| 32 | // runSweepTimeout bounds one pass, so a hung list or read cannot stall the |
| 33 | // sweep for good. A variable so a test can shorten it. |
| 34 | var runSweepTimeout = 5 * time.Minute |
| 35 | |
| 36 | // StartRunSweep runs the run sweep every 30 seconds (run spec §5.8). The |
| 37 | // console has no deadline-sweep loop to join (index item 1), so this is a |
| 38 | // claim worker of its own, started from cmd/console like the others. |
| 39 | func (s *Server) StartRunSweep(ctx context.Context) { |
| 40 | if s.cfg.HostClaim { |
| 41 | log.Printf("console: run sweep: off on a host-claim console, where carlos run is refused") |
| 42 | return |
| 43 | } |
| 44 | t := time.NewTicker(runSweepInterval) |
| 45 | defer t.Stop() |
| 46 | for pass := 0; ; pass++ { |
| 47 | s.runSweepPass(ctx, time.Now(), pass%runSweepHousekeepingEvery == 0) |
| 48 | select { |
| 49 | case <-ctx.Done(): |
| 50 | return |
| 51 | case <-t.C: |
| 52 | } |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | // runSweep is one full pass: every phase. |
| 57 | func (s *Server) runSweep(ctx context.Context, now time.Time) { s.runSweepPass(ctx, now, true) } |
| 58 | |
| 59 | // runSweepPass is one pass. Every transition is a CAS on the record, so a box |
| 60 | // or a CLI that acts first wins and the sweep changes nothing: a queued run |
| 61 | // a box picked in the meantime stays picked, a run a box finished stays |
| 62 | // finished. housekeeping adds the terminal-retention and orphan phases to the |
| 63 | // expiry and lost-run phases every pass runs. It is called from one goroutine. |
| 64 | // |
| 65 | // The records are read tolerantly: one account or record that will not read |
| 66 | // must not stop expiry and lost-run closing for the rest of the fleet, and |
| 67 | // each of those acts on a record it read. The orphan phase is the exception: |
| 68 | // a run directory no record names is deleted, so it runs only on a read that |
| 69 | // saw every record. |
| 70 | // |
| 71 | // With the record view armed the records are read from it (recordWalkStore); |
| 72 | // every write, and the orphan phase's listing of runs/, stays on the bucket. |
| 73 | func (s *Server) runSweepPass(ctx context.Context, now time.Time, housekeeping bool) { |
| 74 | if s.cfg.Releases == nil { |
| 75 | return |
| 76 | } |
| 77 | ctx, cancel := context.WithTimeout(ctx, runSweepTimeout) |
| 78 | defer cancel() |
| 79 | src, _ := s.recordWalkStore(ctx, s.cfg.Releases) |
| 80 | recs, complete, err := provision.ListAllRecordsTolerant(ctx, src) |
| 81 | if err != nil { |
| 82 | log.Printf("console: run sweep: list records: %v", err) |
| 83 | return |
| 84 | } |
| 85 | if !complete { |
| 86 | log.Printf("console: run sweep: some records could not be read; their runs wait for a later pass, and no orphan run is deleted on this one") |
| 87 | } |
| 88 | named := map[string]bool{} |
| 89 | apps := map[[2]string]bool{} |
| 90 | for _, rec := range recs { |
| 91 | if ctx.Err() != nil { |
| 92 | log.Printf("console: run sweep: pass cut short: %v", ctx.Err()) |
| 93 | return |
| 94 | } |
| 95 | apps[[2]string{rec.Account, rec.App}] = true |
| 96 | r := rec.ActiveRun |
| 97 | if r == nil { |
| 98 | continue |
| 99 | } |
| 100 | named[runs.Prefix(rec.Account, rec.App, r.ID)] = true |
| 101 | acct, err := s.store.AccountBySqid(rec.Account) |
| 102 | if err != nil { |
| 103 | log.Printf("console: run sweep: account %s: %v", rec.Account, err) |
| 104 | continue |
| 105 | } |
| 106 | if acct == nil { |
| 107 | continue |
| 108 | } |
| 109 | c := runCall{acct: acct, app: rec.App, host: rec.Host, id: r.ID} |
| 110 | switch { |
| 111 | case r.State == provision.RunQueued && now.After(r.Deadline): |
| 112 | s.sweepExpireQueued(ctx, c) |
| 113 | case (r.State == provision.RunPicked || r.State == provision.RunCancel) && now.After(r.Deadline.Add(runs.LostAfter)): |
| 114 | s.sweepLostRun(ctx, c, r, now) |
| 115 | case !r.Open() && housekeeping: |
| 116 | s.sweepTerminalRun(ctx, c, r, now) |
| 117 | } |
| 118 | } |
| 119 | if !housekeeping || !complete { |
| 120 | return |
| 121 | } |
| 122 | candidates, full := s.orphanCandidates(apps, now) |
| 123 | for app := range candidates { |
| 124 | if ctx.Err() != nil { |
| 125 | log.Printf("console: run sweep: pass cut short: %v", ctx.Err()) |
| 126 | return |
| 127 | } |
| 128 | s.sweepOrphanRuns(ctx, app[0], app[1], named, now) |
| 129 | } |
| 130 | if full && s.recordView != nil { |
| 131 | s.lastOrphanFull = now |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // orphanCandidates is the apps whose runs/ the orphan phase lists, out of |
| 136 | // apps, every app with records, and whether that is all of them. Disarmed it |
| 137 | // is all of them, as before the index. Armed (S3 change-index design §4.6, |
| 138 | // codex R2-1) it is only those with a run-start row in the last |
| 139 | // orphanCandidateWindow, except on the first housekeeping after start and |
| 140 | // then once every orphanFullSweepEvery, which take every app, as does a |
| 141 | // failed read of the rows. Orphans are still found by age, not by index |
| 142 | // movement: runs/ is outside every scope. The narrowed set is always a |
| 143 | // subset of apps, so it never lists an app the full sweep would not. |
| 144 | func (s *Server) orphanCandidates(apps map[[2]string]bool, now time.Time) (map[[2]string]bool, bool) { |
| 145 | if s.recordView == nil || s.lastOrphanFull.IsZero() || now.Sub(s.lastOrphanFull) >= orphanFullSweepEvery { |
| 146 | return apps, true |
| 147 | } |
| 148 | recent, err := s.store.AppsWithActionSince("run-start", now.Add(-orphanCandidateWindow)) |
| 149 | if err != nil { |
| 150 | log.Printf("console: run sweep: recent runs: %v (every app's runs are swept this pass)", err) |
| 151 | return apps, true |
| 152 | } |
| 153 | out := map[[2]string]bool{} |
| 154 | for _, r := range recent { |
| 155 | if app := [2]string{r.Sqid, r.App}; apps[app] { |
| 156 | out[app] = true |
| 157 | } |
| 158 | } |
| 159 | return out, false |
| 160 | } |
| 161 | |
| 162 | // sweepExpireQueued expires a queued run past its pickup deadline. The |
| 163 | // deadline is the record's, stamped at admission after the upload, never |
| 164 | // request.json's requested_at. A box that picked it first wins the CAS, and |
| 165 | // then nothing is written. The result.json and the run-end row are the ones |
| 166 | // the expire route writes. |
| 167 | func (s *Server) sweepExpireQueued(ctx context.Context, c runCall) { |
| 168 | ctx, cancel := runWriteCtx(ctx) |
| 169 | defer cancel() |
| 170 | if _, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunExpire(c.id)); err != nil { |
| 171 | if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 172 | log.Printf("console: run sweep: %s %s: expire: %v", c.host, c.id, err) |
| 173 | } |
| 174 | return |
| 175 | } |
| 176 | if err := s.endUnpicked(ctx, c, provision.RunExpired); err != nil { |
| 177 | // The record says expired; the next sweep finds the missing result. |
| 178 | log.Printf("console: run sweep: %s %s: end after expire: %v", c.host, c.id, err) |
| 179 | } |
| 180 | } |
| 181 | |
| 182 | // sweepLostRun ends a picked or cancelling run whose box has not reported an |
| 183 | // hour past its deadline. The lost result is create-only, so a result the |
| 184 | // box wrote first stands; the run-end row then says what the result says. |
| 185 | func (s *Server) sweepLostRun(ctx context.Context, c runCall, r *provision.RunState, now time.Time) { |
| 186 | ctx, cancel := runWriteCtx(ctx) |
| 187 | defer cancel() |
| 188 | rel := s.cfg.Releases |
| 189 | // The ack refuses a last_seq that is not the result's, so a lost result |
| 190 | // that guessed 0 would let the CLI acknowledge a run with chunks left. |
| 191 | last, err := runs.LastChunkSeq(ctx, rel, c.acct.Sqid, c.app, c.id) |
| 192 | if err != nil { |
| 193 | log.Printf("console: run sweep: %s %s: list chunks: %v", c.host, c.id, err) |
| 194 | return |
| 195 | } |
| 196 | lost := runs.Result{Interrupted: runs.InterruptedLost, Box: r.Box, LastSeq: last, Ended: now.UTC()} |
| 197 | if err := runs.WriteResult(ctx, rel, c.acct.Sqid, c.app, c.id, lost); err != nil { |
| 198 | log.Printf("console: run sweep: %s %s: lost result: %v", c.host, c.id, err) |
| 199 | return |
| 200 | } |
| 201 | if _, err := provision.SetRecordRun(ctx, rel, c.acct.Sqid, c.app, c.host, provision.RunFinish(c.id)); err != nil { |
| 202 | if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 203 | log.Printf("console: run sweep: %s %s: finish: %v", c.host, c.id, err) |
| 204 | } |
| 205 | return |
| 206 | } |
| 207 | res, err := runs.ReadResult(ctx, rel, c.acct.Sqid, c.app, c.id) |
| 208 | if err != nil || res == nil { |
| 209 | return // the run is done; the terminal phase writes the row |
| 210 | } |
| 211 | s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res) |
| 212 | } |
| 213 | |
| 214 | // sweepTerminalRun clears a terminal run and deletes its objects once its |
| 215 | // result is a day old, or, with no result (a run the record has as done), |
| 216 | // once its deadline is. The record is cleared first and the objects deleted |
| 217 | // only if that landed, as the ack does: a run the record still names keeps |
| 218 | // its objects. A delete cut short leaves request.json, which the orphan pass |
| 219 | // finishes from. |
| 220 | func (s *Server) sweepTerminalRun(ctx context.Context, c runCall, r *provision.RunState, now time.Time) { |
| 221 | ctx, cancel := runWriteCtx(ctx) |
| 222 | defer cancel() |
| 223 | rel := s.cfg.Releases |
| 224 | res, err := runs.ReadResult(ctx, rel, c.acct.Sqid, c.app, c.id) |
| 225 | if err != nil { |
| 226 | return |
| 227 | } |
| 228 | ended := r.Deadline |
| 229 | switch { |
| 230 | case res != nil && !res.Ended.IsZero(): |
| 231 | ended = res.Ended |
| 232 | case res == nil && (r.State == provision.RunCancelled || r.State == provision.RunExpired): |
| 233 | // The console that ended this run died before it wrote the result |
| 234 | // and the run-end row. Write them now (both create-only); the day |
| 235 | // starts from here. |
| 236 | if _, err := runs.ReadRequest(ctx, rel, c.acct.Sqid, c.app, c.id); err == nil { |
| 237 | if err := s.endUnpicked(ctx, c, r.State); err != nil { |
| 238 | log.Printf("console: run sweep: %s %s: end %s run: %v", c.host, c.id, r.State, err) |
| 239 | } |
| 240 | return |
| 241 | } |
| 242 | } |
| 243 | if res != nil { |
| 244 | // The run-end row for any terminal run with a result, before anything |
| 245 | // is cleared: a run nobody followed or acked (a box's own finish), or |
| 246 | // one whose sweep died after the done CAS, has no row yet. The marker |
| 247 | // makes this a no-op when the row exists. |
| 248 | s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res) |
| 249 | } |
| 250 | if now.Sub(ended) <= runs.KeepFor { |
| 251 | return |
| 252 | } |
| 253 | if _, err := provision.SetRecordRun(ctx, rel, c.acct.Sqid, c.app, c.host, provision.RunClear(c.id)); err != nil { |
| 254 | if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 255 | log.Printf("console: run sweep: %s %s: clear: %v", c.host, c.id, err) |
| 256 | } |
| 257 | return |
| 258 | } |
| 259 | if err := runs.DeleteRun(ctx, rel, c.acct.Sqid, c.app, c.id); err != nil { |
| 260 | log.Printf("console: run sweep: %s %s: delete: %v", c.host, c.id, err) |
| 261 | } |
| 262 | } |
| 263 | |
| 264 | // sweepOrphanRuns deletes run directories no record names: acknowledged |
| 265 | // runs whose delete was cut short, runs replaced by a later admission, |
| 266 | // admissions that lost their CAS or whose outcome was unknown, runs of |
| 267 | // deleted hosts. With a request.json the directory goes once the request is |
| 268 | // a day old, after its run-end row (orphanRunEnd). Without one it is a stray |
| 269 | // (request.json is written first and deleted last but for the audited |
| 270 | // marker, so it is neither a create in flight nor a run anyone can read): it |
| 271 | // goes once its newest object is a day old, and stays when the store cannot |
| 272 | // say. |
| 273 | func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named map[string]bool, now time.Time) { |
| 274 | rel := s.cfg.Releases |
| 275 | dirs, err := rel.ListDirs(ctx, runs.RunsPrefix(acct, app)) |
| 276 | if err != nil { |
| 277 | return |
| 278 | } |
| 279 | var account *Account |
| 280 | accountRead := false |
| 281 | for _, dir := range dirs { |
| 282 | if named[dir] { |
| 283 | continue |
| 284 | } |
| 285 | id := strings.TrimSuffix(strings.TrimPrefix(dir, runs.RunsPrefix(acct, app)), "/") |
| 286 | if !runs.ValidID(id) { |
| 287 | continue |
| 288 | } |
| 289 | req, err := runs.ReadRequest(ctx, rel, acct, app, id) |
| 290 | switch { |
| 291 | case err == nil: |
| 292 | if now.Sub(req.RequestedAt) <= runs.KeepFor { |
| 293 | continue |
| 294 | } |
| 295 | if !accountRead { |
| 296 | if account, err = s.store.AccountBySqid(acct); err != nil { |
| 297 | log.Printf("console: run sweep: account %s: %v", acct, err) |
| 298 | return // no run of it is deleted before its row can be written |
| 299 | } |
| 300 | accountRead = true |
| 301 | } |
| 302 | if account != nil && !s.orphanRunEnd(ctx, account, app, req.Host, id, now) { |
| 303 | continue |
| 304 | } |
| 305 | case errors.Is(err, os.ErrNotExist): |
| 306 | if !s.strayIsOld(ctx, dir, now) { |
| 307 | continue |
| 308 | } |
| 309 | default: |
| 310 | continue |
| 311 | } |
| 312 | if err := runs.DeleteRun(ctx, rel, acct, app, id); err != nil { |
| 313 | log.Printf("console: run sweep: %s/%s %s: delete orphan: %v", acct, app, id, err) |
| 314 | } |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | // orphanRunEnd writes an orphan run's run-end row before its objects go |
| 319 | // (run spec §6). A run that ended with no follower and was replaced by a |
| 320 | // later admission before a housekeeping pass, or whose record let it go, |
| 321 | // has none yet. With a result the row says what it says; without one the |
| 322 | // run never came back, so it gets a lost result (create-only: a result the |
| 323 | // box writes first stands) and the row says that. The audited marker makes |
| 324 | // this a no-op for a run that has its row. It reports whether the run may |
| 325 | // be deleted. |
| 326 | func (s *Server) orphanRunEnd(ctx context.Context, acct *Account, app, host, id string, now time.Time) bool { |
| 327 | rel := s.cfg.Releases |
| 328 | res, err := runs.ReadResult(ctx, rel, acct.Sqid, app, id) |
| 329 | if err != nil { |
| 330 | log.Printf("console: run sweep: %s/%s %s: read orphan result: %v", acct.Sqid, app, id, err) |
| 331 | return false |
| 332 | } |
| 333 | if res == nil { |
| 334 | last, err := runs.LastChunkSeq(ctx, rel, acct.Sqid, app, id) |
| 335 | if err != nil { |
| 336 | log.Printf("console: run sweep: %s/%s %s: list orphan chunks: %v", acct.Sqid, app, id, err) |
| 337 | return false |
| 338 | } |
| 339 | lost := runs.Result{Interrupted: runs.InterruptedLost, LastSeq: last, Ended: now.UTC()} |
| 340 | if err := runs.WriteResult(ctx, rel, acct.Sqid, app, id, lost); err != nil { |
| 341 | log.Printf("console: run sweep: %s/%s %s: orphan lost result: %v", acct.Sqid, app, id, err) |
| 342 | return false |
| 343 | } |
| 344 | if res, err = runs.ReadResult(ctx, rel, acct.Sqid, app, id); err != nil || res == nil { |
| 345 | log.Printf("console: run sweep: %s/%s %s: read orphan result back: %v", acct.Sqid, app, id, err) |
| 346 | return false |
| 347 | } |
| 348 | } |
| 349 | s.auditRunEnd(ctx, acct, app, host, id, runEndOutcome(res), res) |
| 350 | // The marker stands only once the row has had its chance; a marker the |
| 351 | // bucket refused leaves the run for the next pass. |
| 352 | if _, _, err := rel.Get(ctx, runs.AuditedKey(acct.Sqid, app, id)); err != nil { |
| 353 | log.Printf("console: run sweep: %s/%s %s: no audit marker yet, delete left for a later pass: %v", acct.Sqid, app, id, err) |
| 354 | return false |
| 355 | } |
| 356 | return true |
| 357 | } |
| 358 | |
| 359 | // strayIsOld reports whether a run directory with no request.json has had |
| 360 | // nothing written to it for a day. A store that cannot list modification |
| 361 | // times (both real stores can) leaves strays alone. |
| 362 | func (s *Server) strayIsOld(ctx context.Context, dir string, now time.Time) bool { |
| 363 | ml, ok := s.cfg.Releases.(release.ModLister) |
| 364 | if !ok { |
| 365 | return false |
| 366 | } |
| 367 | objs, err := ml.ListMod(ctx, dir) |
| 368 | if err != nil || len(objs) == 0 { |
| 369 | return false |
| 370 | } |
| 371 | newest := objs[0].Mod |
| 372 | for _, o := range objs[1:] { |
| 373 | if o.Mod.After(newest) { |
| 374 | newest = o.Mod |
| 375 | } |
| 376 | } |
| 377 | return now.Sub(newest) > runs.KeepFor |
| 378 | } |
| 379 | |
amadan