| 1 | package schedule |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "net/http" |
| 9 | "strconv" |
| 10 | "sync" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/hostwrite" |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | ) |
| 16 | |
| 17 | const ( |
| 18 | DeliveryTimeout = 30 * time.Minute |
| 19 | TickEvery = 15 * time.Second |
| 20 | MaxAttempts = 5 |
| 21 | MaxParallelHosts = 4 |
| 22 | dueBatch = 64 |
| 23 | backoffBase = time.Minute |
| 24 | backoffCap = 15 * time.Minute |
| 25 | // pruneEvery is the prune leg's cadence — see Runner.prune. Daily |
| 26 | // because the thing it deletes is a 7-day-old finished row: a sweep an |
| 27 | // hour earlier or later changes nothing a member can see. |
| 28 | pruneEvery = 24 * time.Hour |
| 29 | ) |
| 30 | |
| 31 | // ErrTimeout is what a Deliverer returns when the handler ran past the |
| 32 | // delivery timeout. The policy treats it as delivered (spec §6.3): the job |
| 33 | // may have finished, or still be running — a retry would double it. |
| 34 | var ErrTimeout = errors.New("schedule: delivery timed out") |
| 35 | |
| 36 | // Deliverer sends one tick and reports the handler's status, or an error. |
| 37 | // |
| 38 | // Status 0 with an error means NO RESPONSE — the tick never got an HTTP |
| 39 | // status line back. On the wake path that is a wake the activator refused |
| 40 | // ("instance is waking", "failed to wake"), on any path a delivery that ran |
| 41 | // past its deadline, and either way a handler that returned having written |
| 42 | // nothing. It does NOT mean the instance is unreachable in any lasting |
| 43 | // sense: the commonest cause is a wake still in progress, and the next |
| 44 | // attempt usually lands. Nor is it every failure — a socket that will not |
| 45 | // dial comes back as the proxy's own 502, a status like any other. |
| 46 | type Deliverer interface { |
| 47 | Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) |
| 48 | } |
| 49 | |
| 50 | // Runner fires due timers: at-least-once, serialized per host, bounded |
| 51 | // parallelism across hosts (spec §6.3). Every seam is a field so a tick is |
| 52 | // testable with no sockets or clocks. |
| 53 | type Runner struct { |
| 54 | StartDelivery DeliveryAdmission |
| 55 | Reg *registry.Registry |
| 56 | Deliver Deliverer |
| 57 | Now func() time.Time |
| 58 | Timeout time.Duration |
| 59 | Token func(host string) (string, error) |
| 60 | Event func(rt registry.Route, level, msg string) |
| 61 | Report func(rt registry.Route, st Status) |
| 62 | Box string |
| 63 | Build string |
| 64 | Idle time.Duration // the activator's idle window; zero when no hibernation is armed on this box |
| 65 | HostWrites hostwrite.Config |
| 66 | // Gate is the per-host gate shared with the converger and a move's |
| 67 | // drain (HostGate): claim enters it for the whole delivery. Nil: no |
| 68 | // gate, main's runner. Independent of StartDelivery: a held row is |
| 69 | // skipped before claim and fire, and the delivery admission is asked |
| 70 | // inside fire for the rows that get that far, so a tick delivers only |
| 71 | // when the hold and the admission both allow it. |
| 72 | Gate *HostGate |
| 73 | // Skip names a host this box must not touch at all, whatever the hold |
| 74 | // wiring says: its due timers are neither delivered nor written, no |
| 75 | // token is minted for it and nothing is reported. The agent sets it for |
| 76 | // a held row and for a host whose offline move's record names another |
| 77 | // region (harness Run 9, defect 8: after open, the source's tick minted |
| 78 | // a fresh <host>.env in the instance directory the move had parked). |
| 79 | // Asked at the tick's route read and again just before the token, so a |
| 80 | // hold written in between is seen. Nil: none is skipped. |
| 81 | Skip func(rt registry.Route) bool |
| 82 | |
| 83 | // beforeClaim is the tests' pause between a tick's route read and its |
| 84 | // claim (TestDrainRaceCannotAdvanceBookkeeping). Nil in the agent. |
| 85 | beforeClaim func(host string) |
| 86 | |
| 87 | // Per-host work is DETACHED from the tick: Tick launches one goroutine |
| 88 | // per host with due rows and returns without waiting for any of them. |
| 89 | // A handler is allowed to run for up to DeliveryTimeout (30m), and a |
| 90 | // tick that waited on it would hold every other host's due timers |
| 91 | // hostage for that long. busy is what keeps "one delivery at a time per |
| 92 | // host" true across ticks: a host whose goroutine is still running is |
| 93 | // skipped, and its rows (still due — nothing has been recorded yet) are |
| 94 | // picked up by the first tick after it finishes. sem bounds hosts in |
| 95 | // flight across ALL ticks, not per tick. inflight is for tests, which |
| 96 | // need "everything this tick launched has finished" — see wait. |
| 97 | mu sync.Mutex |
| 98 | busy map[string]bool |
| 99 | sem chan struct{} |
| 100 | inflight sync.WaitGroup |
| 101 | |
| 102 | // lastPruned is the prune leg's stamp — see prune. |
| 103 | lastPruned time.Time |
| 104 | } |
| 105 | |
| 106 | // Start runs Tick every TickEvery until ctx ends. |
| 107 | func (r *Runner) Start(ctx context.Context) { |
| 108 | go func() { |
| 109 | t := time.NewTicker(TickEvery) |
| 110 | defer t.Stop() |
| 111 | for { |
| 112 | select { |
| 113 | case <-ctx.Done(): |
| 114 | return |
| 115 | case <-t.C: |
| 116 | r.Tick(ctx) |
| 117 | } |
| 118 | } |
| 119 | }() |
| 120 | } |
| 121 | |
| 122 | func (r *Runner) now() time.Time { |
| 123 | if r.Now != nil { |
| 124 | return r.Now() |
| 125 | } |
| 126 | return time.Now() |
| 127 | } |
| 128 | |
| 129 | func (r *Runner) timeout() time.Duration { |
| 130 | if r.Timeout > 0 { |
| 131 | return r.Timeout |
| 132 | } |
| 133 | return DeliveryTimeout |
| 134 | } |
| 135 | |
| 136 | // Tick is one pass: everything due, grouped by host; each host not already |
| 137 | // in flight gets a goroutine (at most MaxParallelHosts running across all |
| 138 | // ticks), timers within a host one after another. Tick returns once every |
| 139 | // goroutine is launched — it never waits for a delivery. |
| 140 | func (r *Runner) Tick(ctx context.Context) { |
| 141 | now := r.now() |
| 142 | r.prune(ctx, now) |
| 143 | // Busy hosts are excluded IN THE QUERY, not after it. The batch is |
| 144 | // capped at dueBatch rows: a host parked in a long handler (or one |
| 145 | // holding more due rows than the batch) would otherwise fill the batch |
| 146 | // with rows this tick is about to skip, and another host's due timer |
| 147 | // would never be seen. The skip below stays as belt-and-braces — a host |
| 148 | // can go busy between the query and the claim. |
| 149 | // |
| 150 | // With the hold wiring armed, held hosts are excluded in the query for |
| 151 | // the same reason: a held row is skipped below with no bookkeeping, so |
| 152 | // a held host's rows would sit at the front of every tick's batch and |
| 153 | // starve the rest. Off, the query is main's. |
| 154 | // |
| 155 | // Skip set (the agent's scheduleSkip) is the same rule for the hosts it |
| 156 | // names, before the limit too (Astra, moves-offline-fixes review 1): a |
| 157 | // held host's rows are left out by the query, and a host Skip names on |
| 158 | // other grounds is added to the exclusion and the batch read again, so |
| 159 | // a skipped host's due rows can never fill the batch. |
| 160 | due, err := r.dueRows(now) |
| 161 | if err != nil { |
| 162 | log.Printf("schedule: due query: %v", err) |
| 163 | return |
| 164 | } |
| 165 | if len(due) == 0 { |
| 166 | return |
| 167 | } |
| 168 | byHost := map[string][]registry.Timer{} |
| 169 | var order []string |
| 170 | for _, t := range due { |
| 171 | if _, seen := byHost[t.Host]; !seen { |
| 172 | order = append(order, t.Host) |
| 173 | } |
| 174 | byHost[t.Host] = append(byHost[t.Host], t) |
| 175 | } |
| 176 | for _, host := range order { |
| 177 | work, err := r.HostWrites.Select(ctx, host, "timers") |
| 178 | if err != nil { |
| 179 | log.Printf("schedule: %s: select lifecycle authority: %v", host, err) |
| 180 | continue |
| 181 | } |
| 182 | hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work}) |
| 183 | var rt *registry.Route |
| 184 | err = withTimerWork(hctx, func() error { |
| 185 | var readErr error |
| 186 | rt, readErr = r.Reg.Get(host) |
| 187 | return readErr |
| 188 | }) |
| 189 | if err != nil { |
| 190 | // A registry error is not "route gone": say so, or a broken |
| 191 | // registry reads as a fleet of retired routes. |
| 192 | log.Printf("schedule: %s: registry read: %v — skipping this tick", host, err) |
| 193 | continue |
| 194 | } |
| 195 | if rt == nil { |
| 196 | // Converge prunes these; until it does, they are not ours to fire. |
| 197 | continue |
| 198 | } |
| 199 | if r.Skip != nil && r.Skip(*rt) { |
| 200 | continue // not this box's host to touch: no bookkeeping either |
| 201 | } |
| 202 | // A held row (moves spec §5.2, §9.4 "frozen") is skipped with no |
| 203 | // bookkeeping at all: before this, its delivery was refused by the |
| 204 | // activator and recorded as a retry, so a held row's attempts |
| 205 | // advanced while its instance was moving. This read is only the |
| 206 | // cheap pre-filter; the decision is the read inside the claim below. |
| 207 | // Only with the hold wiring armed: off, the row is main's. |
| 208 | if r.holdsArmed() && registry.IsHold(rt.Status) { |
| 209 | continue |
| 210 | } |
| 211 | // An uncommitted import's row gets no ticks (seeded-instances spec |
| 212 | // §5.4): only verify requests reach it until commit. Whether or not |
| 213 | // the hold wiring is armed. |
| 214 | if rt.ImportGate != "" { |
| 215 | continue |
| 216 | } |
| 217 | if r.beforeClaim != nil { |
| 218 | r.beforeClaim(host) |
| 219 | } |
| 220 | if !r.claim(host) { |
| 221 | continue // a previous tick's delivery to this host is still running, or a move is draining it |
| 222 | } |
| 223 | if r.holdsArmed() { |
| 224 | // The decision read, inside the claim. A move holds the row and |
| 225 | // then drains the host (HostGate.Drain), and the drain is sound |
| 226 | // only if every tick that looks at the row after the hold does |
| 227 | // so while it holds the claim: a tick that read the row ready |
| 228 | // above and was descheduled before claim must not slip past the |
| 229 | // drain and record an attempt. |
| 230 | err = withTimerWork(hctx, func() error { |
| 231 | var readErr error |
| 232 | rt, readErr = r.Reg.Get(host) |
| 233 | return readErr |
| 234 | }) |
| 235 | // The import gate is part of the decision too: a swap that |
| 236 | // landed after the pre-filter read gets no tick. |
| 237 | if err != nil || rt == nil || registry.IsHold(rt.Status) || rt.ImportGate != "" { |
| 238 | if err != nil { |
| 239 | log.Printf("schedule: %s: registry read inside the claim: %v. skipping this tick", host, err) |
| 240 | } |
| 241 | r.release(host) |
| 242 | continue |
| 243 | } |
| 244 | } |
| 245 | // The slot is taken HERE, before the goroutine exists, so the bound |
| 246 | // is on hosts in flight and never on goroutines waiting to be. Hosts |
| 247 | // are in due order, so when every slot is busy the earliest-due host |
| 248 | // is the one waiting for the next free one. |
| 249 | select { |
| 250 | case r.slots() <- struct{}{}: |
| 251 | case <-ctx.Done(): |
| 252 | r.release(host) |
| 253 | return |
| 254 | } |
| 255 | r.inflight.Add(1) |
| 256 | go func(ctx context.Context, host string, rt registry.Route, timers []registry.Timer) { |
| 257 | defer r.inflight.Done() |
| 258 | defer func() { <-r.slots() }() |
| 259 | defer r.release(host) |
| 260 | for i, t := range timers { |
| 261 | // A hold written while this host's earlier timers were |
| 262 | // delivering stops the rest: each later timer re-reads the |
| 263 | // row first. The drain waits for this goroutine either way. |
| 264 | // Only with the hold wiring armed. |
| 265 | if i > 0 && r.holdsArmed() && r.held(ctx, host) { |
| 266 | break |
| 267 | } |
| 268 | r.fire(ctx, rt, t) |
| 269 | } |
| 270 | if r.Report != nil { |
| 271 | // A report that cannot be built is a silent hole in `carlos |
| 272 | // schedule ls` — the member's only view of this box. Say so: |
| 273 | // the fires themselves still landed on the rows. |
| 274 | var st Status |
| 275 | err := withTimerWork(ctx, func() error { |
| 276 | var statusErr error |
| 277 | st, statusErr = StatusFor(r.Reg, rt, r.Box, r.now(), r.Idle) |
| 278 | return statusErr |
| 279 | }) |
| 280 | if err != nil { |
| 281 | log.Printf("schedule: %s: build status report: %v", rt.Host, err) |
| 282 | } else { |
| 283 | r.Report(rt, st) |
| 284 | } |
| 285 | } |
| 286 | }(hctx, host, *rt, byHost[host]) |
| 287 | } |
| 288 | } |
| 289 | |
| 290 | // dueRows is Tick's due query: busy hosts excluded, held hosts too when the |
| 291 | // hold wiring is armed or Skip is set, and, with Skip set, every host it |
| 292 | // names, read again with each one found added to the exclusion. Each read |
| 293 | // either has no skipped host or adds at least one host to the exclusion, so |
| 294 | // the loop ends within one read per host with due rows. A host whose row |
| 295 | // cannot be read is left in the batch: the tick's own read reports it. |
| 296 | func (r *Runner) dueRows(now time.Time) ([]registry.Timer, error) { |
| 297 | exclude := r.busyHosts() |
| 298 | for { |
| 299 | var due []registry.Timer |
| 300 | var err error |
| 301 | if r.holdsArmed() || r.Skip != nil { |
| 302 | due, err = r.Reg.DueUnheldTimers(now.Unix(), dueBatch, exclude) |
| 303 | } else { |
| 304 | due, err = r.Reg.DueTimers(now.Unix(), dueBatch, exclude) |
| 305 | } |
| 306 | if err != nil || r.Skip == nil || len(due) == 0 { |
| 307 | return due, err |
| 308 | } |
| 309 | var skipped []string |
| 310 | seen := map[string]bool{} |
| 311 | for _, t := range due { |
| 312 | if seen[t.Host] { |
| 313 | continue |
| 314 | } |
| 315 | seen[t.Host] = true |
| 316 | if rt, err := r.Reg.Get(t.Host); err == nil && rt != nil && r.Skip(*rt) { |
| 317 | skipped = append(skipped, t.Host) |
| 318 | } |
| 319 | } |
| 320 | if len(skipped) == 0 { |
| 321 | return due, nil |
| 322 | } |
| 323 | exclude = append(exclude, skipped...) |
| 324 | } |
| 325 | } |
| 326 | |
| 327 | // busyHosts is the in-flight set as DueTimers wants it. A snapshot: a host |
| 328 | // that goes busy after this read is caught by claim instead. |
| 329 | func (r *Runner) busyHosts() []string { |
| 330 | r.mu.Lock() |
| 331 | defer r.mu.Unlock() |
| 332 | if len(r.busy) == 0 { |
| 333 | return nil |
| 334 | } |
| 335 | hosts := make([]string, 0, len(r.busy)) |
| 336 | for h := range r.busy { |
| 337 | hosts = append(hosts, h) |
| 338 | } |
| 339 | return hosts |
| 340 | } |
| 341 | |
| 342 | // prune is the runner's OWN sweep of dead timer rows, and it exists because |
| 343 | // the converger's identical sweep is not always there: the converger needs |
| 344 | // the deploy store, and a box with the activator armed and no store still |
| 345 | // accumulates rows — once-timers registered by instances on their control |
| 346 | // socket (spec §7.2). Without this leg those finished rows would sit in the |
| 347 | // table forever and could pin an app against the 1000-per-host cap on the |
| 348 | // one kind of box that has no other way to shed them. |
| 349 | // |
| 350 | // Daily, stamped rather than scheduled: it rides the 15s tick so there is no |
| 351 | // second goroutine to stop, and the first tick after start prunes once. |
| 352 | // Cheap either way (one DELETE over an indexed table), and idempotent with |
| 353 | // the converger's sweep on a box that has both. |
| 354 | func (r *Runner) prune(ctx context.Context, now time.Time) { |
| 355 | r.mu.Lock() |
| 356 | if !r.lastPruned.IsZero() && now.Sub(r.lastPruned) < pruneEvery { |
| 357 | r.mu.Unlock() |
| 358 | return |
| 359 | } |
| 360 | // Stamped before the work, not after: a registry that cannot be read is |
| 361 | // not worth retrying every 15s, and the rows a failed sweep would have |
| 362 | // dropped are inert until it next runs. |
| 363 | r.lastPruned = now |
| 364 | r.mu.Unlock() |
| 365 | routes, err := r.Reg.List() |
| 366 | if err != nil { |
| 367 | log.Printf("schedule: runner prune: list routes: %v — leaving the rows alone", err) |
| 368 | return |
| 369 | } |
| 370 | sockets := map[string]bool{} |
| 371 | alive := map[string]bool{} |
| 372 | for _, rt := range routes { |
| 373 | alive[rt.Host] = true |
| 374 | if rt.Socket != "" { |
| 375 | sockets[rt.Socket] = true |
| 376 | } |
| 377 | } |
| 378 | // The sweep is also the delivery leg's eviction point: a route that goes |
| 379 | // away stops receiving deliveries, so nothing it caches for that route's |
| 380 | // socket would ever error and be dropped on its own. Optional — a |
| 381 | // Deliverer that caches nothing (every test's, and any future one) simply |
| 382 | // does not implement it. |
| 383 | if sr, ok := r.Deliver.(SocketRetainer); ok { |
| 384 | sr.RetainSockets(sockets) |
| 385 | } |
| 386 | if len(routes) == 0 { |
| 387 | // PruneTimers reads an empty alive set as "every host is gone" and |
| 388 | // deletes the whole table — correct for a genuinely empty registry |
| 389 | // (routes and timers live in the same file, so every row really is |
| 390 | // orphaned), but it is a wipe reachable from two callers now, and a |
| 391 | // caller that ever passed an empty set by mistake would take the |
| 392 | // table with it. This leg declines it explicitly: rows on a box with |
| 393 | // no routes are inert, and the next sweep after a route appears is |
| 394 | // the one that can tell orphans from arrivals. |
| 395 | log.Printf("schedule: runner prune: the registry has no routes — declining to sweep (that would delete every timer row)") |
| 396 | return |
| 397 | } |
| 398 | if !r.HostWrites.Enabled { |
| 399 | if n, err := r.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil { |
| 400 | log.Printf("schedule: runner prune: %v", err) |
| 401 | } else if n > 0 { |
| 402 | log.Printf("schedule: runner prune: pruned %d timer row(s)", n) |
| 403 | } |
| 404 | return |
| 405 | } |
| 406 | cutoff := now.Add(-onceRetention).Unix() |
| 407 | for _, rt := range routes { |
| 408 | work, err := r.HostWrites.Select(ctx, rt.Host, "timers") |
| 409 | if err != nil { |
| 410 | log.Printf("schedule: runner prune %s: select lifecycle authority: %v", rt.Host, err) |
| 411 | continue |
| 412 | } |
| 413 | hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work}) |
| 414 | var rows []registry.Timer |
| 415 | if err := withTimerWork(hctx, func() error { |
| 416 | var readErr error |
| 417 | rows, readErr = r.Reg.TimersFor(rt.Host) |
| 418 | return readErr |
| 419 | }); err != nil { |
| 420 | log.Printf("schedule: runner prune %s: %v", rt.Host, err) |
| 421 | continue |
| 422 | } |
| 423 | for _, row := range rows { |
| 424 | if row.Kind != KindOnce || row.NextAt != 0 || row.LastAt >= cutoff { |
| 425 | continue |
| 426 | } |
| 427 | if err := deleteSelectedTimer(hctx, r.Reg, row); err != nil { |
| 428 | log.Printf("schedule: runner prune %s/%s: %v", row.Host, row.Name, err) |
| 429 | } |
| 430 | } |
| 431 | } |
| 432 | } |
| 433 | |
| 434 | type timerWorkKey struct{} |
| 435 | type timerWork struct { |
| 436 | ctx context.Context |
| 437 | work hostwrite.Work |
| 438 | } |
| 439 | |
| 440 | func withTimerWork(ctx context.Context, fn func() error) error { |
| 441 | selected, ok := ctx.Value(timerWorkKey{}).(timerWork) |
| 442 | if !ok || selected.ctx == nil || selected.work == nil { |
| 443 | return hostwrite.ErrUnavailable |
| 444 | } |
| 445 | return selected.work.With(selected.ctx, hostwrite.Identity{}, fn) |
| 446 | } |
| 447 | |
| 448 | // SocketRetainer is the optional half of Deliverer that keeps per-socket |
| 449 | // state (a connection pool, in the agent's case). The runner's prune leg |
| 450 | // hands it the set of sockets still routed on this box; everything else the |
| 451 | // implementation holds is for a route that is gone. |
| 452 | type SocketRetainer interface { |
| 453 | RetainSockets(live map[string]bool) |
| 454 | } |
| 455 | |
| 456 | // holdsArmed reports whether the move-hold wiring is on. The host package |
| 457 | // sets Gate (armScheduleHolds) only with the Phase 1 switch on and |
| 458 | // scheduled work armed; Gate nil is the switch off, and then every hold |
| 459 | // check in Tick is skipped and the runner is main's code path exactly: the |
| 460 | // due query is DueTimers, there is no decision read inside the claim, no |
| 461 | // check between a host's timers, and a row whose status is a hold is |
| 462 | // delivered and recorded like any other. |
| 463 | func (r *Runner) holdsArmed() bool { return r.Gate != nil } |
| 464 | |
| 465 | // claim marks host in flight and enters its gate; false if it already was in |
| 466 | // flight, or a move's drain holds or waits for the host (HostGate). |
| 467 | func (r *Runner) claim(host string) bool { |
| 468 | r.mu.Lock() |
| 469 | defer r.mu.Unlock() |
| 470 | if r.busy[host] { |
| 471 | return false |
| 472 | } |
| 473 | if !r.Gate.tryEnter(host) { |
| 474 | return false |
| 475 | } |
| 476 | if r.busy == nil { |
| 477 | r.busy = map[string]bool{} |
| 478 | } |
| 479 | r.busy[host] = true |
| 480 | return true |
| 481 | } |
| 482 | |
| 483 | // Busy reports whether a delivery to host is in flight: the per-host claim |
| 484 | // (moves spec §9.4 "drained"). A snapshot: the cut's drain is |
| 485 | // Gate.Drain, which also waits for a claim taken after this answer. |
| 486 | func (r *Runner) Busy(host string) bool { |
| 487 | r.mu.Lock() |
| 488 | defer r.mu.Unlock() |
| 489 | return r.busy[host] |
| 490 | } |
| 491 | |
| 492 | func (r *Runner) release(host string) { |
| 493 | r.mu.Lock() |
| 494 | delete(r.busy, host) |
| 495 | r.Gate.leave(host) |
| 496 | r.mu.Unlock() |
| 497 | } |
| 498 | |
| 499 | // held reports whether host's row is held by an operation now, or cannot be |
| 500 | // read: either way the rest of the host's timers wait for the next tick. |
| 501 | func (r *Runner) held(ctx context.Context, host string) bool { |
| 502 | var rt *registry.Route |
| 503 | err := withTimerWork(ctx, func() error { |
| 504 | var readErr error |
| 505 | rt, readErr = r.Reg.Get(host) |
| 506 | return readErr |
| 507 | }) |
| 508 | return err != nil || rt == nil || registry.IsHold(rt.Status) |
| 509 | } |
| 510 | |
| 511 | // slots is the cross-tick semaphore, made on first use so a zero Runner |
| 512 | // literal (every test, and the agent's) needs no constructor. |
| 513 | func (r *Runner) slots() chan struct{} { |
| 514 | r.mu.Lock() |
| 515 | defer r.mu.Unlock() |
| 516 | if r.sem == nil { |
| 517 | r.sem = make(chan struct{}, MaxParallelHosts) |
| 518 | } |
| 519 | return r.sem |
| 520 | } |
| 521 | |
| 522 | // wait blocks until every host goroutine launched so far has finished — |
| 523 | // the tests' seam, so a Tick's post-conditions can be asserted without |
| 524 | // sleeping. Nothing in the agent calls it. |
| 525 | func (r *Runner) wait() { r.inflight.Wait() } |
| 526 | |
| 527 | func (r *Runner) timerStillDue(ctx context.Context, t registry.Timer) bool { |
| 528 | var current *registry.Timer |
| 529 | err := withTimerWork(ctx, func() error { |
| 530 | var readErr error |
| 531 | current, readErr = r.Reg.GetTimer(t.Host, t.Name) |
| 532 | return readErr |
| 533 | }) |
| 534 | return err == nil && current.Registration == t.Registration && current.NextAt == t.NextAt && current.NextAt > 0 && current.NextAt <= r.now().Unix() |
| 535 | } |
| 536 | |
| 537 | func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) { |
| 538 | if !r.timerStillDue(ctx, t) { |
| 539 | return |
| 540 | } |
| 541 | |
| 542 | sp, err := ParseSpec(t.Spec) |
| 543 | if err != nil { |
| 544 | log.Printf("schedule: %s/%s: bad spec %q: %v — dropping the row", rt.Host, t.Name, t.Spec, err) |
| 545 | if err := deleteSelectedTimer(ctx, r.Reg, t); err != nil { |
| 546 | log.Printf("schedule: %s/%s: drop bad row: %v", rt.Host, t.Name, err) |
| 547 | } |
| 548 | return |
| 549 | } |
| 550 | if r.Skip != nil { |
| 551 | // The token is the one write a tick makes on the instance's own |
| 552 | // ground, so the row is read again right before it. |
| 553 | var current *registry.Route |
| 554 | err := withTimerWork(ctx, func() error { |
| 555 | var readErr error |
| 556 | current, readErr = r.Reg.Get(rt.Host) |
| 557 | return readErr |
| 558 | }) |
| 559 | if err != nil || current == nil || r.Skip(*current) { |
| 560 | return |
| 561 | } |
| 562 | } |
| 563 | start := r.now() |
| 564 | token, err := r.Token(rt.Host) |
| 565 | if err != nil { |
| 566 | r.record(ctx, rt, t, sp, start, Outcome{Status: "no instance token: " + err.Error(), Retry: true}, 0) |
| 567 | return |
| 568 | } |
| 569 | req, err := BuildRequest(rt, t, token, r.Build) |
| 570 | if err != nil { |
| 571 | // Not retried: the path is the row's, and it will be exactly as bad |
| 572 | // next time. The row advances (or finishes) with the reason on it. |
| 573 | r.record(ctx, rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0) |
| 574 | return |
| 575 | } |
| 576 | if !r.timerStillDue(ctx, t) { |
| 577 | return |
| 578 | } |
| 579 | dctx, cancel := context.WithTimeout(ctx, r.timeout()) |
| 580 | req = req.WithContext(dctx) |
| 581 | var status int |
| 582 | var derr error |
| 583 | if r.HostWrites.Enabled { |
| 584 | status, derr = r.deliverClaim(dctx, rt, t, req) |
| 585 | } else { |
| 586 | status, derr = r.Deliver.Deliver(dctx, rt, req) |
| 587 | } |
| 588 | cancel() |
| 589 | if errors.Is(derr, hostwrite.ErrUnavailable) { |
| 590 | return |
| 591 | } |
| 592 | if errors.Is(derr, context.DeadlineExceeded) { |
| 593 | derr = ErrTimeout |
| 594 | } |
| 595 | r.record(ctx, rt, t, sp, start, classify(status, derr, r.timeout()), r.now().Sub(start)) |
| 596 | } |
| 597 | |
| 598 | // record applies the policy table to one outcome and writes the row, the |
| 599 | // member-facing event, and nothing else. |
| 600 | func (r *Runner) record(ctx context.Context, rt registry.Route, t registry.Timer, sp Spec, at time.Time, o Outcome, took time.Duration) { |
| 601 | attempts := 0 |
| 602 | status := o.Status |
| 603 | natural := sp.Next(at) // zero for a finished once |
| 604 | next := natural.Unix() |
| 605 | if natural.IsZero() { |
| 606 | next = 0 |
| 607 | } |
| 608 | // due is the occurrence the NEXT delivery is for (X-Carlos-Schedule-At, |
| 609 | // the app's dedupe key): the new natural occurrence when the row |
| 610 | // advances, and the one just attempted when it is retried. |
| 611 | due := next |
| 612 | level := "info" |
| 613 | if o.Retry { |
| 614 | attempts = t.Attempts + 1 |
| 615 | if attempts >= MaxAttempts { |
| 616 | status = fmt.Sprintf("failed after %d attempts: %s", MaxAttempts, o.Status) |
| 617 | attempts = 0 |
| 618 | level = "warn" |
| 619 | } else { |
| 620 | backoff := backoffBase << (attempts - 1) |
| 621 | if backoff > backoffCap { |
| 622 | backoff = backoffCap |
| 623 | } |
| 624 | retryAt := at.Add(backoff) |
| 625 | // Never let a retry outlive the next natural occurrence: past it |
| 626 | // the retry and the occurrence would be the same delivery, and a |
| 627 | // 15m backoff on a 1m schedule would skip 14 fires. A once row |
| 628 | // has no natural next (zero), so its retries are never clamped. |
| 629 | if !natural.IsZero() && retryAt.After(natural) { |
| 630 | retryAt = natural |
| 631 | } |
| 632 | next = retryAt.Unix() |
| 633 | due = dueOf(t) |
| 634 | level = "warn" |
| 635 | } |
| 636 | } else if !isSuccess(o.Status) { |
| 637 | level = "warn" |
| 638 | } |
| 639 | var applied bool |
| 640 | err := withTimerWork(ctx, func() error { |
| 641 | var writeErr error |
| 642 | applied, writeErr = r.Reg.MarkTimerFired(rt.Host, t.Name, t.Registration, at.Unix(), status, took.Milliseconds(), attempts, next, due) |
| 643 | return writeErr |
| 644 | }) |
| 645 | if err != nil { |
| 646 | log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err) |
| 647 | // Flag off keeps the old behaviour: the fire event still goes out. |
| 648 | // Under the claim, a refused or failed guarded write stops here. |
| 649 | if selected, ok := ctx.Value(timerWorkKey{}).(timerWork); !ok || !hostwrite.Legacy(selected.work) { |
| 650 | return |
| 651 | } |
| 652 | } |
| 653 | if r.Event != nil { |
| 654 | msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond)) |
| 655 | if attempts > 0 && applied { |
| 656 | msg += fmt.Sprintf(" (attempt %d of %d; retrying at %s)", attempts, MaxAttempts, time.Unix(next, 0).UTC().Format(time.RFC3339)) |
| 657 | } |
| 658 | r.Event(rt, level, msg) |
| 659 | } |
| 660 | } |
| 661 | |
| 662 | func isSuccess(status string) bool { return len(status) == 3 && status[0] == '2' } |
| 663 | |
| 664 | // Outcome is one line of the policy table (spec §6.3). |
| 665 | type Outcome struct { |
| 666 | Status string |
| 667 | Retry bool |
| 668 | } |
| 669 | |
| 670 | func classify(status int, err error, timeout time.Duration) Outcome { |
| 671 | switch { |
| 672 | case errors.Is(err, ErrTimeout): |
| 673 | return Outcome{Status: "timeout after " + timeout.String()} |
| 674 | case err != nil: |
| 675 | // No response / transport error (status 0). Recorded as 503 because |
| 676 | // that is what a member's own client would have seen, and retried |
| 677 | // for the same reason: nothing says the handler ran. |
| 678 | return Outcome{Status: "503: " + err.Error(), Retry: true} |
| 679 | case status >= 200 && status < 300: |
| 680 | return Outcome{Status: strconv.Itoa(status)} |
| 681 | case status >= 500, status == 408, status == 429: |
| 682 | return Outcome{Status: strconv.Itoa(status), Retry: true} |
| 683 | default: |
| 684 | return Outcome{Status: strconv.Itoa(status) + " (app refused)"} |
| 685 | } |
| 686 | } |
| 687 | |
| 688 | // Classify is the exported policy for the host package's tests. |
| 689 | func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) } |
| 690 | |
| 691 | func dueOf(t registry.Timer) int64 { |
| 692 | if t.DueAt != 0 { |
| 693 | return t.DueAt |
| 694 | } |
| 695 | if t.Kind == KindOnce { |
| 696 | if sp, err := ParseSpec(t.Spec); err == nil && sp.Kind == KindOnce && sp.At.Unix() == 0 { |
| 697 | return 0 |
| 698 | } |
| 699 | } |
| 700 | return t.NextAt |
| 701 | } |
| 702 | |
| 703 | // BuildRequest is the tick (spec §6.3's header table). Content-Length 0 and |
| 704 | // no body: a tick carries no data, by design. A path the URL parser refuses |
| 705 | // is an error, never a quiet POST somewhere else — ValidPath runs the same |
| 706 | // parser at declaration, so this only fires for a row written some other |
| 707 | // way, and the runner records it on the row rather than delivering. |
| 708 | func BuildRequest(rt registry.Route, t registry.Timer, token, build string) (*http.Request, error) { |
| 709 | if err := ValidPath(t.Path); err != nil { |
| 710 | return nil, err |
| 711 | } |
| 712 | req, err := http.NewRequest(http.MethodPost, "http://"+rt.Host+t.Path, http.NoBody) |
| 713 | if err != nil { |
| 714 | return nil, err |
| 715 | } |
| 716 | req.Host = rt.Host |
| 717 | req.ContentLength = 0 |
| 718 | req.Header.Set("Authorization", "Bearer "+token) |
| 719 | req.Header.Set("X-Carlos-Schedule", t.Name) |
| 720 | req.Header.Set("X-Carlos-Schedule-Kind", t.Kind) |
| 721 | req.Header.Set("X-Carlos-Schedule-At", strconv.FormatInt(dueOf(t), 10)) |
| 722 | req.Header.Set("User-Agent", "carlos-schedule/"+build) |
| 723 | return req, nil |
| 724 | } |
| 725 | |
| 726 | // TokenFromEnvPaths is the production Token with the env file's location |
| 727 | // taken from envPath rather than assumed flat: isolated-exec's nested rows |
| 728 | // keep <host>.env inside their own instance directory (design §7), and the |
| 729 | // host wiring (schedule_wiring.go) resolves that through the same |
| 730 | // classification the exec backend used to decide where it spawned the |
| 731 | // instance. Every caller passes a resolver here now; a caller that predates |
| 732 | // the nested layout gets one that always names the flat path. |
| 733 | func TokenFromEnvPaths(envPath func(host string) string) func(host string) (string, error) { |
| 734 | return func(host string) (string, error) { |
| 735 | env, err := readEnvFile(envPath(host)) |
| 736 | if err != nil { |
| 737 | return "", err |
| 738 | } |
| 739 | if env["CARLOS_ADMIN_TOKEN"] == "" { |
| 740 | return "", errors.New("no CARLOS_ADMIN_TOKEN") |
| 741 | } |
| 742 | return env["CARLOS_ADMIN_TOKEN"], nil |
| 743 | } |
| 744 | } |
| 745 | |
amadan