| 1 | // The timers table: scheduled work materialized per (host, name) — spec |
| 2 | // docs/superpowers/specs/2026-08-23-scheduled-work-design.md §5. Rows come |
| 3 | // from two writers with disjoint columns: converge (schedule.Converge) and |
| 4 | // the control socket own spec/path/kind; the runner owns next_at after the |
| 5 | // first fire, last_*, attempts. UpsertTimer encodes that split so neither |
| 6 | // writer can clobber the other. Deliberately NOT in Fingerprint(): a timer |
| 7 | // changes nothing about routing. |
| 8 | package registry |
| 9 | |
| 10 | import ( |
| 11 | "context" |
| 12 | "crypto/rand" |
| 13 | "database/sql" |
| 14 | "encoding/hex" |
| 15 | "errors" |
| 16 | "fmt" |
| 17 | "github.com/carlosframework/platform/internal/hostwrite" |
| 18 | "os" |
| 19 | "sort" |
| 20 | "strings" |
| 21 | ) |
| 22 | |
| 23 | type Timer struct { |
| 24 | ClaimOwnership |
| 25 | |
| 26 | Host, Name, Kind, Spec, Path string |
| 27 | NextAt, LastAt int64 |
| 28 | LastStatus string |
| 29 | LastMs int64 |
| 30 | Attempts int |
| 31 | CreatedAt int64 |
| 32 | // DueAt is the natural occurrence the row is (or was last) delivering — |
| 33 | // the value X-Carlos-Schedule-At carries, and the key an app dedupes |
| 34 | // on. It is NOT next_at: after a retry next_at is the retry instant and |
| 35 | // after `run` it is "now", while the occurrence stays put. Moves only |
| 36 | // when the row advances to a new occurrence (MarkTimerFired with a new |
| 37 | // due, FireTimerNow, or an upsert that moves next_at). Zero on rows |
| 38 | // written before the column existed; readers fall back to NextAt. |
| 39 | DueAt int64 |
| 40 | Registration string `json:"-"` |
| 41 | } |
| 42 | |
| 43 | var ErrTimerKind = errors.New("registry: timer exists with a different kind") |
| 44 | |
| 45 | const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration, claim_incarnation, claim_stamp, claim_owner, claim_row_id` |
| 46 | |
| 47 | func scanTimer(row interface{ Scan(...any) error }) (Timer, error) { |
| 48 | var t Timer |
| 49 | err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt, &t.Registration, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID) |
| 50 | return t, err |
| 51 | } |
| 52 | |
| 53 | // UpsertTimer inserts t, or updates an existing row's spec/path. A changed spec/path or non-zero NextAt starts a new registration and resets attempts; unchanged declarations preserve delivery state. |
| 54 | func (r *Registry) UpsertTimer(t Timer) error { |
| 55 | if r.remote != nil { |
| 56 | _, err := r.remote.call(nil, remoteRequest{Method: "UpsertTimer", T: t}, nil) |
| 57 | return err |
| 58 | } |
| 59 | |
| 60 | if !hostwrite.Enabled(os.Getenv) { |
| 61 | return r.upsertLegacyTimer(t) |
| 62 | } |
| 63 | tx, err := r.beginClaimTransaction(context.Background(), false) |
| 64 | if err != nil { |
| 65 | return err |
| 66 | } |
| 67 | defer tx.Rollback() |
| 68 | rt, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host=?`, t.Host).Scan) |
| 69 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 70 | return err |
| 71 | } |
| 72 | if err == nil && rt.ClaimOwnership.Valid() { |
| 73 | if t.ClaimOwner != "" && (t.ClaimIncarnation != rt.ClaimIncarnation || t.ClaimStamp != rt.ClaimStamp || t.ClaimOwner != rt.ClaimOwner) { |
| 74 | return ErrClaimOwnership |
| 75 | } |
| 76 | t.ClaimOwnership = rt.ClaimOwnership |
| 77 | } |
| 78 | var id [16]byte |
| 79 | if _, err = rand.Read(id[:]); err != nil { |
| 80 | return err |
| 81 | } |
| 82 | t.Registration = hex.EncodeToString(id[:]) |
| 83 | if t.ClaimOwnership.Valid() { |
| 84 | t.ClaimRowID = t.Registration |
| 85 | } |
| 86 | if err = upsertTimer(tx, t); err != nil { |
| 87 | return err |
| 88 | } |
| 89 | return tx.Commit() |
| 90 | } |
| 91 | func upsertTimer(tx interface { |
| 92 | Exec(string, ...any) (sql.Result, error) |
| 93 | }, t Timer) error { |
| 94 | const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)` |
| 95 | res, err := tx.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| 96 | ON CONFLICT(host, name) DO UPDATE SET |
| 97 | spec=excluded.spec, path=excluded.path, |
| 98 | next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END, |
| 99 | due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END, |
| 100 | attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END, |
| 101 | registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END, |
| 102 | claim_row_id=CASE WHEN `+rearm+` THEN excluded.claim_row_id ELSE timers.claim_row_id END |
| 103 | WHERE timers.kind=excluded.kind AND timers.claim_incarnation=excluded.claim_incarnation |
| 104 | AND timers.claim_stamp=excluded.claim_stamp AND timers.claim_owner=excluded.claim_owner`, |
| 105 | t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration, t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID) |
| 106 | if err != nil { |
| 107 | return err |
| 108 | } |
| 109 | n, err := res.RowsAffected() |
| 110 | if err != nil { |
| 111 | return err |
| 112 | } |
| 113 | if n == 0 { |
| 114 | return ErrTimerKind |
| 115 | } |
| 116 | return nil |
| 117 | } |
| 118 | |
| 119 | // ReplaceTimers makes host's timer rows exactly rows, in one transaction, |
| 120 | // keeping every column as given: next_at, due_at, attempts and |
| 121 | // registration included. It is how an installation imports the wake state |
| 122 | // it was handed (region HA spec 1 §5.7, moves spec §9.4). |
| 123 | // |
| 124 | // UpsertTimer is the wrong tool for that: it mints a fresh registration |
| 125 | // and resets the runner's bookkeeping on a rearm. |
| 126 | func (r *Registry) ReplaceTimers(host string, rows []Timer) error { |
| 127 | if r.remote != nil { |
| 128 | _, err := r.remote.call(nil, remoteRequest{Method: "ReplaceTimers", Host: host, Rows: rows}, nil) |
| 129 | return err |
| 130 | } |
| 131 | |
| 132 | if !hostwrite.Enabled(os.Getenv) { |
| 133 | return r.replaceLegacyTimers(host, rows) |
| 134 | } |
| 135 | for _, t := range rows { |
| 136 | if t.Host != host { |
| 137 | return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host) |
| 138 | } |
| 139 | } |
| 140 | tx, err := r.beginClaimTransaction(context.Background(), false) |
| 141 | if err != nil { |
| 142 | return err |
| 143 | } |
| 144 | defer tx.Rollback() |
| 145 | rt, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host=?`, host).Scan) |
| 146 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 147 | return err |
| 148 | } |
| 149 | if err == nil && rt.ClaimOwnership.Valid() { |
| 150 | for i := range rows { |
| 151 | if rows[i].Registration == "" { |
| 152 | return ErrClaimOwnership |
| 153 | } |
| 154 | rows[i].ClaimOwnership = rt.ClaimOwnership |
| 155 | rows[i].ClaimRowID = rows[i].Registration |
| 156 | } |
| 157 | } |
| 158 | if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil { |
| 159 | return err |
| 160 | } |
| 161 | for _, t := range rows { |
| 162 | if _, err := tx.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, |
| 163 | t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration, t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID); err != nil { |
| 164 | return fmt.Errorf("registry: insert timer %q: %w", t.Name, err) |
| 165 | } |
| 166 | } |
| 167 | return tx.Commit() |
| 168 | } |
| 169 | |
| 170 | func (r *Registry) GetTimer(host, name string) (*Timer, error) { |
| 171 | if r.remote != nil { |
| 172 | out, err := r.remote.call(nil, remoteRequest{Method: "GetTimer", Host: host, Name: name}, nil) |
| 173 | return out.Timer, err |
| 174 | } |
| 175 | |
| 176 | t, err := scanTimer(r.sql.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, host, name)) |
| 177 | if errors.Is(err, sql.ErrNoRows) { |
| 178 | return nil, os.ErrNotExist |
| 179 | } |
| 180 | if err != nil { |
| 181 | return nil, err |
| 182 | } |
| 183 | return &t, nil |
| 184 | } |
| 185 | |
| 186 | func (r *Registry) DeleteTimer(host, name string) error { |
| 187 | if r.remote != nil { |
| 188 | _, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimer", Host: host, Name: name}, nil) |
| 189 | return err |
| 190 | } |
| 191 | |
| 192 | _, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=?`, host, name) |
| 193 | return err |
| 194 | } |
| 195 | |
| 196 | func (r *Registry) DeleteTimerRegistration(host, name, registration string) error { |
| 197 | if r.remote != nil { |
| 198 | _, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimerRegistration", Host: host, Name: name, Registration: registration}, nil) |
| 199 | return err |
| 200 | } |
| 201 | |
| 202 | _, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=? AND registration=?`, host, name, registration) |
| 203 | return err |
| 204 | } |
| 205 | |
| 206 | func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) { |
| 207 | rows, err := r.sql.Query(q, args...) |
| 208 | if err != nil { |
| 209 | return nil, err |
| 210 | } |
| 211 | defer rows.Close() |
| 212 | var out []Timer |
| 213 | for rows.Next() { |
| 214 | t, err := scanTimer(rows) |
| 215 | if err != nil { |
| 216 | return nil, err |
| 217 | } |
| 218 | out = append(out, t) |
| 219 | } |
| 220 | return out, rows.Err() |
| 221 | } |
| 222 | |
| 223 | func (r *Registry) TimersFor(host string) ([]Timer, error) { |
| 224 | if r.remote != nil { |
| 225 | out, err := r.remote.call(nil, remoteRequest{Method: "TimersFor", Host: host}, nil) |
| 226 | return out.Timers, err |
| 227 | } |
| 228 | |
| 229 | return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE host=? ORDER BY name`, host) |
| 230 | } |
| 231 | |
| 232 | // DueTimers is the runner's batch: rows due at or before now, earliest |
| 233 | // first, at most limit of them. |
| 234 | // |
| 235 | // excludeHosts is what keeps the batch FAIR. The runner delivers one timer |
| 236 | // at a time per host and skips a host whose delivery is still running, so a |
| 237 | // host with more due rows than the batch holds — or one parked in a |
| 238 | // 30-minute handler — would otherwise fill every slot of every batch with |
| 239 | // rows the runner is about to skip, and another host's due timer would never |
| 240 | // be seen at all. Passing the busy set here moves the filter into the query, |
| 241 | // where the LIMIT is applied, instead of after it. An empty (or nil) slice |
| 242 | // is no filter, which is the ordinary case: nothing in flight. |
| 243 | func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) { |
| 244 | if r.remote != nil { |
| 245 | out, err := r.remote.call(nil, remoteRequest{Method: "DueTimers", Now: now, Limit: limit, ExcludeHosts: excludeHosts}, nil) |
| 246 | return out.Timers, err |
| 247 | } |
| 248 | |
| 249 | return r.dueTimers(now, limit, excludeHosts, false) |
| 250 | } |
| 251 | |
| 252 | // DueUnheldTimers is DueTimers without the rows of held routes (moves spec |
| 253 | // §5.2, HoldStatuses). The exclusion is in the query, before the LIMIT, so |
| 254 | // a held host's rows cannot fill the batch with rows the runner would only |
| 255 | // skip and hide another host's due timer behind them (moves Task 6 fix |
| 256 | // round 1). A row whose host has no route is still returned, as DueTimers |
| 257 | // returns it. The runner asks this only with its hold wiring armed; |
| 258 | // DueTimers stays main's query. |
| 259 | func (r *Registry) DueUnheldTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) { |
| 260 | if r.remote != nil { |
| 261 | out, err := r.remote.call(nil, remoteRequest{Method: "DueUnheldTimers", Now: now, Limit: limit, ExcludeHosts: excludeHosts}, nil) |
| 262 | return out.Timers, err |
| 263 | } |
| 264 | |
| 265 | return r.dueTimers(now, limit, excludeHosts, true) |
| 266 | } |
| 267 | |
| 268 | // dueTimers is both queries. Neither returns the rows of a gated route (an |
| 269 | // uncommitted import's, seeded-instances spec §5.4), and the exclusion is |
| 270 | // in the query, before the LIMIT (final review F5): the runner skips such a |
| 271 | // row without advancing it, so gated rows left in the batch would stay due |
| 272 | // for ever and could fill every slot of it, hiding every other host's due |
| 273 | // timer. The runner keeps its own check. |
| 274 | func (r *Registry) dueTimers(now int64, limit int, excludeHosts []string, skipHeld bool) ([]Timer, error) { |
| 275 | q := `SELECT ` + timerCols + ` FROM timers WHERE next_at > 0 AND next_at <= ?` + |
| 276 | ` AND host NOT IN (SELECT host FROM routes WHERE import_gate != '')` |
| 277 | args := []any{now} |
| 278 | if skipHeld { |
| 279 | statuses := make([]string, 0, len(HoldStatuses)) |
| 280 | for s := range HoldStatuses { |
| 281 | statuses = append(statuses, s) |
| 282 | } |
| 283 | sort.Strings(statuses) |
| 284 | q += ` AND host NOT IN (SELECT host FROM routes WHERE status IN (?` + strings.Repeat(",?", len(statuses)-1) + `))` |
| 285 | for _, s := range statuses { |
| 286 | args = append(args, s) |
| 287 | } |
| 288 | } |
| 289 | if len(excludeHosts) > 0 { |
| 290 | q += ` AND host NOT IN (?` + strings.Repeat(",?", len(excludeHosts)-1) + `)` |
| 291 | for _, h := range excludeHosts { |
| 292 | args = append(args, h) |
| 293 | } |
| 294 | } |
| 295 | q += ` ORDER BY next_at, host, name LIMIT ?` |
| 296 | return r.queryTimers(q, append(args, limit)...) |
| 297 | } |
| 298 | |
| 299 | // FireTimerNow makes the row due at now, and now IS the occurrence being |
| 300 | // delivered (a `carlos schedule run` is its own occurrence, not an early |
| 301 | // delivery of the next natural one). |
| 302 | func (r *Registry) FireTimerNow(host, name string, now int64) error { |
| 303 | if r.remote != nil { |
| 304 | _, err := r.remote.call(nil, remoteRequest{Method: "FireTimerNow", Host: host, Name: name, Now: now}, nil) |
| 305 | return err |
| 306 | } |
| 307 | |
| 308 | var id [16]byte |
| 309 | if _, err := rand.Read(id[:]); err != nil { |
| 310 | return err |
| 311 | } |
| 312 | registration := hex.EncodeToString(id[:]) |
| 313 | _, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=?, attempts=0, registration=? , claim_row_id=CASE WHEN claim_owner='' THEN claim_row_id ELSE ? END WHERE host=? AND name=?`, now, now, registration, registration, host, name) |
| 314 | return err |
| 315 | } |
| 316 | |
| 317 | // MarkTimerFired records one delivery. next is when the row is due again; |
| 318 | // due is the occurrence THAT next delivery is for: the same value as next |
| 319 | // when the row advances, and the row's current DueAt when next is a retry |
| 320 | // of the occurrence just attempted — the runner decides which. |
| 321 | func (r *Registry) MarkTimerFired(host, name, registration string, at int64, status string, ms int64, attempts int, next, due int64) (bool, error) { |
| 322 | if r.remote != nil { |
| 323 | out, err := r.remote.call(nil, remoteRequest{Method: "MarkTimerFired", Host: host, Name: name, Registration: registration, At: at, Status: status, Ms: ms, Attempts: attempts, NextAt: next, Due: due}, nil) |
| 324 | return out.Changed, err |
| 325 | } |
| 326 | |
| 327 | res, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=? AND registration=?`, |
| 328 | at, status, ms, attempts, next, due, host, name, registration) |
| 329 | if err != nil { |
| 330 | return false, err |
| 331 | } |
| 332 | n, err := res.RowsAffected() |
| 333 | return n != 0, err |
| 334 | } |
| 335 | |
| 336 | // SetTimerStatus writes last_status alone — converge's one word on a row |
| 337 | // it parked (schedule.StatusNeverFires); last_at stays as it was, so the |
| 338 | // status reads as a condition, not a fire. |
| 339 | func (r *Registry) SetTimerStatus(host, name, status string) error { |
| 340 | if r.remote != nil { |
| 341 | _, err := r.remote.call(nil, remoteRequest{Method: "SetTimerStatus", Host: host, Name: name, Status: status}, nil) |
| 342 | return err |
| 343 | } |
| 344 | |
| 345 | _, err := r.sql.Exec(`UPDATE timers SET last_status=? WHERE host=? AND name=?`, status, host, name) |
| 346 | return err |
| 347 | } |
| 348 | |
| 349 | func (r *Registry) CountTimers(host, kind string) (int, error) { |
| 350 | if r.remote != nil { |
| 351 | out, err := r.remote.call(nil, remoteRequest{Method: "CountTimers", Host: host, Kind: kind}, nil) |
| 352 | return out.Number, err |
| 353 | } |
| 354 | |
| 355 | var n int |
| 356 | err := r.sql.QueryRow(`SELECT COUNT(*) FROM timers WHERE host=? AND kind=?`, host, kind).Scan(&n) |
| 357 | return n, err |
| 358 | } |
| 359 | |
| 360 | // PruneTimers: rows for hosts no longer routed here, and finished once-rows |
| 361 | // older than doneBefore. |
| 362 | // |
| 363 | // An empty alive set deletes EVERY row. That is the right answer for a |
| 364 | // genuinely empty registry — routes and timers live in the same file, so |
| 365 | // every row really is orphaned — but it is a wipe, and there are two callers |
| 366 | // now: schedule.Converger.Pass (every pass, on a box with a deploy store) and |
| 367 | // schedule.Runner.prune (daily, on every box). Both build alive from the |
| 368 | // registry's own route list; the runner declines the sweep outright when that |
| 369 | // list is empty rather than hand the wipe over. A third caller should do the |
| 370 | // same, or say in its own comment why the wipe is what it wants. |
| 371 | func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error) { |
| 372 | if r.remote != nil { |
| 373 | out, err := r.remote.call(nil, remoteRequest{Method: "PruneTimers", Alive: alive, DoneBefore: doneBefore}, nil) |
| 374 | return out.Number, err |
| 375 | } |
| 376 | |
| 377 | hosts := make([]any, 0, len(alive)) |
| 378 | for h := range alive { |
| 379 | hosts = append(hosts, h) |
| 380 | } |
| 381 | q := `DELETE FROM timers WHERE (kind='once' AND next_at=0 AND last_at < ?)` |
| 382 | args := []any{doneBefore} |
| 383 | if len(hosts) == 0 { |
| 384 | q += ` OR 1=1` |
| 385 | } else { |
| 386 | q += ` OR host NOT IN (?` + strings.Repeat(",?", len(hosts)-1) + `)` |
| 387 | args = append(args, hosts...) |
| 388 | } |
| 389 | res, err := r.sql.Exec(q, args...) |
| 390 | if err != nil { |
| 391 | return 0, err |
| 392 | } |
| 393 | n, err := res.RowsAffected() |
| 394 | if err != nil { |
| 395 | return 0, err |
| 396 | } |
| 397 | return int(n), nil |
| 398 | } |
| 399 | |
| 400 | func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (int, error) { |
| 401 | if r.remote != nil { |
| 402 | out, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimersExcept", Host: host, Kind: kind, Keep: keep}, nil) |
| 403 | return out.Number, err |
| 404 | } |
| 405 | |
| 406 | names := make([]any, 0, len(keep)) |
| 407 | for n := range keep { |
| 408 | names = append(names, n) |
| 409 | } |
| 410 | q := `DELETE FROM timers WHERE host=? AND kind=?` |
| 411 | args := []any{host, kind} |
| 412 | if len(names) > 0 { |
| 413 | q += ` AND name NOT IN (?` + strings.Repeat(",?", len(names)-1) + `)` |
| 414 | args = append(args, names...) |
| 415 | } |
| 416 | res, err := r.sql.Exec(q, args...) |
| 417 | if err != nil { |
| 418 | return 0, err |
| 419 | } |
| 420 | n, err := res.RowsAffected() |
| 421 | if err != nil { |
| 422 | return 0, err |
| 423 | } |
| 424 | return int(n), nil |
| 425 | } |
| 426 | |
| 427 | // EarliestTimers is the hibernation guard's batch: for every host with at |
| 428 | // least one armed row, the soonest next_at among them. One query, because |
| 429 | // the activator's sweep reads it BEFORE taking its mutex (activator.go's |
| 430 | // "nothing under a.mu should wait on SQLite") and a per-entry query inside |
| 431 | // the sweep loop would put SQLite exactly where that comment forbids it. |
| 432 | // |
| 433 | // next_at = 0 is "not armed" — a parked cron, or a row converge wrote and |
| 434 | // the runner has not yet dated — and is excluded here rather than by the |
| 435 | // caller, matching DueTimers' own `next_at > 0` gate. |
| 436 | func (r *Registry) EarliestTimers() (map[string]int64, error) { |
| 437 | if r.remote != nil { |
| 438 | out, err := r.remote.call(nil, remoteRequest{Method: "EarliestTimers"}, nil) |
| 439 | return out.Times, err |
| 440 | } |
| 441 | |
| 442 | rows, err := r.sql.Query(`SELECT host, MIN(next_at) FROM timers WHERE next_at > 0 GROUP BY host`) |
| 443 | if err != nil { |
| 444 | return nil, err |
| 445 | } |
| 446 | defer rows.Close() |
| 447 | out := make(map[string]int64) |
| 448 | for rows.Next() { |
| 449 | var host string |
| 450 | var next int64 |
| 451 | if err := rows.Scan(&host, &next); err != nil { |
| 452 | return nil, err |
| 453 | } |
| 454 | out[host] = next |
| 455 | } |
| 456 | return out, rows.Err() |
| 457 | } |
| 458 | |
| 459 | const legacyTimerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration` |
| 460 | |
| 461 | func (r *Registry) upsertLegacyTimer(t Timer) error { |
| 462 | const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)` |
| 463 | res, err := r.sql.Exec(`INSERT INTO timers (`+legacyTimerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,lower(hex(randomblob(16)))) |
| 464 | ON CONFLICT(host, name) DO UPDATE SET |
| 465 | spec=excluded.spec, path=excluded.path, |
| 466 | next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END, |
| 467 | due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END, |
| 468 | attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END, |
| 469 | registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END |
| 470 | WHERE timers.kind=excluded.kind`, |
| 471 | t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt) |
| 472 | if err != nil { |
| 473 | return err |
| 474 | } |
| 475 | n, err := res.RowsAffected() |
| 476 | if err != nil { |
| 477 | return err |
| 478 | } |
| 479 | if n == 0 { |
| 480 | return ErrTimerKind |
| 481 | } |
| 482 | return nil |
| 483 | } |
| 484 | |
| 485 | func (r *Registry) replaceLegacyTimers(host string, rows []Timer) error { |
| 486 | for _, t := range rows { |
| 487 | if t.Host != host { |
| 488 | return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host) |
| 489 | } |
| 490 | } |
| 491 | tx, err := r.sql.Begin() |
| 492 | if err != nil { |
| 493 | return err |
| 494 | } |
| 495 | defer tx.Rollback() |
| 496 | if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil { |
| 497 | return err |
| 498 | } |
| 499 | for _, t := range rows { |
| 500 | if _, err := tx.Exec(`INSERT INTO timers (`+legacyTimerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`, |
| 501 | t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration); err != nil { |
| 502 | return fmt.Errorf("registry: insert timer %q: %w", t.Name, err) |
| 503 | } |
| 504 | } |
| 505 | return tx.Commit() |
| 506 | } |
| 507 | |
amadan