From d52beebb3d21d8956b322a6ee039ffdac3a62a18 Mon Sep 17 00:00:00 2001 From: Kemal Hadimli Date: Thu, 15 Oct 2026 11:58:37 +0100 Subject: [PATCH 1/4] fix: Don't drop work queued by the last active table in shuffle-queue --- scheduler/queue/active_work_signal.go | 47 ++++++------------- scheduler/queue/active_work_signal_test.go | 25 ++++++++++ scheduler/queue/scheduler.go | 53 +++++++++++----------- scheduler/queue/scheduler_test.go | 39 ++++++++++++++++ 4 files changed, 105 insertions(+), 59 deletions(-) create mode 100644 scheduler/queue/active_work_signal_test.go diff --git a/scheduler/queue/active_work_signal.go b/scheduler/queue/active_work_signal.go index f1c59d025c..dbb78aa658 100644 --- a/scheduler/queue/active_work_signal.go +++ b/scheduler/queue/active_work_signal.go @@ -1,9 +1,7 @@ package queue import ( - "sync" "sync/atomic" - "time" ) // activeWorkSignal is a thread-safe coordinator for awaiting a worker pool @@ -22,36 +20,33 @@ import ( // - When a worker finishes a task, call `Done()` (like a WaitGroup) // // - If the queue is empty, check `IsIdle()` to check if no workers are active. -// - If workers are still active, call `Wait()` to block until state changes. +// - If workers are still active, call `Wait()` to block until one of them finishes. type activeWorkSignal struct { - countChangeSignal *sync.Cond - activeWorkUnitCount *atomic.Int32 - isStarted *atomic.Bool + activeWorkUnitCount atomic.Int32 + isStarted atomic.Bool + // Buffered with capacity 1 so a Done that happens before Wait is kept rather than lost. + workDone chan struct{} } func newActiveWorkSignal() *activeWorkSignal { - return &activeWorkSignal{ - countChangeSignal: sync.NewCond(&sync.Mutex{}), - activeWorkUnitCount: &atomic.Int32{}, - isStarted: &atomic.Bool{}, - } + return &activeWorkSignal{workDone: make(chan struct{}, 1)} } // Add means the dispatcher has taken a task off the queue for a worker. -// -// Wake up the work queuing goroutine. func (s *activeWorkSignal) Add() { s.activeWorkUnitCount.Add(1) s.isStarted.Store(true) - s.countChangeSignal.Signal() } // Done means a worker has finished working on a task. // -// If the count became zero, wake up the work queuing goroutine (might have finished). +// Wake up the work queuing goroutine: the worker might have queued new tasks, or all work might have finished. func (s *activeWorkSignal) Done() { s.activeWorkUnitCount.Add(-1) - s.countChangeSignal.Signal() + select { + case s.workDone <- struct{}{}: + default: + } } // IsIdle returns true if no workers are active. If queue is empty and workers idle, done! @@ -59,22 +54,8 @@ func (s *activeWorkSignal) IsIdle() bool { return s.isStarted.Load() && s.activeWorkUnitCount.Load() <= 0 } -// Wait blocks until the count of active workers changes. +// Wait blocks until a worker finishes a task. It returns immediately if one has +// finished since the previous Wait returned. func (s *activeWorkSignal) Wait() { - // A race condition is possible when the last active table asynchronously - // queues a relation. The table finishes (calling `.Done()`) a moment - // before the queue receives the `.Push()`. At this point, the queue is - // empty and there are no active workers. - // - // A moment later, the queue receives the `.Push()` and queues a new task. - // - // This is a very infrequent case according to tests, but it happens. - time.Sleep(10 * time.Millisecond) - - if s.activeWorkUnitCount.Load() <= 0 { - return - } - s.countChangeSignal.L.Lock() - defer s.countChangeSignal.L.Unlock() - s.countChangeSignal.Wait() + <-s.workDone } diff --git a/scheduler/queue/active_work_signal_test.go b/scheduler/queue/active_work_signal_test.go new file mode 100644 index 0000000000..4c8129a042 --- /dev/null +++ b/scheduler/queue/active_work_signal_test.go @@ -0,0 +1,25 @@ +package queue + +import ( + "testing" + "time" +) + +func TestActiveWorkSignalWaitReturnsAfterEarlierDone(t *testing.T) { + s := newActiveWorkSignal() + s.Add() + s.Add() + s.Done() + + waited := make(chan struct{}) + go func() { + s.Wait() + close(waited) + }() + + select { + case <-waited: + case <-time.After(5 * time.Second): + t.Fatal("Wait missed a Done that happened before it was called") + } +} diff --git a/scheduler/queue/scheduler.go b/scheduler/queue/scheduler.go index 1a0241d1e7..a7447cc4dc 100644 --- a/scheduler/queue/scheduler.go +++ b/scheduler/queue/scheduler.go @@ -117,34 +117,35 @@ func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedR }) } - // Work distribution - go func() { - defer close(jobs) - for { - select { - case <-ctx.Done(): - return - default: - item := queue.Pop() - - // There is work to do. Mark it active before handing it off so the - // idle check below cannot fire while the item is in flight to a worker. - if item != nil { - activeWorkSignal.Add() - jobs <- item - continue - } - - // Queue is empty and no active work, done! - if activeWorkSignal.IsIdle() { - return - } - - // Queue is empty and there is active work, wait for changes + go dispatch(ctx, queue, jobs, activeWorkSignal) + + _ = workerPool.Wait() +} + +type workQueue interface { + Pop() *WorkUnit +} + +// dispatch hands queued work units to workers until the queue is empty and no work is active. +func dispatch(ctx context.Context, queue workQueue, jobs chan<- *WorkUnit, activeWorkSignal *activeWorkSignal) { + defer close(jobs) + for ctx.Err() == nil { + item := queue.Pop() + if item == nil { + if !activeWorkSignal.IsIdle() { activeWorkSignal.Wait() + continue + } + // Workers push relations before calling Done, so nothing can be pushed once idle, + // but the last active work unit may have pushed between the Pop above and the idle check. + if item = queue.Pop(); item == nil { + return } } - }() - _ = workerPool.Wait() + // Mark the item active before handing it off so the idle check cannot fire + // while the item is in flight to a worker. + activeWorkSignal.Add() + jobs <- item + } } diff --git a/scheduler/queue/scheduler_test.go b/scheduler/queue/scheduler_test.go index 8d9b71e4ec..11280f3175 100644 --- a/scheduler/queue/scheduler_test.go +++ b/scheduler/queue/scheduler_test.go @@ -99,3 +99,42 @@ func TestScheduler(t *testing.T) { // 2 top level tables, each with 10 resources, and each top level has a relation with 10 resources require.Len(t, gotResources, 2*10+2*10*10) } + +type pushOnFirstEmptyPopQueue struct { + *ConcurrentRandomQueue[WorkUnit] + onFirstEmptyPop func() +} + +func (q *pushOnFirstEmptyPopQueue) Pop() *WorkUnit { + item := q.ConcurrentRandomQueue.Pop() + if item == nil && q.onFirstEmptyPop != nil { + onEmpty := q.onFirstEmptyPop + q.onFirstEmptyPop = nil + onEmpty() + } + return item +} + +func TestDispatchHandsOffWorkQueuedByLastActiveUnit(t *testing.T) { + activeWorkSignal := newActiveWorkSignal() + queue := &pushOnFirstEmptyPopQueue{ConcurrentRandomQueue: NewConcurrentRandomQueue[WorkUnit](0, 1)} + queue.Push(WorkUnit{Table: &schema.Table{Name: "parent"}}) + // The parent's worker queues a relation and finishes after the dispatcher saw an + // empty queue but before it checked for active work. + queue.onFirstEmptyPop = func() { + queue.Push(WorkUnit{Table: &schema.Table{Name: "relation"}}) + activeWorkSignal.Done() + } + + jobs := make(chan *WorkUnit) + go dispatch(context.Background(), queue, jobs, activeWorkSignal) + + require.Equal(t, "parent", (<-jobs).Table.Name) + relation, ok := <-jobs + require.True(t, ok, "dispatcher stopped without handing off the queued relation") + require.Equal(t, "relation", relation.Table.Name) + activeWorkSignal.Done() + + _, ok = <-jobs + require.False(t, ok) +} From 940549bc7a3a47e63646126e5f7a62e617c74c06 Mon Sep 17 00:00:00 2001 From: Kemal Hadimli Date: Thu, 15 Oct 2026 12:38:37 +0100 Subject: [PATCH 2/4] fix: Wake the shuffle-queue dispatcher when a table queues a relation --- scheduler/queue/active_work_signal.go | 61 -------------- scheduler/queue/active_work_signal_test.go | 25 ------ scheduler/queue/scheduler.go | 41 +++------- scheduler/queue/scheduler_test.go | 39 --------- scheduler/queue/work_queue.go | 76 +++++++++++++++++ scheduler/queue/work_queue_test.go | 94 ++++++++++++++++++++++ scheduler/queue/worker.go | 12 +-- 7 files changed, 183 insertions(+), 165 deletions(-) delete mode 100644 scheduler/queue/active_work_signal.go delete mode 100644 scheduler/queue/active_work_signal_test.go create mode 100644 scheduler/queue/work_queue.go create mode 100644 scheduler/queue/work_queue_test.go diff --git a/scheduler/queue/active_work_signal.go b/scheduler/queue/active_work_signal.go deleted file mode 100644 index dbb78aa658..0000000000 --- a/scheduler/queue/active_work_signal.go +++ /dev/null @@ -1,61 +0,0 @@ -package queue - -import ( - "sync/atomic" -) - -// activeWorkSignal is a thread-safe coordinator for awaiting a worker pool -// that relies on a queue that might be temporarily empty. -// -// If queue is empty and workers idle, done! -// -// If the queue is empty but a worker is working on a task, we must wait and check -// if it's empty after that worker finishes. That's why we need this. -// -// Use it like this: -// -// - When the dispatcher takes a task off the queue to hand it to a worker, call `Add()` (like a WaitGroup). -// Marking the task active before the handoff ensures the idle check cannot fire while -// the task is in flight between the queue and a worker. -// - When a worker finishes a task, call `Done()` (like a WaitGroup) -// -// - If the queue is empty, check `IsIdle()` to check if no workers are active. -// - If workers are still active, call `Wait()` to block until one of them finishes. -type activeWorkSignal struct { - activeWorkUnitCount atomic.Int32 - isStarted atomic.Bool - // Buffered with capacity 1 so a Done that happens before Wait is kept rather than lost. - workDone chan struct{} -} - -func newActiveWorkSignal() *activeWorkSignal { - return &activeWorkSignal{workDone: make(chan struct{}, 1)} -} - -// Add means the dispatcher has taken a task off the queue for a worker. -func (s *activeWorkSignal) Add() { - s.activeWorkUnitCount.Add(1) - s.isStarted.Store(true) -} - -// Done means a worker has finished working on a task. -// -// Wake up the work queuing goroutine: the worker might have queued new tasks, or all work might have finished. -func (s *activeWorkSignal) Done() { - s.activeWorkUnitCount.Add(-1) - select { - case s.workDone <- struct{}{}: - default: - } -} - -// IsIdle returns true if no workers are active. If queue is empty and workers idle, done! -func (s *activeWorkSignal) IsIdle() bool { - return s.isStarted.Load() && s.activeWorkUnitCount.Load() <= 0 -} - -// Wait blocks until a worker finishes a task. It returns immediately if one has -// finished since the previous Wait returned. -func (s *activeWorkSignal) Wait() { - <-s.workDone -} diff --git a/scheduler/queue/active_work_signal_test.go b/scheduler/queue/active_work_signal_test.go deleted file mode 100644 index 4c8129a042..0000000000 --- a/scheduler/queue/active_work_signal_test.go +++ /dev/null @@ -1,25 +0,0 @@ -package queue - -import ( - "testing" - "time" -) - -func TestActiveWorkSignalWaitReturnsAfterEarlierDone(t *testing.T) { - s := newActiveWorkSignal() - s.Add() - s.Add() - s.Done() - - waited := make(chan struct{}) - go func() { - s.Wait() - close(waited) - }() - - select { - case <-waited: - case <-time.After(5 * time.Second): - t.Fatal("Wait missed a Done that happened before it was called") - } -} diff --git a/scheduler/queue/scheduler.go b/scheduler/queue/scheduler.go index a7447cc4dc..066ca53580 100644 --- a/scheduler/queue/scheduler.go +++ b/scheduler/queue/scheduler.go @@ -89,13 +89,12 @@ func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedR if len(tableClients) == 0 { return } - queue := NewConcurrentRandomQueue[WorkUnit](d.seed, len(tableClients)) + queue := newWorkQueue(d.seed, len(tableClients)) for _, tc := range tableClients { queue.Push(tc) } jobs := make(chan *WorkUnit) - activeWorkSignal := newActiveWorkSignal() // Worker pool workerPool, _ := errgroup.WithContext(ctx) @@ -112,40 +111,18 @@ func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedR d.metrics, msgChan, d.errorClassifier, - ).work(ctx, activeWorkSignal) + ).work(ctx) return nil }) } - go dispatch(ctx, queue, jobs, activeWorkSignal) - - _ = workerPool.Wait() -} - -type workQueue interface { - Pop() *WorkUnit -} - -// dispatch hands queued work units to workers until the queue is empty and no work is active. -func dispatch(ctx context.Context, queue workQueue, jobs chan<- *WorkUnit, activeWorkSignal *activeWorkSignal) { - defer close(jobs) - for ctx.Err() == nil { - item := queue.Pop() - if item == nil { - if !activeWorkSignal.IsIdle() { - activeWorkSignal.Wait() - continue - } - // Workers push relations before calling Done, so nothing can be pushed once idle, - // but the last active work unit may have pushed between the Pop above and the idle check. - if item = queue.Pop(); item == nil { - return - } + // Work distribution + go func() { + defer close(jobs) + for item := queue.Next(ctx); item != nil; item = queue.Next(ctx) { + jobs <- item } + }() - // Mark the item active before handing it off so the idle check cannot fire - // while the item is in flight to a worker. - activeWorkSignal.Add() - jobs <- item - } + _ = workerPool.Wait() } diff --git a/scheduler/queue/scheduler_test.go b/scheduler/queue/scheduler_test.go index 11280f3175..8d9b71e4ec 100644 --- a/scheduler/queue/scheduler_test.go +++ b/scheduler/queue/scheduler_test.go @@ -99,42 +99,3 @@ func TestScheduler(t *testing.T) { // 2 top level tables, each with 10 resources, and each top level has a relation with 10 resources require.Len(t, gotResources, 2*10+2*10*10) } - -type pushOnFirstEmptyPopQueue struct { - *ConcurrentRandomQueue[WorkUnit] - onFirstEmptyPop func() -} - -func (q *pushOnFirstEmptyPopQueue) Pop() *WorkUnit { - item := q.ConcurrentRandomQueue.Pop() - if item == nil && q.onFirstEmptyPop != nil { - onEmpty := q.onFirstEmptyPop - q.onFirstEmptyPop = nil - onEmpty() - } - return item -} - -func TestDispatchHandsOffWorkQueuedByLastActiveUnit(t *testing.T) { - activeWorkSignal := newActiveWorkSignal() - queue := &pushOnFirstEmptyPopQueue{ConcurrentRandomQueue: NewConcurrentRandomQueue[WorkUnit](0, 1)} - queue.Push(WorkUnit{Table: &schema.Table{Name: "parent"}}) - // The parent's worker queues a relation and finishes after the dispatcher saw an - // empty queue but before it checked for active work. - queue.onFirstEmptyPop = func() { - queue.Push(WorkUnit{Table: &schema.Table{Name: "relation"}}) - activeWorkSignal.Done() - } - - jobs := make(chan *WorkUnit) - go dispatch(context.Background(), queue, jobs, activeWorkSignal) - - require.Equal(t, "parent", (<-jobs).Table.Name) - relation, ok := <-jobs - require.True(t, ok, "dispatcher stopped without handing off the queued relation") - require.Equal(t, "relation", relation.Table.Name) - activeWorkSignal.Done() - - _, ok = <-jobs - require.False(t, ok) -} diff --git a/scheduler/queue/work_queue.go b/scheduler/queue/work_queue.go new file mode 100644 index 0000000000..6d9c263239 --- /dev/null +++ b/scheduler/queue/work_queue.go @@ -0,0 +1,76 @@ +package queue + +import ( + "context" + "sync" +) + +// workQueue holds the work units waiting for a worker and counts the ones in progress. +// +// The sync is done when no work unit is queued and none is in progress. Workers queue +// relations while their table is in progress, so an empty queue alone does not mean done. +// +// - Push queues a work unit. +// - Next takes a work unit off the queue and marks it in progress. +// - Done marks a work unit taken by Next as finished. +type workQueue struct { + mu sync.Mutex + queue *ConcurrentRandomQueue[WorkUnit] + inProgress int + // Buffered with capacity 1 so a change that happens before Next starts to wait is kept rather than lost. + changed chan struct{} +} + +func newWorkQueue(seed int64, capacityHint int) *workQueue { + return &workQueue{ + queue: NewConcurrentRandomQueue[WorkUnit](seed, capacityHint), + changed: make(chan struct{}, 1), + } +} + +func (q *workQueue) Push(item WorkUnit) { + q.mu.Lock() + q.queue.Push(item) + q.mu.Unlock() + q.notify() +} + +// Next blocks until a work unit is queued and returns it marked in progress. +// It returns nil when no work unit is queued or in progress, or when ctx is done. +func (q *workQueue) Next(ctx context.Context) *WorkUnit { + for { + q.mu.Lock() + item := q.queue.Pop() + if item != nil { + q.inProgress++ + } + idle := item == nil && q.inProgress == 0 + q.mu.Unlock() + + if item != nil { + return item + } + if idle { + return nil + } + select { + case <-q.changed: + case <-ctx.Done(): + return nil + } + } +} + +func (q *workQueue) Done() { + q.mu.Lock() + q.inProgress-- + q.mu.Unlock() + q.notify() +} + +func (q *workQueue) notify() { + select { + case q.changed <- struct{}{}: + default: + } +} diff --git a/scheduler/queue/work_queue_test.go b/scheduler/queue/work_queue_test.go new file mode 100644 index 0000000000..507829c302 --- /dev/null +++ b/scheduler/queue/work_queue_test.go @@ -0,0 +1,94 @@ +package queue + +import ( + "context" + "testing" + "testing/synctest" + + "github.com/cloudquery/plugin-sdk/v4/schema" + "github.com/stretchr/testify/require" +) + +func workUnit(name string) WorkUnit { + return WorkUnit{Table: &schema.Table{Name: name}} +} + +// startNext calls Next in a goroutine and returns after Next blocks or returns. +func startNext(ctx context.Context, q *workQueue) <-chan *WorkUnit { + next := make(chan *WorkUnit, 1) + go func() { next <- q.Next(ctx) }() + synctest.Wait() + return next +} + +func requireNextReturned(t *testing.T, next <-chan *WorkUnit) *WorkUnit { + t.Helper() + synctest.Wait() + select { + case item := <-next: + return item + default: + require.FailNow(t, "Next is still blocked") + return nil + } +} + +func TestWorkQueue(t *testing.T) { + t.Run("returns nil when nothing is queued or in progress", func(t *testing.T) { + require.Nil(t, newWorkQueue(0, 0).Next(t.Context())) + }) + + t.Run("wakes up when a work unit in progress queues a relation", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + q := newWorkQueue(0, 1) + q.Push(workUnit("parent")) + require.Equal(t, "parent", q.Next(t.Context()).Table.Name) + + next := startNext(t.Context(), q) + q.Push(workUnit("relation")) + + require.Equal(t, "relation", requireNextReturned(t, next).Table.Name) + }) + }) + + t.Run("returns a relation that the last work unit in progress queued before it finished", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + q := newWorkQueue(0, 1) + q.Push(workUnit("parent")) + require.Equal(t, "parent", q.Next(t.Context()).Table.Name) + + next := startNext(t.Context(), q) + q.Push(workUnit("relation")) + q.Done() + + require.Equal(t, "relation", requireNextReturned(t, next).Table.Name) + }) + }) + + t.Run("returns nil when the last work unit in progress finishes", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + q := newWorkQueue(0, 1) + q.Push(workUnit("parent")) + require.Equal(t, "parent", q.Next(t.Context()).Table.Name) + + next := startNext(t.Context(), q) + q.Done() + + require.Nil(t, requireNextReturned(t, next)) + }) + }) + + t.Run("returns nil when the context is done", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + q := newWorkQueue(0, 1) + q.Push(workUnit("parent")) + require.Equal(t, "parent", q.Next(ctx).Table.Name) + + next := startNext(ctx, q) + cancel() + + require.Nil(t, requireNextReturned(t, next)) + }) + }) +} diff --git a/scheduler/queue/worker.go b/scheduler/queue/worker.go index 1ee06eea42..6dc4f651c5 100644 --- a/scheduler/queue/worker.go +++ b/scheduler/queue/worker.go @@ -23,7 +23,7 @@ import ( type worker struct { jobs <-chan *WorkUnit - queue *ConcurrentRandomQueue[WorkUnit] + queue *workQueue resolvedResources chan<- *schema.Resource logger zerolog.Logger @@ -36,20 +36,16 @@ type worker struct { errorClassifier schema.ErrorClassifier } -func (w *worker) work(ctx context.Context, activeWorkSignal *activeWorkSignal) { +func (w *worker) work(ctx context.Context) { for j := range w.jobs { - // the work unit was already marked active by the dispatcher before it was - // handed off, so the dispatcher can never observe an idle state while a - // job is in flight between the queue and a worker w.resolveTable(ctx, j.Table, j.Client, j.Parent) - - activeWorkSignal.Done() + w.queue.Done() } } func newWorker( jobs <-chan *WorkUnit, - queue *ConcurrentRandomQueue[WorkUnit], + queue *workQueue, resolvedResources chan<- *schema.Resource, logger zerolog.Logger, From 4febb4244c13423fa8be53f4023222bc767cc4c5 Mon Sep 17 00:00:00 2001 From: Kemal Hadimli Date: Thu, 15 Oct 2026 13:43:12 +0100 Subject: [PATCH 3/4] fix: Stop dispatching queued work once the sync context is done --- scheduler/queue/work_queue.go | 4 ++-- scheduler/queue/work_queue_test.go | 12 ++++++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/scheduler/queue/work_queue.go b/scheduler/queue/work_queue.go index 6d9c263239..60c7e8a1e5 100644 --- a/scheduler/queue/work_queue.go +++ b/scheduler/queue/work_queue.go @@ -38,7 +38,7 @@ func (q *workQueue) Push(item WorkUnit) { // Next blocks until a work unit is queued and returns it marked in progress. // It returns nil when no work unit is queued or in progress, or when ctx is done. func (q *workQueue) Next(ctx context.Context) *WorkUnit { - for { + for ctx.Err() == nil { q.mu.Lock() item := q.queue.Pop() if item != nil { @@ -56,9 +56,9 @@ func (q *workQueue) Next(ctx context.Context) *WorkUnit { select { case <-q.changed: case <-ctx.Done(): - return nil } } + return nil } func (q *workQueue) Done() { diff --git a/scheduler/queue/work_queue_test.go b/scheduler/queue/work_queue_test.go index 507829c302..d45a99a94b 100644 --- a/scheduler/queue/work_queue_test.go +++ b/scheduler/queue/work_queue_test.go @@ -78,6 +78,18 @@ func TestWorkQueue(t *testing.T) { }) }) + t.Run("returns nil when the context is done and work is still queued", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + q := newWorkQueue(0, 2) + q.Push(workUnit("first")) + q.Push(workUnit("second")) + require.NotNil(t, q.Next(ctx)) + + cancel() + + require.Nil(t, q.Next(ctx)) + }) + t.Run("returns nil when the context is done", func(t *testing.T) { synctest.Test(t, func(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) From d36c4f491d0159dc1eecb6a18d1d02df8c98467d Mon Sep 17 00:00:00 2001 From: Kemal Hadimli Date: Thu, 15 Oct 2026 15:49:02 +0100 Subject: [PATCH 4/4] test: Run the idle workQueue test in a synctest bubble --- scheduler/queue/work_queue_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scheduler/queue/work_queue_test.go b/scheduler/queue/work_queue_test.go index d45a99a94b..5e4e863db7 100644 --- a/scheduler/queue/work_queue_test.go +++ b/scheduler/queue/work_queue_test.go @@ -35,7 +35,9 @@ func requireNextReturned(t *testing.T, next <-chan *WorkUnit) *WorkUnit { func TestWorkQueue(t *testing.T) { t.Run("returns nil when nothing is queued or in progress", func(t *testing.T) { - require.Nil(t, newWorkQueue(0, 0).Next(t.Context())) + synctest.Test(t, func(t *testing.T) { + require.Nil(t, newWorkQueue(0, 0).Next(t.Context())) + }) }) t.Run("wakes up when a work unit in progress queues a relation", func(t *testing.T) {