filed

Job queue using FUSE

git clone git://mccd.space/filed

commit 051df985a2adabdb1cb41ff11981fb3c7b738109
parent 8100148abe21cbba9e9bd7332702f3239942c1be
Author: Marc Coquand <marc@coquand.email>
Date:   Mon, 15 Dec 2025 17:07:11 +0100

Max attempts ensure

Diffstat:
M
manager.go
 | 
15
+++++++++------

1 file changed, 9 insertions(+), 6 deletions(-)
diff --git a/manager.go b/manager.go
@@ -55,22 +55,24 @@ func (jm *JobManager) StartWorker(ctx context.Context) {
 
 func (jm *JobManager) processPendingJobs() {
 	conf := jm.store.GetConfig()
-	jobs, err := jm.store.ListJobsByState(store.StatePending)
-	if err != nil {
-		slog.Error("Worker: Failed to list jobs", "error", err)
-		return
-	}
 	activeJobs, err := jm.store.ListJobsByState(store.StateRunning)
 	if err != nil {
 		slog.Error("Worker: Failed to list jobs", "error", err)
 		return
 	}
+	activeJobCount := len(activeJobs)
 
-	if len(activeJobs) >= conf.MaxJobCount {
+	jobs, err := jm.store.ListJobsByState(store.StatePending)
+	if err != nil {
+		slog.Error("Worker: Failed to list jobs", "error", err)
 		return
 	}
 
 	for _, job := range jobs {
+		if activeJobCount >= conf.MaxJobCount {
+			return
+		}
+
 		if job.Attempts >= conf.MaxAttempts {
 			slog.Warn("Worker: job exceeded max attempts. Moving to failed", "Job", job.ID)
 			jm.store.FailJob(job.ID)
@@ -83,6 +85,7 @@ func (jm *JobManager) processPendingJobs() {
 		}
 
 		go jm.runJob(job.ID, job.Command)
+		activeJobCount++
 	}
 }