fix: preserve first stop shutdown ownership

logger/queue.go

@@ -197,7 +197,9 @@ func (q *Queue) Stop(contexts ...context.Context) error {
 	}
 
 	q.mu.Lock()
+	shutdownOwner := false
 	if !q.stopped {
+		shutdownOwner = true
 		q.stopped = true
 		q.shutdownCtx = ctx
 		close(q.ch)
@@ -222,7 +224,7 @@ func (q *Queue) Stop(contexts ...context.Context) error {
 	case <-done:
 		return nil
 	case <-ctx.Done():
-		if workCancel != nil {
+		if shutdownOwner && workCancel != nil {
 			workCancel()
 		}
 		return ctx.Err()

tests/logger/queue_test.go

@@ -443,6 +443,48 @@ func TestQueueStopAndSubmitAreConcurrentAndRepeatSafe(t *testing.T) {
 	}
 }
 
+func TestConcurrentStopCannotCancelFirstStopDrain(t *testing.T) {
+	const initialEntries = 32
+	firstPrepareStarted := make(chan struct{})
+	releaseFirstPrepare := make(chan struct{})
+	var prepareCalls atomic.Int32
+	backend := newQueueBackend(func(context.Context, string) (logger.Batch, error) {
+		if prepareCalls.Add(1) == 1 {
+			close(firstPrepareStarted)
+			<-releaseFirstPrepare
+		}
+		return &fakeBatch{}, nil
+	})
+	q := logger.NewQueueWithBackend(backend, 1024, 1, 1, time.Hour)
+	q.Start(context.Background())
+	for i := 0; i < initialEntries; i++ {
+		q.Submit(&logger.LogEntry{RequestID: "queued"})
+	}
+	<-firstPrepareStarted
+
+	firstResult := make(chan error, 1)
+	firstCtx, cancelFirst := context.WithTimeout(context.Background(), time.Second)
+	defer cancelFirst()
+	go func() { firstResult <- q.Stop(firstCtx) }()
+	for q.Stats().Dropped == 0 {
+		q.Submit(&logger.LogEntry{RequestID: "stop-probe"})
+	}
+	enqueued := q.Stats().Enqueued
+
+	secondCtx, cancelSecond := context.WithCancel(context.Background())
+	cancelSecond()
+	if err := q.Stop(secondCtx); !errors.Is(err, context.Canceled) {
+		t.Fatalf("second Stop error=%v want canceled", err)
+	}
+	close(releaseFirstPrepare)
+	if err := <-firstResult; err != nil {
+		t.Fatalf("first Stop: %v", err)
+	}
+	if calls := uint64(backend.calls.Load()); calls != enqueued {
+		t.Fatalf("PrepareBatch calls=%d want %d; second Stop interrupted drain", calls, enqueued)
+	}
+}
+
 func TestStopIsIndependentFromStartContext(t *testing.T) {
 	root, cancelRoot := context.WithCancel(context.Background())
 	q := logger.NewQueue(nil, 4, 4, 1, time.Hour, 1024)