diff --git a/backend/helpers/pluginhelper/api/worker_scheduler.go b/backend/helpers/pluginhelper/api/worker_scheduler.go index 7e64ec151aa..84a6e52aca9 100644 --- a/backend/helpers/pluginhelper/api/worker_scheduler.go +++ b/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) diff --git a/backend/helpers/pluginhelper/api/worker_scheduler_test.go b/backend/helpers/pluginhelper/api/worker_scheduler_test.go index e25c69b034e..4d562bef4fb 100644 --- a/backend/helpers/pluginhelper/api/worker_scheduler_test.go +++ b/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() + } +}