fix(supervisor): queue slot - second request hitting a busy node now gets a 429 immediately, pushing backpressure to a load balancer, if any

pkg/gotenberg/supervisor.go

@@ -199,13 +199,19 @@ func (s *processSupervisor) Run(ctx context.Context, logger *zap.Logger, task fu
 		}
 	}
 
+	// Decrement when Run() returns, regardless of which path is taken
+	// (context timeout, task completion, error, etc.). This ensures the
+	// request is counted as "in the queue" for the entire duration of Run(),
+	// preventing new requests from entering while one is being processed.
+	// See https://github.com/gotenberg/gotenberg/issues/1502.
+	defer s.reqQueueSize.Add(-1)
+
 	for {
 		err := func() error {
 			if err := s.acquireSlot(ctx, logger); err != nil {
 				return err
 			}
 
-			s.reqQueueSize.Add(-1)
 			s.reqCounter.Add(1)
 			s.activeTasks.Add(1)
 			semaphoreOwned := true
@@ -264,7 +270,6 @@ func (s *processSupervisor) acquireSlot(ctx context.Context, logger *zap.Logger)
 		return nil
 	case <-ctx.Done():
 		logger.Debug("failed to acquire process lock before deadline")
-		s.reqQueueSize.Add(-1)
 
 		return fmt.Errorf("acquire process lock: %w", ctx.Err())
 	}

pkg/gotenberg/supervisor_test.go

@@ -607,6 +607,56 @@ func TestProcessSupervisor_QueueSizeCAS(t *testing.T) {
 	}
 }
 
+func TestProcessSupervisor_QueueSizeIncludesActiveTasks(t *testing.T) {
+	logger := zap.NewNop()
+
+	process := &ProcessMock{
+		StartMock: func(logger *zap.Logger) error {
+			return nil
+		},
+		HealthyMock: func(logger *zap.Logger) bool {
+			return true
+		},
+	}
+
+	// maxQueueSize=1, maxConcurrency=1: only one request at a time.
+	ps := NewProcessSupervisor(logger, process, 0, 1, 1)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	taskStarted := make(chan struct{})
+	taskDone := make(chan struct{})
+
+	// Start a long-running task that holds the slot.
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		err := ps.Run(ctx, logger, func() error {
+			close(taskStarted)
+			<-taskDone
+			return nil
+		})
+		if err != nil {
+			t.Errorf("first task: unexpected error: %v", err)
+		}
+	})
+
+	// Wait for the first task to be running.
+	<-taskStarted
+
+	// A second request should be rejected immediately because the queue
+	// slot is still held by the active task.
+	err := ps.Run(ctx, logger, func() error {
+		return nil
+	})
+	if !errors.Is(err, ErrMaximumQueueSizeExceeded) {
+		t.Fatalf("expected ErrMaximumQueueSizeExceeded but got: %v", err)
+	}
+
+	close(taskDone)
+	wg.Wait()
+}
+
 func TestProcessSupervisor_RestartsCount(t *testing.T) {
 	for _, tc := range []struct {
 		scenario              string