fix: job queue panic recover (#219)

kaytu-io · Jun 18, 2024 · fa9a485 · fa9a485
1 parent 321339b
commit fa9a485
Showing 1 changed file with 18 additions and 12 deletions.
diff --git a/pkg/plugin/sdk/job_queue.go b/pkg/plugin/sdk/job_queue.go
@@ -53,21 +53,27 @@ func (q *JobQueue) Push(job Job) {
 	q.queue <- job
 }
 
-func (q *JobQueue) Start() {
-	for i := 0; i < q.maxConcurrent; i++ {
-		go q.run()
+func (q *JobQueue) finisher() {
+	if err := recover(); err != nil {
+		log.Printf("Job queue finisher panic: %v", err)
+		go q.finisher()
 	}
 
-	go func() {
-		for {
-			if q.finishedCounter.Load() == q.pendingCounter.Load() && q.onFinish != nil {
-				time.Sleep(500 * time.Millisecond)
-				log.Printf("All jobs are finished - calling onFinish, job counts: %d/%d", q.finishedCounter.Load(), q.pendingCounter.Load())
-				q.onFinish()
-			}
+	for {
+		if q.finishedCounter.Load() == q.pendingCounter.Load() && q.onFinish != nil {
 			time.Sleep(500 * time.Millisecond)
+			log.Printf("All jobs are finished - calling onFinish, job counts: %d/%d", q.finishedCounter.Load(), q.pendingCounter.Load())
+			q.onFinish()
 		}
-	}()
+		time.Sleep(500 * time.Millisecond)
+	}
+}
+
+func (q *JobQueue) Start() {
+	for i := 0; i < q.maxConcurrent; i++ {
+		go q.run()
+	}
+	go q.finisher()
 }
 
 func (q *JobQueue) SetOnFinish(f func()) {
@@ -101,7 +107,7 @@ func (q *JobQueue) run() {
 			log.Printf("Finished job %s", job.Id())
 		}
 
-		q.stream.Send(&golang.PluginMessage{
+		_ = q.stream.Send(&golang.PluginMessage{
 			PluginMessage: &golang.PluginMessage_Job{
 				Job: jobResult,
 			},