fix(helpers): record a failed task's error before WaitAsync returns

backend/helpers/pluginhelper/api/worker_scheduler.go

@@ -93,7 +93,14 @@ func (s *WorkerScheduler) SubmitBlocking(task func() errors.Error) {
 	}
 	s.waitGroup.Add(1)
 	s.checkError(s.pool.Submit(func() {
-		defer s.waitGroup.Done()
+		// Record a panic before Done: the pool's panic handler runs only after
+		// this function has returned, so WaitAsync could return nil first.
+		defer func() {
+			if r := recover(); r != nil {
+				s.checkError(r)
+			}
+			s.waitGroup.Done()
+		}()
 
 		id := atomic.AddInt32(&s.counter, 1)
 		s.logger.Debug("schedulerJob >>> %d started", id)

backend/helpers/pluginhelper/api/worker_scheduler_test.go

@@ -64,3 +64,21 @@ func TestWorkerSchedulerQpsControl(t *testing.T) {
 	}
 	cancel()
 }
+
+// A failed task must be reported by the WaitAsync call that waits for it, not
+// by a later one: a stateful collector saves its state when WaitAsync
+// returns nil.
+func TestWorkerSchedulerWaitAsyncReturnsTaskError(t *testing.T) {
+	for i := 0; i < 1000; i++ {
+		s, err := NewWorkerScheduler(context.Background(), 1, time.Microsecond, unithelper.DummyLogger())
+		assert.Nil(t, err)
+		s.SubmitBlocking(func() errors.Error {
+			return errors.Default.New("request failed")
+		})
+		if !assert.NotNil(t, s.WaitAsync(), "run %d: WaitAsync returned before the task error was recorded", i) {
+			s.Release()
+			return
+		}
+		s.Release()
+	}
+}