TSI-3766: fix(scheduler): starvation sweep WORKER_CRASH 判定时刻跨 run 抖动，(120s,150s] 窗口边界敏感

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)

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)
+	}
+}

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() {