From da2dfc7aee027c2189c676761a437056e7cab355 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ullrich=20Sch=C3=A4fer?= Date: Wed, 4 Nov 2026 14:19:51 +0100 Subject: [PATCH] fix(helpers): record a failed task's error before WaitAsync returns SubmitBlocking marks a task done in a deferred waitGroup.Done(), and a failed task reports its error by panicking into the ants pool's panic handler. Go runs the task's own defers while the panic unwinds, so Done() runs before ants recovers and calls the handler. WaitAsync can wake up, find no errors yet and return nil; the error is recorded a moment later. A collector then counts as successful, and StatefulApiCollector saves its incremental state. The API client is shared by all subtasks of a task, so the lost error is reported by the next subtask instead. On our instance a CircleCI token lost access and every request got a 404: on most nights collectPipelines passed and moved its `since` marker forward, the error appeared in collectWorkflows, and two weeks of pipelines were never collected again. Recover the panic inside the task and record it before Done(). Co-Authored-By: Claude Opus 5.5 --- .../pluginhelper/api/worker_scheduler.go | 9 ++++++++- .../pluginhelper/api/worker_scheduler_test.go | 18 ++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) 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() + } +}