From d83efe24ef051786d6839b2c920fb70e3ef27db2 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Tue, 7 Apr 2026 00:03:46 +0700 Subject: [PATCH] refactor: optimize job listing and counting logic for concurrency safety. Separate tenant filtering and status checks to reduce lock contention, improving performance and ensuring accurate job status retrieval. --- internal/jobs/job.go | 43 ++++++++++++++++++++++++++++++------------- 1 file changed, 30 insertions(+), 13 deletions(-) diff --git a/internal/jobs/job.go b/internal/jobs/job.go index b33f5a3..da70f73 100644 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -229,22 +229,32 @@ func (r *Registry) List(tenantID, statusFilter, kindFilter, cursor string, limit if limit <= 0 { limit = 50 } + // Build candidate set under r.mu, but read mutable status fields (j.Status) outside + // of r.mu so we don't race with Job.mu-protected writes (MarkRunning/Succeed/Fail). r.mu.RLock() - var all []*Job + all := make([]*Job, 0, len(r.byID)) for _, j := range r.byID { - if j.TenantID != tenantID { - continue + if j.TenantID == tenantID { + all = append(all, j) } - if statusFilter != "" && j.Status != statusFilter { - continue - } - if kindFilter != "" && j.Kind != kindFilter { - continue - } - all = append(all, j) } r.mu.RUnlock() + // Apply filters outside of r.mu to synchronize with Job.mu. + if statusFilter != "" || kindFilter != "" { + filtered := all[:0] + for _, j := range all { + if kindFilter != "" && j.Kind != kindFilter { + continue + } + if statusFilter != "" && j.statusLocked() != statusFilter { + continue + } + filtered = append(filtered, j) + } + all = filtered + } + sort.Slice(all, func(i, j int) bool { return all[i].CreatedAt.After(all[j].CreatedAt) }) @@ -273,9 +283,9 @@ func (r *Registry) CountOtherActiveModuleRefresh(tenantID, excludeJobID string) if r == nil { return 0 } + // Two-phase: snapshot job pointers under r.mu, then read j.Status under Job.mu. r.mu.RLock() - defer r.mu.RUnlock() - n := 0 + candidates := make([]*Job, 0, 8) for _, j := range r.byID { if j.TenantID != tenantID || j.Kind != KindModuleRefresh { continue @@ -283,7 +293,14 @@ func (r *Registry) CountOtherActiveModuleRefresh(tenantID, excludeJobID string) if j.ID == excludeJobID { continue } - if j.Status == StatusQueued || j.Status == StatusRunning { + candidates = append(candidates, j) + } + r.mu.RUnlock() + + n := 0 + for _, j := range candidates { + st := j.statusLocked() + if st == StatusQueued || st == StatusRunning { n++ } }