diff --git a/pkg/server/db/db.go b/pkg/server/db/db.go index 719538f..5647342 100644 --- a/pkg/server/db/db.go +++ b/pkg/server/db/db.go @@ -1901,6 +1901,43 @@ type StarvedJob struct { Crashed bool } +// pendingUnassigned is the shared WHERE predicate of the starvation anchor +// lookup and the starved-job sweep: a job counts toward starvation only while +// it is pending and unclaimed. One predicate keeps the two queries from +// drifting apart. +const pendingUnassigned = "status = ? AND worker_id IS NULL" + +// EarliestPendingCreatedAt returns the created_at of the oldest pending, +// unassigned job — the anchor the starvation sweep's pending clock is measured +// from, since every requeue path (failover, timeout requeue, restart recovery) +// refreshes created_at on the way back to pending. +// +// ok is false when no such job exists, i.e. no starvation verdict can be due. +// The scheduler arms a one-shot wake for anchor + NoWorkerJobTimeout so a due +// verdict is emitted just past the deadline instead of at the next check +// interval, which would place it anywhere in the (120s, 150s] acceptance +// window the tick quantizes (AC-17). +// +// ORDER BY created_at ASC must pick the job whose clock expires first: the +// sweep fails every pending job past the cutoff, so arming for anything newer +// would postpone the older jobs' verdict past the acceptance window. +func (d *Database) EarliestPendingCreatedAt() (time.Time, bool, error) { + var anchor time.Time + err := d.db.QueryRow(` + SELECT created_at FROM jobs + WHERE `+pendingUnassigned+` + ORDER BY created_at ASC + LIMIT 1 + `, protocol.JobStatusPending).Scan(&anchor) + if errors.Is(err, sql.ErrNoRows) { + return time.Time{}, false, nil + } + if err != nil { + return time.Time{}, false, fmt.Errorf("failed to read the earliest pending job: %w", err) + } + return anchor, true, nil +} + // GetStarvedPendingJobs lists the pending jobs created before the cutoff (the // set the starvation sweep fails) and classifies each by its most recent // migration event. The classification is computed once here and reused by @@ -1921,7 +1958,7 @@ func (d *Database) GetStarvedPendingJobs(cutoff time.Time, crashReason string) ( LIMIT 1 ), '') = ? FROM jobs j - WHERE j.status = ? AND j.worker_id IS NULL AND j.created_at < ? + WHERE `+pendingUnassigned+` AND j.created_at < ? `, crashReason, protocol.JobStatusPending, cutoff) if err != nil { return nil, fmt.Errorf("failed to list starved pending jobs: %w", err) diff --git a/pkg/server/db/db_test.go b/pkg/server/db/db_test.go index c9e6ade..f0e88ea 100644 --- a/pkg/server/db/db_test.go +++ b/pkg/server/db/db_test.go @@ -2075,3 +2075,92 @@ func TestZeroBusyTimeoutSurfacesBurstConflicts(t *testing.T) { t.Fatal("busy timeout 0 produced no write conflict in a 100-writer burst; submit_conflict would be unreachable from an E2E") } } + +// The starvation clock's anchor must be the oldest *pending, unassigned* job: +// the scheduler arms its expiry wake from it, and a job already claimed by a +// worker is never swept, so a claimed job must not pull the wake earlier than +// the deadline the sweep would really act on. +func TestEarliestPendingCreatedAt(t *testing.T) { + database, cleanup := setupDBTest(t) + defer cleanup() + + if _, ok, err := database.EarliestPendingCreatedAt(); err != nil || ok { + t.Fatalf("EarliestPendingCreatedAt with no jobs = (ok %v, err %v), want (false, nil)", ok, err) + } + + worker, err := database.CreateWorker("", "anchor-worker", protocol.WorkerCapabilities{ + Encoders: []string{"libx264"}, + FFmpegVersion: "5.0", + MaxConcurrent: 1, + }) + if err != nil { + t.Fatalf("CreateWorker: %v", err) + } + + // Claimed (queued) job, older than the pending one. + claimed, err := database.CreateJob(`["in.mkv"]`, `["-c:v","libx264"]`, "out.mkv", false) + if err != nil { + t.Fatalf("CreateJob: %v", err) + } + if _, err := database.GetDB().Exec(`UPDATE jobs SET created_at = ? WHERE id = ?`, + time.Now().Add(-time.Hour), claimed.ID); err != nil { + t.Fatalf("backdate claimed job: %v", err) + } + if err := database.AssignJobToWorker(claimed.ID, worker.ID); err != nil { + t.Fatalf("AssignJobToWorker: %v", err) + } + + waiting, err := database.CreateJob(`["in.mkv"]`, `["-c:v","libx264"]`, "out.mkv", false) + if err != nil { + t.Fatalf("CreateJob: %v", err) + } + + anchor, ok, err := database.EarliestPendingCreatedAt() + if err != nil { + t.Fatalf("EarliestPendingCreatedAt: %v", err) + } + if !ok { + t.Fatal("EarliestPendingCreatedAt reported no pending job while one exists") + } + if !anchor.Equal(waiting.CreatedAt) { + t.Errorf("anchor = %s, want the pending job's created_at %s (the older claimed job must not count)", anchor, waiting.CreatedAt) + } +} + +// The anchor must be the *oldest* pending unassigned job, not the newest: +// the scheduler arms its expiry wake from the anchor, and the sweep fails +// every pending job past the cutoff, so arming for a newer job would postpone +// the older jobs' verdict past the acceptance window. ORDER BY must therefore +// be ASC; a DESC regression must fail this test. +func TestEarliestPendingCreatedAtSelectsOldest(t *testing.T) { + database, cleanup := setupDBTest(t) + defer cleanup() + + oldest, err := database.CreateJob(`["in.mkv"]`, `["-c:v","libx264"]`, "out.mkv", false) + if err != nil { + t.Fatalf("CreateJob oldest: %v", err) + } + newest, err := database.CreateJob(`["in.mkv"]`, `["-c:v","libx264"]`, "out.mkv", false) + if err != nil { + t.Fatalf("CreateJob newest: %v", err) + } + + // Separate the two created_at values far enough that a wrong ordering is + // unambiguous even under a coarse clock. + backdated := time.Now().Add(-time.Hour) + if _, err := database.GetDB().Exec(`UPDATE jobs SET created_at = ? WHERE id = ?`, backdated, oldest.ID); err != nil { + t.Fatalf("backdate oldest: %v", err) + } + + anchor, ok, err := database.EarliestPendingCreatedAt() + if err != nil { + t.Fatalf("EarliestPendingCreatedAt: %v", err) + } + if !ok { + t.Fatal("EarliestPendingCreatedAt reported no pending job while two exist") + } + if !anchor.Equal(backdated) { + t.Errorf("anchor = %s, want the OLDEST pending job's created_at %s (newest job %s must not be chosen)", + anchor, backdated, newest.ID) + } +} diff --git a/pkg/server/scheduler/scheduler.go b/pkg/server/scheduler/scheduler.go index a9b6c3c..713f8ea 100644 --- a/pkg/server/scheduler/scheduler.go +++ b/pkg/server/scheduler/scheduler.go @@ -20,6 +20,16 @@ import ( // generic no-worker message so the crash semantics stay observable. const crashStarvationErrMsg = "worker crashed (heartbeat timeout) and no surviving worker to take over the job" +// starvationExpiryMargin is how far past a pending job's no-worker deadline +// the expiry wake runs the sweep. The sweep's cutoff is strict (a job must be +// pending for longer than NoWorkerJobTimeout), so a wake landing exactly on +// the deadline would judge nothing and hand the verdict back to the next +// timeout check — the quantization this wake exists to remove. The margin is +// three orders of magnitude below the default 30s check interval, so the +// verdict stays far inside the (120s, 150s] acceptance window (AC-17) and well +// before the CLI's own no-worker give-up at deadline+check interval. +const starvationExpiryMargin = time.Second + type Config struct { JobTimeout time.Duration // Maximum time a job can run before being considered timed out ScheduleInterval time.Duration // Interval for scheduling checks @@ -155,6 +165,17 @@ func (s *Scheduler) run() { timeoutTicker := time.NewTicker(s.config.TimeoutCheckInterval) defer timeoutTicker.Stop() + // starvationWake fires once at the oldest pending job's no-worker deadline. + // The sweep on the timeout tick alone quantizes the verdict to that tick's + // phase, which spreads it over the whole (120s, 150s] acceptance window and + // can leave the CLI giving up (cancelling) before the server's verdict is + // written. Armed by armStarvationWake; stopped while no verdict is due. + starvationWake := time.NewTimer(time.Hour) + starvationWake.Stop() + defer starvationWake.Stop() + + s.armStarvationWake(starvationWake) + for { select { case <-s.stop: @@ -165,13 +186,72 @@ func (s *Scheduler) run() { case <-s.resched: // Immediate rescheduling triggered (e.g., after worker offline migration) s.schedulePendingJobs() + s.armStarvationWake(starvationWake) case <-timeoutTicker.C: s.checkTimeouts() s.checkNoWorkerStarvation() + s.armStarvationWake(starvationWake) + case <-starvationWake.C: + // A pending job's no-worker clock just expired: judge it now rather + // than at the next timeout tick. The sweep is idempotent and keeps + // its live-worker short-circuit, so an early wake only makes the + // verdict's timing exact. + s.checkNoWorkerStarvation() + s.armStarvationWake(starvationWake) } } } +// armStarvationWake (re)points the expiry wake at the oldest pending job's +// no-worker deadline. It runs at loop start, on every timeout tick, after +// each wake, and after a reschedule that may have requeued jobs. +// +// With at least one live schedulable worker the sweep can never judge, so the +// wake stays stopped and the pending backlog is not scanned — the same +// short-circuit checkNoWorkerStarvation applies before its own query. +// +// A deadline already in the past is left to the timeout tick: only a sweep +// that has not run yet (loop start) leaves one, and re-arming for "now" would +// just re-run it at tick frequency. +func (s *Scheduler) armStarvationWake(timer *time.Timer) { + if !s.noWorkerEnabled { + return + } + // Stop and drain before reading: a value delivered while the select took + // another ready branch would otherwise wake the loop again immediately. + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + + schedulable, err := s.db.GetLiveSchedulableWorkers(s.config.HeartbeatFreshness) + if err != nil { + log.Printf("Scheduler: Failed to check schedulable workers: %v", err) + return + } + if len(schedulable) > 0 { + // A live worker can still take pending jobs, so no starvation verdict + // can be due. + return + } + + anchor, ok, err := s.db.EarliestPendingCreatedAt() + if err != nil { + log.Printf("Scheduler: Failed to read the earliest pending job: %v", err) + return + } + if !ok { + return + } + wait := time.Until(anchor.Add(s.config.NoWorkerJobTimeout + starvationExpiryMargin)) + if wait <= 0 { + return + } + timer.Reset(wait) +} + // TriggerReschedule triggers an immediate scheduling pass. // This is called by the worker health monitor after jobs have been migrated from an offline worker. func (s *Scheduler) TriggerReschedule() { diff --git a/pkg/server/scheduler/scheduler_starvation_wake_test.go b/pkg/server/scheduler/scheduler_starvation_wake_test.go new file mode 100644 index 0000000..f00dfc5 --- /dev/null +++ b/pkg/server/scheduler/scheduler_starvation_wake_test.go @@ -0,0 +1,237 @@ +package scheduler + +import ( + "testing" + "time" + + "github.com/tsic404/rffmpeg/pkg/protocol" + "github.com/tsic404/rffmpeg/pkg/server/db" +) + +// offlineWorkerFixture creates the worker row the starvation sweep's +// live-worker guard needs to see a cluster with nothing schedulable. +func offlineWorkerFixture(t *testing.T, database *db.Database) { + t.Helper() + if _, err := database.CreateWorker("worker-1", "test-worker", protocol.WorkerCapabilities{ + Encoders: []string{"libx264"}, + FFmpegVersion: "5.0", + MaxConcurrent: 1, + }); err != nil { + t.Fatalf("Failed to create worker: %v", err) + } + if err := database.UpdateWorkerStatus("worker-1", protocol.WorkerStatusOffline); err != nil { + t.Fatalf("Failed to mark worker offline: %v", err) + } +} + +// backdatePendingAnchor rewrites a pending job's created_at, the clock the +// starvation sweep measures the no-worker timeout from. +func backdatePendingAnchor(t *testing.T, database *db.Database, jobID string, at time.Time) { + t.Helper() + if _, err := database.GetDB().Exec(`UPDATE jobs SET created_at = ? WHERE id = ?`, at, jobID); err != nil { + t.Fatalf("Failed to backdate created_at: %v", err) + } +} + +func pendingJob(t *testing.T, database *db.Database) *db.Job { + t.Helper() + job, err := database.CreateJob(`["file1.mp4"]`, `["-i", "input.mp4"]`, "output.mp4", false) + if err != nil { + t.Fatalf("Failed to create job: %v", err) + } + return job +} + +func assertStarvedNoWorker(t *testing.T, database *db.Database, jobID string) { + t.Helper() + updated, err := database.GetJob(jobID) + if err != nil { + t.Fatalf("Failed to get job: %v", err) + } + if updated.Status != protocol.JobStatusFailed { + t.Errorf("Expected status failed, got %s", updated.Status) + } + if updated.FailureType != string(protocol.FailureNoWorkerAvailable) { + t.Errorf("Expected failure_type %q, got %q", protocol.FailureNoWorkerAvailable, updated.FailureType) + } +} + +// The no-worker verdict must land just past the pending clock's deadline, not +// at whatever phase the timeout tick happens to be in: the tick-only spread — +// anywhere in (deadline, deadline+interval] — is what let the CLI give up and +// cancel a job the server was about to fail as WORKER_CRASH (AC-17). An +// hour-long check interval makes the tick irrelevant here, so only the expiry +// wake can produce a verdict at all. +func TestStarvationWakeJudgesJustPastDeadline(t *testing.T) { + database, err := db.New(":memory:") + if err != nil { + t.Fatalf("Failed to create database: %v", err) + } + defer database.Close() + + const noWorkerTimeout = 3 * time.Second + s := New(database, Config{ + ScheduleInterval: time.Hour, + TimeoutCheckInterval: time.Hour, // the expiry wake is the only trigger + MaxJobsPerWorker: 1, + NoWorkerJobTimeout: noWorkerTimeout, + HeartbeatFreshness: 90 * time.Second, + }) + + offlineWorkerFixture(t, database) + job := pendingJob(t, database) + + notifyCh, cancel := database.JobNotifier().Subscribe(job.ID) + defer cancel() + + s.Start() + defer s.Stop() + + // Anchor the pending clock only after the loop is running: a deadline + // computed before Start can already be past by the time the loop arms the + // wake under CI load, silently falling back to the hour-long tick. The + // reschedule trigger re-arms from the freshly backdated anchor. + deadline := time.Now().Add(2 * time.Second) + backdatePendingAnchor(t, database, job.ID, deadline.Add(-noWorkerTimeout)) + s.TriggerReschedule() + + var verdictAt time.Time + select { + case <-notifyCh: + verdictAt = time.Now() + case <-time.After(30 * time.Second): + t.Fatalf("expiry wake did not judge the job (deadline was %s after start)", time.Until(deadline)) + } + + if !verdictAt.After(deadline) { + t.Errorf("verdict at %s, before the pending clock's deadline %s: the sweep must not judge a job that has waited no longer than the no-worker timeout", verdictAt, deadline) + } + if bound := deadline.Add(starvationExpiryMargin + 4*time.Second); verdictAt.After(bound) { + t.Errorf("verdict at %s, more than %s past the deadline %s: it fell back to the check interval instead of the expiry wake", verdictAt, starvationExpiryMargin+4*time.Second, deadline) + } + assertStarvedNoWorker(t, database, job.ID) +} + +// A job whose deadline elapsed before the scheduler started has no future +// instant to wake on; the timeout tick must still judge it. The expiry wake is +// an accelerator, never the only path to a verdict. +func TestStarvationWakeFallsBackToTickForAnExpiredDeadline(t *testing.T) { + database, err := db.New(":memory:") + if err != nil { + t.Fatalf("Failed to create database: %v", err) + } + defer database.Close() + + s := New(database, Config{ + ScheduleInterval: time.Hour, + TimeoutCheckInterval: 500 * time.Millisecond, + MaxJobsPerWorker: 1, + NoWorkerJobTimeout: time.Minute, + }) + + offlineWorkerFixture(t, database) + job := pendingJob(t, database) + backdatePendingAnchor(t, database, job.ID, time.Now().Add(-10*time.Minute)) + + notifyCh, cancel := database.JobNotifier().Subscribe(job.ID) + defer cancel() + + s.Start() + defer s.Stop() + + select { + case <-notifyCh: + case <-time.After(10 * time.Second): + t.Fatal("timeout tick did not judge a job whose no-worker deadline was already past at start") + } + assertStarvedNoWorker(t, database, job.ID) +} + +// While a live schedulable worker exists the sweep can never judge, so the +// wake must stay disarmed: arming it would run the unindexed anchor lookup on +// every tick for a verdict that cannot happen. +func TestStarvationWakeSkippedWhileWorkerIsLive(t *testing.T) { + database, err := db.New(":memory:") + if err != nil { + t.Fatalf("Failed to create database: %v", err) + } + defer database.Close() + + s := New(database, Config{ + ScheduleInterval: time.Hour, + TimeoutCheckInterval: time.Hour, + MaxJobsPerWorker: 1, + NoWorkerJobTimeout: time.Minute, + HeartbeatFreshness: 90 * time.Second, + }) + + if _, err := database.CreateWorker("worker-1", "test-worker", protocol.WorkerCapabilities{ + Encoders: []string{"libx264"}, + FFmpegVersion: "5.0", + MaxConcurrent: 1, + }); err != nil { + t.Fatalf("Failed to create worker: %v", err) + } + // The pending job's deadline is in the future, so only the live-worker + // short-circuit keeps the wake from being armed. + pendingJob(t, database) + + timer := time.NewTimer(time.Hour) + timer.Stop() + s.armStarvationWake(timer) + + if timer.Stop() { + t.Fatal("wake was armed while a live schedulable worker exists: the sweep cannot judge, so the anchor lookup is wasted") + } +} + +// Two pending jobs with different deadlines: the wake must be armed for the +// OLDEST one's deadline, because the sweep fails every pending job past the +// cutoff. Arming for the newest would let the oldest starve until the next +// check tick — with the hour-long tick here, never — so this test fails under +// a DESC-ordered anchor. +func TestStarvationWakeJudgesOldestOfMultiplePending(t *testing.T) { + database, err := db.New(":memory:") + if err != nil { + t.Fatalf("Failed to create database: %v", err) + } + defer database.Close() + + s := New(database, Config{ + ScheduleInterval: time.Hour, + TimeoutCheckInterval: time.Hour, // the expiry wake is the only trigger + MaxJobsPerWorker: 1, + NoWorkerJobTimeout: 3 * time.Second, + HeartbeatFreshness: 90 * time.Second, + }) + + offlineWorkerFixture(t, database) + oldest := pendingJob(t, database) + // The newest job expires a minute out; a wake armed for it (a DESC-ordered + // anchor) would judge the oldest only after the newest deadline, which this + // 20s window cannot observe. + newest := pendingJob(t, database) + + notifyCh, cancel := database.JobNotifier().Subscribe(oldest.ID) + defer cancel() + + s.Start() + defer s.Stop() + + deadline := time.Now().Add(2 * time.Second) + backdatePendingAnchor(t, database, oldest.ID, deadline.Add(-s.config.NoWorkerJobTimeout)) + backdatePendingAnchor(t, database, newest.ID, time.Now().Add(57*time.Second)) + s.TriggerReschedule() + + select { + case <-notifyCh: + case <-time.After(20 * time.Second): + t.Fatalf("the oldest pending job was not judged within 20s: the wake must be armed for the earliest pending deadline, not a newer one") + } + + bound := deadline.Add(starvationExpiryMargin + 4*time.Second) + if verdictAt := time.Now(); verdictAt.After(bound) { + t.Errorf("verdict at %s, more than %s past the oldest job's deadline %s: it fell back to the check interval instead of the expiry wake", verdictAt, starvationExpiryMargin+4*time.Second, deadline) + } + assertStarvedNoWorker(t, database, oldest.ID) +}