diff --git a/scheduler/queue/active_work_signal.go b/scheduler/queue/active_work_signal.go
deleted file mode 100644
index f1c59d025c..0000000000
--- a/scheduler/queue/active_work_signal.go
+++ /dev/null
@@ -1,80 +0,0 @@
-package queue
-
-import (
-	"sync"
-	"sync/atomic"
-	"time"
-)
-
-// 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 state changes.
-type activeWorkSignal struct {
-	countChangeSignal   *sync.Cond
-	activeWorkUnitCount *atomic.Int32
-	isStarted           *atomic.Bool
-}
-
-func newActiveWorkSignal() *activeWorkSignal {
-	return &activeWorkSignal{
-		countChangeSignal:   sync.NewCond(&sync.Mutex{}),
-		activeWorkUnitCount: &atomic.Int32{},
-		isStarted:           &atomic.Bool{},
-	}
-}
-
-// 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).
-func (s *activeWorkSignal) Done() {
-	s.activeWorkUnitCount.Add(-1)
-	s.countChangeSignal.Signal()
-}
-
-// 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 the count of active workers changes.
-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()
-}
diff --git a/scheduler/queue/scheduler.go b/scheduler/queue/scheduler.go
index 1a0241d1e7..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,7 +111,7 @@ func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedR
 				d.metrics,
 				msgChan,
 				d.errorClassifier,
-			).work(ctx, activeWorkSignal)
+			).work(ctx)
 			return nil
 		})
 	}
@@ -120,29 +119,8 @@ 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
-				activeWorkSignal.Wait()
-			}
+		for item := queue.Next(ctx); item != nil; item = queue.Next(ctx) {
+			jobs <- item
 		}
 	}()
 
diff --git a/scheduler/queue/work_queue.go b/scheduler/queue/work_queue.go
new file mode 100644
index 0000000000..60c7e8a1e5
--- /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 ctx.Err() == nil {
+		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..5e4e863db7
--- /dev/null
+++ b/scheduler/queue/work_queue_test.go
@@ -0,0 +1,108 @@
+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) {
+		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) {
+		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 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())
+			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,