121 строка
4.1 KiB
Go
121 строка
4.1 KiB
Go
package models
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
)
|
|
|
|
// DeadWorkerHeartbeatTimeout is the threshold for ReapDeadWorkers — a worker
|
|
// whose last_seen is older than this is considered dead and any leased tasks
|
|
// it owns are reassigned to the pool. Five minutes mirrors the StaleWorkers()
|
|
// check in check_jobs.go so the two reapers cannot disagree about who is
|
|
// dead. See docs/todo.md Phase 4 §5.
|
|
const DeadWorkerHeartbeatTimeout = 5 * time.Minute
|
|
|
|
// ReapDeadWorkers marks any non-dead worker whose last_seen is older than
|
|
// DeadWorkerHeartbeatTimeout as "dead", then reassigns its leased tasks
|
|
// back to the queued pool so other workers (or freshly registered ones)
|
|
// can pick them up. It mirrors the structure of ReapExpiredTasks — two
|
|
// short UPDATE statements, cheap enough to run from the web process
|
|
// every 30s.
|
|
//
|
|
// The returned tuple is (reaped, reassigned, err): reaped counts the
|
|
// workers that flipped to dead during this call; reassigned counts the
|
|
// leased tasks that were given back to the pool. A zero count on either
|
|
// is normal — the reaper is idempotent and the call is silent when
|
|
// nothing is due.
|
|
func ReapDeadWorkers() (reaped int, reassigned int, err error) {
|
|
now := time.Now()
|
|
cutoff := now.Add(-DeadWorkerHeartbeatTimeout)
|
|
|
|
// First flip the workers to dead so the second UPDATE can match the
|
|
// freshly-stamped ids without having to re-derive them in Go.
|
|
r := DB().Exec(`
|
|
UPDATE worker_nodes
|
|
SET status = ?, updated_at = ?
|
|
WHERE status <> ? AND last_seen IS NOT NULL AND last_seen < ?`,
|
|
"dead", now, "dead", cutoff,
|
|
)
|
|
if r.Error != nil {
|
|
return 0, 0, r.Error
|
|
}
|
|
reaped = int(r.RowsAffected)
|
|
|
|
// Nothing flipped → no tasks to return. Cheaper than running an
|
|
// UPDATE that touches 0 rows on every tick when the fleet is healthy.
|
|
if reaped == 0 {
|
|
return 0, 0, nil
|
|
}
|
|
|
|
// Second: clear any leased tasks owned by the now-dead workers.
|
|
// The selector stashes the worker's WorkerID string in lease_owner;
|
|
// matching against the (now-stale) worker row's WorkerID is the
|
|
// same identifier the selector uses, so we don't need an extra
|
|
// join. Tasks in other states (queued, succeeded, dead, …) are
|
|
// unaffected — only leased work the dead worker still owned has
|
|
// to go back to the queue.
|
|
r2 := DB().Exec(`
|
|
UPDATE tasks
|
|
SET state = ?, lease_owner = '', lease_expires_at = NULL, updated_at = ?
|
|
WHERE state = ? AND lease_owner IN (
|
|
SELECT worker_id FROM worker_nodes WHERE status = ?
|
|
)`,
|
|
TaskStateQueued, now, TaskStateLeased, "dead",
|
|
)
|
|
if r2.Error != nil {
|
|
return reaped, 0, r2.Error
|
|
}
|
|
reassigned = int(r2.RowsAffected)
|
|
return reaped, reassigned, nil
|
|
}
|
|
|
|
// StartDeadWorkerReaper launches a goroutine that runs ReapDeadWorkers
|
|
// on the given interval until ctx is canceled. Mirrors StartTaskReaper
|
|
// in this package: same ticker pattern, same logging style, same
|
|
// recover() safety net so a malformed row cannot crash the web process.
|
|
//
|
|
// The default interval is 30s; values <= 0 fall back to the default so
|
|
// the helper is safe to call from any call site without a guard. Wire
|
|
// from main.init() once per process — the reaper uses short row-level
|
|
// locks and is cheap under load (two indexed UPDATEs of <= a few
|
|
// hundred rows in steady state).
|
|
//
|
|
// Passing a nil context falls back to context.Background() so callers
|
|
// can write `models.StartDeadWorkerReaper(nil, ...)` in one-liners
|
|
// (main, tests) without having to import "context" first.
|
|
func StartDeadWorkerReaper(ctx context.Context, interval time.Duration) {
|
|
if interval <= 0 {
|
|
interval = 30 * time.Second
|
|
}
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
go func() {
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
func() {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Printf("dead_worker_reaper: panic recovered: %v", r)
|
|
}
|
|
}()
|
|
reaped, reassigned, err := ReapDeadWorkers()
|
|
if err != nil {
|
|
log.Printf("dead_worker_reaper: error: %v", err)
|
|
return
|
|
}
|
|
if reaped > 0 || reassigned > 0 {
|
|
log.Printf("dead_worker_reaper: reaped=%d reassigned=%d", reaped, reassigned)
|
|
}
|
|
}()
|
|
}
|
|
}
|
|
}()
|
|
}
|