From cd82e663ad3c1f1e40ea7a69e7fb34cddf1f8a90 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 20:45:39 -0800 Subject: [PATCH] admin: credit an unitemized slot to a pending job only after a later heartbeat (#12385) * admin: credit an unitemized slot to a pending job only after a later heartbeat A pending reservation the worker does not list took one of the slots the worker counts without itemizing, even when that heartbeat came before the reservation and could not have counted the job. The slot belonged to unknown work, so the admin undercounted it and could admit one job too many, past a job type's per-worker limit, until the next heartbeat. A reservation now records when it was made, and claims an unitemized slot only from a heartbeat received at or after that time. Co-Authored-By: Claude Opus 5.5 * admin: time unitemized-slot credit by the heartbeat, not the last hello A reconnect's hello moves LastSeenAt but keeps the old heartbeat, so a pending reservation could claim a slot counted before it existed, and a completed job's echo could end without a new heartbeat. Record when the heartbeat itself arrived and compare reservations against that. Co-Authored-By: Claude Opus 5.5 * admin: a completed job claims no slot from a heartbeat before its reservation A job that completes before any heartbeat counts it has no echo, so the slot the last heartbeat counted is other work. Crediting it to the job's type let another job type pass its per-worker limit until the next heartbeat. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/admin/plugin/plugin_scheduler.go | 19 +++--- weed/admin/plugin/plugin_scheduler_test.go | 72 ++++++++++++++++++++++ weed/admin/plugin/registry.go | 5 +- 3 files changed, 87 insertions(+), 9 deletions(-) diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index ceed99b7e6..5d80254d90 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -1220,6 +1220,7 @@ type execReservation struct { workerID string jobType string jobID string + reservedAt time.Time releasedAt time.Time } @@ -1282,9 +1283,10 @@ func (l executorLoad) holders() string { // jobType "" counts work of every type, otherwise only that type. A // heartbeat entry whose job id matches a reservation is the same slot seen // twice and counts once. Slots the worker counts without itemizing them go -// first to this admin's jobs it does not list, a completed one only until the -// worker's next heartbeat; the rest are of unknown type and count against -// every type. Caller holds r.execResMu. +// first to this admin's jobs it does not list, and only from a heartbeat +// that came after the job's reservation: a pending job's from then on, a +// completed job's until the worker's next heartbeat. The rest are of unknown +// type and count against every type. Caller holds r.execResMu. func (r *Plugin) executorLoadLocked(executor *WorkerSession, jobType string) executorLoad { load := executorLoad{ reservedLanes: make(map[SchedulerLane]int), @@ -1330,13 +1332,13 @@ func (r *Plugin) executorLoadLocked(executor *WorkerSession, jobType string) exe } switch { case !released: - if !isListed { + if !isListed && !executor.heartbeatAt.Before(res.reservedAt) { unitemized-- } countReserved(res) case isListed: countReserved(res) - case executor.LastSeenAt.Before(res.releasedAt): + case !executor.heartbeatAt.Before(res.reservedAt) && executor.heartbeatAt.Before(res.releasedAt): unlistedEchoes = append(unlistedEchoes, res) } } @@ -1442,9 +1444,10 @@ func (r *Plugin) reserveExecutorSlot( r.execResSeq++ token := r.execResSeq r.execRes[token] = execReservation{ - workerID: strings.TrimSpace(executor.WorkerID), - jobType: jobType, - jobID: jobID, + workerID: strings.TrimSpace(executor.WorkerID), + jobType: jobType, + jobID: jobID, + reservedAt: time.Now(), } release := func() { r.execResMu.Lock() diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index d5ce9e77fe..cd1622c67f 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -1656,6 +1656,78 @@ func TestExecutorCapacityStopsCreditingACompletedJobAfterTheNextHeartbeat(t *tes } } +// A count-only heartbeat taken before a reservation cannot count that job, +// so the slot it counted is still unknown work rather than the new job's. +// A reconnect's hello keeps that heartbeat and does not change this. +func TestExecutorCapacityKeepsSlotsCountedBeforeAReservation(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + hello := &plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + } + pluginSvc.registry.UpsertFromHello(hello) + policy := schedulerPolicy{PerWorkerConcurrency: 2} + oneUsed := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4} + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + if _, _, err := pluginSvc.reserveExecutorByID("vacuum", policy, workerID, "vacuum-1"); err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + pluginSvc.registry.UpsertFromHello(hello) + + executor, _ := pluginSvc.registry.Get(workerID) + if pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("the slot counted before vacuum-1 was reserved may hold vacuum work") + } + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + executor, _ = pluginSvc.registry.Get(workerID) + if !pluginSvc.executorHasCapacity(executor, "vacuum", policy) { + t.Fatalf("a heartbeat after the reservation counts vacuum-1 itself") + } +} + +// A job that completes before any heartbeat counts it has no echo: the slot +// the last heartbeat counted is still unknown work, not the job's. +func TestExecutorCapacityKeepsSlotsCountedBeforeACompletedJob(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-a" + pluginSvc.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: workerID, + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "vacuum", CanExecute: true, MaxExecutionConcurrency: 4}, + {JobType: "balance", CanExecute: true, MaxExecutionConcurrency: 4}, + }, + }) + oneUsed := &plugin_pb.WorkerHeartbeat{WorkerId: workerID, ExecutionSlotsUsed: 1, ExecutionSlotsTotal: 4} + pluginSvc.registry.UpdateHeartbeat(workerID, oneUsed) + _, release, err := pluginSvc.reserveExecutorByID("vacuum", schedulerPolicy{PerWorkerConcurrency: 2}, workerID, "vacuum-1") + if err != nil { + t.Fatalf("reserve vacuum-1: %v", err) + } + release() + + executor, _ := pluginSvc.registry.Get(workerID) + if pluginSvc.executorHasCapacity(executor, "balance", schedulerPolicy{PerWorkerConcurrency: 1}) { + t.Fatalf("the slot counted before vacuum-1 was reserved may hold balance work") + } +} + // The reason a dispatch gives up names every holder of the worker's slots, // the unitemized ones included. func TestCapacityReasonNamesUnitemizedSlots(t *testing.T) { diff --git a/weed/admin/plugin/registry.go b/weed/admin/plugin/registry.go index 4fa141fb06..3ff48ab397 100644 --- a/weed/admin/plugin/registry.go +++ b/weed/admin/plugin/registry.go @@ -24,6 +24,8 @@ type WorkerSession struct { LastSeenAt time.Time Capabilities map[string]*plugin_pb.JobTypeCapability Heartbeat *plugin_pb.WorkerHeartbeat + // heartbeatAt is when Heartbeat arrived; a hello moves LastSeenAt alone. + heartbeatAt time.Time } // Registry tracks connected plugin workers and capability-based selection. @@ -91,7 +93,8 @@ func (r *Registry) UpdateHeartbeat(workerID string, heartbeat *plugin_pb.WorkerH return } session.Heartbeat = cloneWorkerHeartbeat(heartbeat) - session.LastSeenAt = time.Now() + session.heartbeatAt = time.Now() + session.LastSeenAt = session.heartbeatAt } func (r *Registry) Get(workerID string) (*WorkerSession, bool) {