| 1 | package schedule |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "sync" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/hostwrite" |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | ) |
| 17 | |
| 18 | // Converger turns apps/<acct>/<app>/schedules.json into timers rows for every |
| 19 | // exec-backed instance this box serves, acts on schedule-run.json, and prunes |
| 20 | // what no longer belongs (spec §6.2). One instance per agent; Pass is called |
| 21 | // on every serial move, hourly, and once at start. |
| 22 | // |
| 23 | // Etag memory is what keeps an idle box cheap and a run-signal |
| 24 | // exactly-once: an unchanged declaration is one GET and no writes; a run |
| 25 | // signal fires when its etag MOVES, and the first sighting only seeds |
| 26 | // (a signal older than this process is history — restart-gen's rule). |
| 27 | // |
| 28 | // One instance, but THREE callers, and they are separate goroutines: the |
| 29 | // agent's fast converge loop runs Pass at arming and on every serial move, |
| 30 | // and its adopt loop runs Pass hourly. Pass is therefore serialized on mu — |
| 31 | // see the field. |
| 32 | type Converger struct { |
| 33 | Reg *registry.Registry |
| 34 | Store release.Store |
| 35 | Now func() time.Time |
| 36 | Box string // reported in Status.Box |
| 37 | Report func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports |
| 38 | Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box |
| 39 | HostWrites hostwrite.Config |
| 40 | // Frozen answers whether a move holds host's claim (moves spec §9.4, |
| 41 | // §19 K9): its timers are neither upserted nor pruned. Nil: none is. |
| 42 | Frozen func(host string) bool |
| 43 | // Gate is the per-host gate shared with the runner and a move's drain; |
| 44 | // convergeHost holds it across its Frozen check and its writes. Nil: |
| 45 | // no gate, main's converger. |
| 46 | Gate *HostGate |
| 47 | |
| 48 | // afterFrozenCheck is the tests' pause inside the gate, after the |
| 49 | // Frozen check (TestConvergerCannotWriteAfterTheBarrier). Nil in the |
| 50 | // agent. |
| 51 | afterFrozenCheck func(host string) |
| 52 | |
| 53 | declEtag map[[2]string]string // (acct, app) → last etag seen |
| 54 | declDoc map[[2]string]Declaration |
| 55 | runEtag map[[2]string]string |
| 56 | |
| 57 | // reportedUnsupported tracks which unit-backed hosts have already had |
| 58 | // their one report — Report fires for them once, not every pass. |
| 59 | reportedUnsupported map[string]bool |
| 60 | |
| 61 | // mu serializes Pass, and is held for the WHOLE of it. Two reasons, and |
| 62 | // the first one alone is fatal: |
| 63 | // |
| 64 | // 1. The four maps above are plain maps written on every pass. The agent |
| 65 | // calls Pass from two goroutines (fastConvergeLoop's seed and serial |
| 66 | // leg; adoptLoop's hourly floor), and they genuinely overlap — an |
| 67 | // hourly pass mid-flight when a console bump fires the fast one. A |
| 68 | // concurrent map read and write is a runtime THROW, not a data race |
| 69 | // you find later: it would kill the agent, and the activator and every |
| 70 | // exec child with it. |
| 71 | // 2. The etag memory IS the exactly-once mechanism for a manual run. Two |
| 72 | // interleaved passes can both read runEtag[k] before either writes it, |
| 73 | // both see movement, and both fire the same `carlos schedule run` — |
| 74 | // the one promise this struct's comment makes, broken. |
| 75 | // |
| 76 | // The lock lives HERE rather than at the agent's call sites (the shape |
| 77 | // configSyncMu/reconcileMu use in host/agent.go) because a Converger is a |
| 78 | // stateful object with an exported method: a future caller cannot forget a |
| 79 | // lock it never has to take. |
| 80 | mu sync.Mutex |
| 81 | } |
| 82 | |
| 83 | const onceRetention = 7 * 24 * time.Hour |
| 84 | |
| 85 | func NewConverger(reg *registry.Registry, store release.Store, box string, report func(registry.Route, Status)) *Converger { |
| 86 | return &Converger{ |
| 87 | Reg: reg, Store: store, Now: time.Now, Box: box, Report: report, |
| 88 | declEtag: map[[2]string]string{}, declDoc: map[[2]string]Declaration{}, |
| 89 | runEtag: map[[2]string]string{}, reportedUnsupported: map[string]bool{}, |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | // Pass is die-free per app: one app's unreadable declaration is logged and |
| 94 | // its rows left exactly as they were (a baseline with holes manufactures |
| 95 | // deletions), and every other app still converges. |
| 96 | // |
| 97 | // Serialized end to end (see mu): concurrent callers queue rather than |
| 98 | // interleave. A pass is store-bound, so a queued caller can wait on a slow |
| 99 | // bucket — that is the intended cost, and cheaper than the alternative for |
| 100 | // both reasons mu documents. |
| 101 | func (c *Converger) Pass(ctx context.Context) error { |
| 102 | c.mu.Lock() |
| 103 | defer c.mu.Unlock() |
| 104 | routes, err := c.Reg.List() |
| 105 | if err != nil { |
| 106 | return err |
| 107 | } |
| 108 | now := c.Now() |
| 109 | // alive is every row, held ones included, so PruneTimers never treats a |
| 110 | // frozen host's rows as orphans. frozenAny holds back the table-wide |
| 111 | // prune itself while a move has frozen any host here: its retention leg |
| 112 | // would otherwise delete a frozen host's finished once-rows. The |
| 113 | // runner's daily sweep still runs (it cannot be frozen; the timers |
| 114 | // digest leaves that history out, joint N14), so nothing accumulates. |
| 115 | alive := map[string]bool{} |
| 116 | frozenAny := false |
| 117 | byApp := map[[2]string][]registry.Route{} |
| 118 | for _, rt := range routes { |
| 119 | alive[rt.Host] = true |
| 120 | if c.Frozen != nil && !frozenAny && c.Frozen(rt.Host) { |
| 121 | frozenAny = true |
| 122 | } |
| 123 | if rt.Kind != "instance" { |
| 124 | continue |
| 125 | } |
| 126 | k := [2]string{release.RouteAccount(rt), rt.App} |
| 127 | if k[1] == "" { // no k[0] check: RouteAccount's floor is release.OpsAccount, never "" |
| 128 | continue |
| 129 | } |
| 130 | byApp[k] = append(byApp[k], rt) |
| 131 | } |
| 132 | var errs []error |
| 133 | for k, rts := range byApp { |
| 134 | if err := c.convergeApp(ctx, k[0], k[1], rts, now); err != nil { |
| 135 | errs = append(errs, fmt.Errorf("%s/%s: %w", k[0], k[1], err)) |
| 136 | } |
| 137 | } |
| 138 | if !c.HostWrites.Enabled && !frozenAny { |
| 139 | if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil { |
| 140 | errs = append(errs, err) |
| 141 | } else if n > 0 { |
| 142 | log.Printf("schedule: pruned %d timer row(s)", n) |
| 143 | } |
| 144 | } |
| 145 | return errors.Join(errs...) |
| 146 | } |
| 147 | |
| 148 | func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []registry.Route, now time.Time) error { |
| 149 | // Capture once before either grouped remote input read. Reuse on every effect. |
| 150 | selected := map[string]hostwrite.Work{} |
| 151 | var errs []error |
| 152 | for _, rt := range rts { |
| 153 | work, err := c.HostWrites.Select(ctx, rt.Host, "timers") |
| 154 | if err != nil { |
| 155 | errs = append(errs, fmt.Errorf("%s: select lifecycle authority: %w", rt.Host, err)) |
| 156 | continue |
| 157 | } |
| 158 | selected[rt.Host] = work |
| 159 | } |
| 160 | k := [2]string{acct, app} |
| 161 | // Reads the object directly rather than calling ReadDeclaration: the |
| 162 | // etag has to be compared against declEtag[k] BEFORE deciding whether |
| 163 | // to unmarshal, so an unchanged declaration costs one GET and zero JSON |
| 164 | // parses (the struct's own doc comment's promise). ReadDeclaration |
| 165 | // always unmarshals what it reads — right for a one-off caller, wrong |
| 166 | // for a loop that runs every 2s per app. |
| 167 | raw, etag, err := c.Store.Get(ctx, DeclarationKey(acct, app)) |
| 168 | var decl Declaration |
| 169 | switch { |
| 170 | case errors.Is(err, os.ErrNotExist): |
| 171 | etag = "" |
| 172 | case err != nil: |
| 173 | return fmt.Errorf("read schedules.json: %w", err) |
| 174 | case etag == c.declEtag[k]: |
| 175 | decl = c.declDoc[k] |
| 176 | default: |
| 177 | if err := json.Unmarshal(raw, &decl); err != nil { |
| 178 | return fmt.Errorf("parse schedules.json: %w", err) |
| 179 | } |
| 180 | } |
| 181 | c.declEtag[k], c.declDoc[k] = etag, decl |
| 182 | |
| 183 | // Run signal: seed silently on first sight, fire on movement. |
| 184 | runRaw, runEtag, err := c.Store.Get(ctx, RunKey(acct, app)) |
| 185 | fireName := "" |
| 186 | switch { |
| 187 | case errors.Is(err, os.ErrNotExist): |
| 188 | runEtag = "" |
| 189 | case err != nil: |
| 190 | return fmt.Errorf("read schedule-run.json: %w", err) |
| 191 | } |
| 192 | if prev, seen := c.runEtag[k]; seen && runEtag != prev && runEtag != "" { |
| 193 | var sig RunSignal |
| 194 | if err := json.Unmarshal(runRaw, &sig); err == nil { |
| 195 | if _, ok := decl.Find(sig.Name); ok { |
| 196 | fireName = sig.Name |
| 197 | } else { |
| 198 | log.Printf("schedule: %s/%s: run requested for undeclared schedule %q — ignored", acct, app, sig.Name) |
| 199 | } |
| 200 | } |
| 201 | } |
| 202 | c.runEtag[k] = runEtag |
| 203 | |
| 204 | for _, rt := range rts { |
| 205 | work, ok := selected[rt.Host] |
| 206 | if !ok { |
| 207 | continue |
| 208 | } |
| 209 | hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work}) |
| 210 | changed, err := c.convergeHost(hctx, rt, decl, fireName, now) |
| 211 | if err != nil { |
| 212 | errs = append(errs, fmt.Errorf("%s: %w", rt.Host, err)) |
| 213 | continue |
| 214 | } |
| 215 | if changed && c.Report != nil { |
| 216 | var st Status |
| 217 | err := withTimerWork(hctx, func() error { |
| 218 | var statusErr error |
| 219 | st, statusErr = StatusFor(c.Reg, rt, c.Box, now, c.Idle) |
| 220 | return statusErr |
| 221 | }) |
| 222 | if err != nil { |
| 223 | errs = append(errs, err) |
| 224 | continue |
| 225 | } |
| 226 | c.Report(rt, st) |
| 227 | } |
| 228 | } |
| 229 | return errors.Join(errs...) |
| 230 | } |
| 231 | |
| 232 | // convergeHost returns whether anything about the host's rows changed (or |
| 233 | // the host is unsupported and should be reported so). |
| 234 | // |
| 235 | // A host a move has frozen (Frozen, moves spec §9.4 and §19 K9) is left |
| 236 | // exactly as it is: nothing upserted, nothing pruned, a run signal not |
| 237 | // fired. The check and every write below happen inside the host's gate, so |
| 238 | // the cut's drain (HostGate.Drain after the hold) waits for a convergence |
| 239 | // that checked before the freeze, and one that starts after the drain sees |
| 240 | // it. A host the gate refuses is being drained: skipped the same way. |
| 241 | func (c *Converger) convergeHost(ctx context.Context, rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) { |
| 242 | if !c.Gate.tryEnter(rt.Host) { |
| 243 | return false, nil |
| 244 | } |
| 245 | defer c.Gate.leave(rt.Host) |
| 246 | if c.Frozen != nil && c.Frozen(rt.Host) { |
| 247 | return false, nil |
| 248 | } |
| 249 | if c.afterFrozenCheck != nil { |
| 250 | c.afterFrozenCheck(rt.Host) |
| 251 | } |
| 252 | var before []registry.Timer |
| 253 | if c.HostWrites.Enabled { |
| 254 | err := withTimerWork(ctx, func() error { |
| 255 | var readErr error |
| 256 | before, readErr = c.Reg.TimersFor(rt.Host) |
| 257 | return readErr |
| 258 | }) |
| 259 | if err != nil { |
| 260 | return false, err |
| 261 | } |
| 262 | cutoff := now.Add(-onceRetention).Unix() |
| 263 | kept := before[:0] |
| 264 | for _, row := range before { |
| 265 | if row.Kind == KindOnce && row.NextAt == 0 && row.LastAt < cutoff { |
| 266 | if err := deleteSelectedTimer(ctx, c.Reg, row); err != nil { |
| 267 | return false, err |
| 268 | } |
| 269 | continue |
| 270 | } |
| 271 | kept = append(kept, row) |
| 272 | } |
| 273 | before = kept |
| 274 | } |
| 275 | if rt.Backing == registry.BackingUnit { |
| 276 | // A host can flip exec -> unit mid-lifetime (the app-capabilities |
| 277 | // migration door, registry.SetBacking): drop any schedule rows it |
| 278 | // still carries from before the flip, so "no rows" stays true for |
| 279 | // every unit-backed host, not just ones that were always unit-backed. |
| 280 | var n int |
| 281 | n, err := c.deleteSchedules(ctx, rt.Host, before, map[string]bool{}) |
| 282 | if err != nil { |
| 283 | return false, err |
| 284 | } |
| 285 | changed := n > 0 |
| 286 | // Reported once per host, and only when there is something to warn |
| 287 | // about — an undeclared app on a unit-backed host is silent. |
| 288 | if len(decl.Schedules) > 0 && c.unsupportedUnreported(rt.Host) { |
| 289 | changed = true |
| 290 | } |
| 291 | return changed, nil |
| 292 | } |
| 293 | if !c.HostWrites.Enabled { |
| 294 | var err error |
| 295 | before, err = c.Reg.TimersFor(rt.Host) |
| 296 | if err != nil { |
| 297 | return false, err |
| 298 | } |
| 299 | } |
| 300 | registrations := map[string]string{} |
| 301 | keep := map[string]bool{} |
| 302 | changed := false |
| 303 | for _, sc := range decl.Schedules { |
| 304 | sp, err := sc.Spec() |
| 305 | if err != nil { |
| 306 | log.Printf("schedule: %s: schedule %q has an invalid spec (%v) — skipped", rt.Host, sc.Name, err) |
| 307 | continue |
| 308 | } |
| 309 | keep[sc.Name] = true |
| 310 | // prev is the ONLY-schedule-kind baseline: a same-named once-row |
| 311 | // (the instance's own, off its control socket) is never "previous |
| 312 | // state" for this comparison — a kind collision decides itself |
| 313 | // below, at the write, not here. |
| 314 | prev := findTimer(before, sc.Name, KindSchedule) |
| 315 | wouldChange := prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path |
| 316 | var next int64 |
| 317 | neverFires := false |
| 318 | if wouldChange { |
| 319 | // Schedule.Validate refuses a spec whose first Next is zero at |
| 320 | // `set` time; one written some other way must not become a row |
| 321 | // due at the zero time's unix value (-62135596800 — in the past, |
| 322 | // so due on the very next tick, forever). It is parked at |
| 323 | // next_at=0 and told why, below. |
| 324 | if n := sp.Next(now); !n.IsZero() { |
| 325 | next = n.Unix() |
| 326 | } else { |
| 327 | neverFires = true |
| 328 | } |
| 329 | } |
| 330 | // A new occurrence is due whenever next moves (DueAt rides NextAt — |
| 331 | // UpsertTimer leaves both alone when NextAt is 0 and nothing changed). |
| 332 | var installed *registry.Timer |
| 333 | t := registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, DueAt: next, CreatedAt: now.Unix()} |
| 334 | handled := false |
| 335 | if c.HostWrites.Enabled { |
| 336 | expected := "" |
| 337 | if prev != nil { |
| 338 | expected = prev.Registration |
| 339 | } |
| 340 | if prev != nil && !wouldChange && hasTimerMutator(ctx) { |
| 341 | installed, handled = prev, true |
| 342 | } else { |
| 343 | installed, handled, err = mutateSelectedTimer(ctx, t, expected, false) |
| 344 | } |
| 345 | } |
| 346 | if !handled { |
| 347 | err = withTimerWork(ctx, func() error { |
| 348 | if !c.HostWrites.Enabled { |
| 349 | return c.Reg.UpsertTimer(t) |
| 350 | } |
| 351 | expected := "" |
| 352 | if prev != nil { |
| 353 | expected = prev.Registration |
| 354 | } |
| 355 | var err error |
| 356 | installed, err = c.Reg.UpsertTimerRegistration(t, expected) |
| 357 | return err |
| 358 | }) |
| 359 | } |
| 360 | |
| 361 | if errors.Is(err, registry.ErrTimerKind) { |
| 362 | // Nothing was written — a collision must not pin `changed`, |
| 363 | // or a permanently-colliding name reports every single pass. |
| 364 | log.Printf("schedule: %s: %q is a once-timer the instance registered; the declared schedule of that name is skipped", rt.Host, sc.Name) |
| 365 | continue |
| 366 | } |
| 367 | if err != nil { |
| 368 | return changed, err |
| 369 | } |
| 370 | if installed != nil { |
| 371 | registrations[sc.Name] = installed.Registration |
| 372 | } |
| 373 | if neverFires { |
| 374 | log.Printf("schedule: %s: schedule %q never fires (no occurrence within the search window) — parked", rt.Host, sc.Name) |
| 375 | if err := withTimerWork(ctx, func() error { |
| 376 | if c.HostWrites.Enabled { |
| 377 | return c.Reg.SetTimerStatusRegistration(rt.Host, sc.Name, installed.Registration, StatusNeverFires) |
| 378 | } |
| 379 | return c.Reg.SetTimerStatus(rt.Host, sc.Name, StatusNeverFires) |
| 380 | }); err != nil { |
| 381 | return changed, err |
| 382 | } |
| 383 | } |
| 384 | if wouldChange { |
| 385 | changed = true |
| 386 | } |
| 387 | } |
| 388 | n, err := c.deleteSchedules(ctx, rt.Host, before, keep) |
| 389 | if err != nil { |
| 390 | return changed, err |
| 391 | } |
| 392 | if n > 0 { |
| 393 | changed = true |
| 394 | } |
| 395 | if fireName != "" && keep[fireName] { |
| 396 | fire := func() error { |
| 397 | if c.HostWrites.Enabled { |
| 398 | row, err := c.Reg.GetTimer(rt.Host, fireName) |
| 399 | if err != nil { |
| 400 | return err |
| 401 | } |
| 402 | if row.Registration != registrations[fireName] { |
| 403 | return registry.ErrTimerChanged |
| 404 | } |
| 405 | row.NextAt, row.DueAt, row.Attempts = now.Unix(), now.Unix(), 0 |
| 406 | expected := row.Registration |
| 407 | row.Registration = "" |
| 408 | if _, handled, err := mutateSelectedTimer(ctx, *row, expected, false); handled { |
| 409 | return err |
| 410 | } |
| 411 | } |
| 412 | return withTimerWork(ctx, func() error { |
| 413 | if c.HostWrites.Enabled { |
| 414 | return c.Reg.FireTimerNowRegistration(rt.Host, fireName, registrations[fireName], now.Unix()) |
| 415 | } |
| 416 | return c.Reg.FireTimerNow(rt.Host, fireName, now.Unix()) |
| 417 | }) |
| 418 | } |
| 419 | if err := fire(); err != nil { |
| 420 | return changed, err |
| 421 | } |
| 422 | changed = true |
| 423 | } |
| 424 | return changed, nil |
| 425 | } |
| 426 | |
| 427 | // unsupportedUnreported reports whether host has NOT yet had its |
| 428 | // unsupported-host report sent, and marks it sent — so the report is |
| 429 | // written once per host's lifetime in this Converger, not every pass. |
| 430 | func (c *Converger) unsupportedUnreported(host string) bool { |
| 431 | if c.reportedUnsupported == nil { |
| 432 | c.reportedUnsupported = map[string]bool{} |
| 433 | } |
| 434 | if c.reportedUnsupported[host] { |
| 435 | return false |
| 436 | } |
| 437 | c.reportedUnsupported[host] = true |
| 438 | return true |
| 439 | } |
| 440 | |
| 441 | // findTimer returns the row named name with the given kind, or nil. Kind |
| 442 | // matters: a once-row and a declared schedule can share a name (the |
| 443 | // once-row is the instance's own, registered off its control socket), and |
| 444 | // only a same-kind row is ever "previous state" for the change comparison |
| 445 | // below — a kind collision is a skip (ErrTimerKind, at the UpsertTimer |
| 446 | // call), never a manufactured "changed". |
| 447 | func findTimer(rows []registry.Timer, name, kind string) *registry.Timer { |
| 448 | for i := range rows { |
| 449 | if rows[i].Name == name && rows[i].Kind == kind { |
| 450 | return &rows[i] |
| 451 | } |
| 452 | } |
| 453 | return nil |
| 454 | } |
| 455 | |
| 456 | const UnsupportedUnit = "unit-backed instance — scheduled work delivers to exec-backed instances in this release" |
| 457 | |
| 458 | // StatusNeverFires is the last_status of a parked row: a declared cron |
| 459 | // expression with no occurrence inside CronExpr.Next's window (Validate |
| 460 | // refuses these at `set`; this is the defence for one that slipped past it). |
| 461 | const StatusNeverFires = "never fires" |
| 462 | |
| 463 | // StatusFor builds the host's report from its rows. idle is the activator's |
| 464 | // idle window on this box, threaded through to holdsAwake — zero when no |
| 465 | // hibernation is armed on the box at all. |
| 466 | // |
| 467 | // Hibernation is also armed PER ROUTE (rt.Hibernate — activator.go will not |
| 468 | // sweep a host unless Hibernate && Status == "ready"), so a box-wide window |
| 469 | // is not sufficient on its own: a route with Hibernate false was never |
| 470 | // going to sleep, schedule or no schedule, and holds_awake must say so by |
| 471 | // reporting false for every one of its rows, not by naming a tight timer |
| 472 | // that changes nothing for it. |
| 473 | func StatusFor(reg *registry.Registry, rt registry.Route, box string, now time.Time, idle time.Duration) (Status, error) { |
| 474 | st := Status{Schema: 1, Host: rt.Host, Box: box, ReportedAt: now.UTC().Format(time.RFC3339), Timers: []TimerStatus{}} |
| 475 | if rt.Backing == registry.BackingUnit { |
| 476 | st.Unsupported = UnsupportedUnit |
| 477 | return st, nil |
| 478 | } |
| 479 | if !rt.Hibernate { |
| 480 | idle = 0 |
| 481 | } |
| 482 | rows, err := reg.TimersFor(rt.Host) |
| 483 | if err != nil { |
| 484 | return st, err |
| 485 | } |
| 486 | for _, r := range rows { |
| 487 | st.Timers = append(st.Timers, toStatus(r, idle, now)) |
| 488 | } |
| 489 | return st, nil |
| 490 | } |
| 491 | |
| 492 | // toStatus maps one timers row onto its wire shape. Task 5 (the `carlos |
| 493 | // schedule ls` reader / the runner's own report path) reuses this so the |
| 494 | // row-to-status mapping is written once. |
| 495 | func toStatus(t registry.Timer, idle time.Duration, now time.Time) TimerStatus { |
| 496 | return TimerStatus{ |
| 497 | Name: t.Name, Kind: t.Kind, Spec: t.Spec, Path: t.Path, |
| 498 | NextAt: t.NextAt, DueAt: t.DueAt, LastAt: t.LastAt, |
| 499 | LastStatus: t.LastStatus, LastMs: t.LastMs, Attempts: t.Attempts, |
| 500 | HoldsAwake: holdsAwake(t, idle, now), |
| 501 | } |
| 502 | } |
| 503 | |
| 504 | // holdsAwake reports whether t's interval is tight enough that its instance |
| 505 | // never gets to sleep: the gap between consecutive occurrences is no wider |
| 506 | // than idle, so the activator's guard (and, before it, the tick's own wake) |
| 507 | // keeps the process resident indefinitely. |
| 508 | // |
| 509 | // The gap comes from two Next calls rather than Spec.Every, so a cron gets a |
| 510 | // real answer instead of a special case — but for an irregular cron (e.g. |
| 511 | // "0,5 * * * *": five minutes then fifty-five) that first gap is a SAMPLE, |
| 512 | // not a bound: the same row can answer true at :59 and false at :01, |
| 513 | // depending on when the report happens to be built. That is the direction |
| 514 | // to err in — over-warning about a bill beats missing one — and is chosen |
| 515 | // deliberately, not an oversight. |
| 516 | // |
| 517 | // idle <= 0 means hibernation is not armed — either the box has no |
| 518 | // activator, or (StatusFor) the route itself has Hibernate off — so |
| 519 | // nothing holds anything awake, because nothing sleeps. A once-timer fires |
| 520 | // once and cannot hold anything. An unparseable spec is not evidence of a |
| 521 | // problem, so it makes no claim. |
| 522 | func holdsAwake(t registry.Timer, idle time.Duration, now time.Time) bool { |
| 523 | if idle <= 0 || t.Kind != KindSchedule { |
| 524 | return false |
| 525 | } |
| 526 | sp, err := ParseSpec(t.Spec) |
| 527 | if err != nil { |
| 528 | return false |
| 529 | } |
| 530 | first := sp.Next(now) |
| 531 | if first.IsZero() { |
| 532 | return false |
| 533 | } |
| 534 | second := sp.Next(first) |
| 535 | if second.IsZero() { |
| 536 | return false |
| 537 | } |
| 538 | return second.Sub(first) <= idle |
| 539 | } |
| 540 | |
| 541 | // deleteSchedules only removes rows observed by this pass. A same-name |
| 542 | // replacement keeps its registration and survives the stale deletion. |
| 543 | func (c *Converger) deleteSchedules(ctx context.Context, host string, before []registry.Timer, keep map[string]bool) (int, error) { |
| 544 | if !c.HostWrites.Enabled { |
| 545 | return c.Reg.DeleteTimersExcept(host, KindSchedule, keep) |
| 546 | } |
| 547 | if err := withTimerWork(ctx, func() error { return nil }); err != nil { |
| 548 | return 0, err |
| 549 | } |
| 550 | n := 0 |
| 551 | for _, row := range before { |
| 552 | if row.Kind != KindSchedule || keep[row.Name] { |
| 553 | continue |
| 554 | } |
| 555 | if err := deleteSelectedTimer(ctx, c.Reg, row); err != nil { |
| 556 | return n, err |
| 557 | } |
| 558 | n++ |
| 559 | } |
| 560 | return n, nil |
| 561 | } |
| 562 | |
amadan