fix(queue): don't deadlock Stop() when the consumer stops reading output

ListContainerDir aborts on the first stat error and stops reading the queue's
output channel, leaving workers blocked on the unbuffered `q.output <- result`
send and the reader blocked on a full `q.queue`; Stop() -> wg.Wait() then hung
forever. Select every pipeline send/wait on q.ctx so a stopped queue unwinds.
On stop a worker returns without msg.cancel(), so later workers also bail via
q.ctx and output ends at a contiguous prefix instead of developing gaps.

Regression test drives the exact hang (unread output -> Stop must return),
verified with -race.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/pkg/queue/queue.go

@@ -150,11 +150,24 @@ func (q *queue[I, O]) worker(wid int) {
 		} else if result, err := q.process(msg.value); err != nil {
 			logger.WithError(err).Error("failed to process message")
 		} else {
-			// wait until the previous message is sent to the output channel
-			<-msg.doneCtx.Done()
+			// Wait for the previous message to be sent (preserves output order),
+			// then send this one. Both waits also select on q.ctx so a stopped
+			// queue whose consumer quit reading output can't block the worker
+			// forever and deadlock Stop() -> wg.Wait(). On stop we return WITHOUT
+			// msg.cancel(): leaving the next message's doneCtx uncancelled makes
+			// every later worker bail via q.ctx too, so output ends at a contiguous
+			// prefix instead of developing gaps.
+			select {
+			case <-msg.doneCtx.Done():
+			case <-q.ctx.Done():
+				return
+			}
 
-			// send the converted events to the output channel
-			q.output <- result
+			select {
+			case q.output <- result:
+			case <-q.ctx.Done():
+				return
+			}
 		}
 
 		// close the context to mark this operation as complete
@@ -196,10 +209,17 @@ func (q *queue[I, O]) reader() {
 			// create a new context for each message
 			newCtx, cancel := context.WithCancel(context.Background())
 
-			q.queue <- &message[I]{
+			// select on q.ctx so a stopped queue whose workers have exited
+			// cannot block the reader forever on a full q.queue.
+			select {
+			case q.queue <- &message[I]{
 				value:   value,
 				doneCtx: lastDoneCtx,
 				cancel:  cancel,
+			}:
+			case <-q.ctx.Done():
+				cancel()
+				return
 			}
 
 			// update lastDoneCtx for next message

backend/pkg/queue/queue_test.go

@@ -129,6 +129,45 @@ func TestQueue_ProcessOrdering(t *testing.T) {
 	}
 }
 
+// A consumer that stops reading output (e.g. ListContainerDir bailing on the
+// first stat error) leaves workers blocked on the unbuffered send; Stop() must
+// still return instead of hanging on wg.Wait().
+func TestQueue_StopDoesNotDeadlockWithUnreadOutput(t *testing.T) {
+	input := make(chan int, 20)
+	output := make(chan int) // unbuffered; deliberately left unread below
+	workers := 4
+
+	q := queue.NewQueue(input, output, workers, func(i int) (int, error) {
+		return i, nil
+	})
+	if err := q.Start(); err != nil {
+		t.Fatalf("failed to start queue: %v", err)
+	}
+
+	for i := 0; i < 20; i++ {
+		input <- i
+	}
+	close(input)
+
+	<-output // take one result, then abandon the channel with items still in flight
+
+	done := make(chan error, 1)
+	go func() { done <- q.Stop() }()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Stop returned an error: %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("Stop() deadlocked with unread output")
+	}
+
+	if q.Running() {
+		t.Error("expected queue to be stopped")
+	}
+}
+
 func BenchmarkQueue_DefaultWorkers(b *testing.B) {
 	simpleBenchmark(b, 0)
 }