feat(chromium): re-add concurrency support for Chromium (#1467)

* feat: add concurrency support to ProcessSupervisor

- Replace the single-slot mutex channel with a configurable semaphore to
allow multiple concurrent tasks.
- Add drain logic to ensure all active
tasks complete before process restarts.

* feat: add chromium-max-concurrency flag

- Add a --chromium-max-concurrency flag (1-6) to the Chromium module to
control how many conversions run in parallel.
- Update LibreOffice to pass maxConcurrency=1 as LibreOffice only supports
a single concurrent conversion.

* test: add integration tests for concurrent Chromium conversions

- Add concurrent request support to the integration test framework with
new step definitions for sending parallel requests and asserting on all
responses.
- Add a feature file for concurrent HTML to PDF conversions.

pkg/gotenberg/supervisor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	"go.uber.org/zap"
@@ -79,27 +80,38 @@ type processSupervisor struct {
 	process         Process
 	maxReqLimit     int64
 	maxQueueSize    int64
-	mutexChan       chan struct{}
+	maxConcurrency  int64
+	semaphore       chan struct{}
 	firstStart      atomic.Bool
+	firstStartOnce  sync.Once
+	firstStartErr   error
 	reqCounter      atomic.Int64
 	reqQueueSize    atomic.Int64
 	restartsCounter atomic.Int64
 	isRestarting    atomic.Bool
+	activeTasks     atomic.Int64
+	restartMutex    sync.Mutex
 }
 
 // NewProcessSupervisor initializes a new [ProcessSupervisor].
-func NewProcessSupervisor(logger *zap.Logger, process Process, maxReqLimit, maxQueueSize int64) ProcessSupervisor {
+func NewProcessSupervisor(logger *zap.Logger, process Process, maxReqLimit, maxQueueSize, maxConcurrency int64) ProcessSupervisor {
+	if maxConcurrency < 1 {
+		maxConcurrency = 1
+	}
+
 	b := &processSupervisor{
-		logger:       logger,
-		process:      process,
-		mutexChan:    make(chan struct{}, 1),
-		maxReqLimit:  maxReqLimit,
-		maxQueueSize: maxQueueSize,
+		logger:         logger,
+		process:        process,
+		semaphore:      make(chan struct{}, maxConcurrency),
+		maxReqLimit:    maxReqLimit,
+		maxQueueSize:   maxQueueSize,
+		maxConcurrency: maxConcurrency,
 	}
 	b.reqCounter.Store(0)
 	b.reqQueueSize.Store(0)
 	b.restartsCounter.Store(0)
 	b.isRestarting.Store(false)
+	b.activeTasks.Store(0)
 
 	return b
 }
@@ -130,15 +142,7 @@ func (s *processSupervisor) Shutdown() error {
 }
 
 func (s *processSupervisor) restart() error {
-	if s.isRestarting.Load() {
-		s.logger.Debug("process already restarting, skip restart")
-
-		return ErrProcessAlreadyRestarting
-	}
-
 	s.logger.Debug("restart process")
-	s.isRestarting.Store(true)
-	defer s.isRestarting.Store(false)
 
 	err := s.Shutdown()
 	if err != nil {
@@ -197,33 +201,43 @@ func (s *processSupervisor) Run(ctx context.Context, logger *zap.Logger, task fu
 	for {
 		err := func() error {
 			select {
-			case s.mutexChan <- struct{}{}:
+			case s.semaphore <- struct{}{}:
 				logger.Debug("process lock acquired")
+
+				// If a restart drain is in progress, release the slot
+				// immediately so the drain can acquire it instead.
+				if s.isRestarting.Load() {
+					<-s.semaphore
+					return ErrProcessAlreadyRestarting
+				}
+
 				s.reqQueueSize.Add(-1)
 				s.reqCounter.Add(1)
-				releaseMutexChan := true
+				s.activeTasks.Add(1)
+				releaseSemaphore := true
 
 				defer func() {
-					if releaseMutexChan {
+					s.activeTasks.Add(-1)
+					if releaseSemaphore {
 						logger.Debug("process lock released")
-						<-s.mutexChan
+						<-s.semaphore
 					}
 				}()
 
 				if !s.firstStart.Load() {
-					err := s.runWithDeadline(ctx, func() error {
-						return s.Launch()
+					s.firstStartOnce.Do(func() {
+						s.firstStartErr = s.runWithDeadline(ctx, func() error {
+							return s.Launch()
+						})
 					})
-					if err != nil {
-						return fmt.Errorf("process first start: %w", err)
+					if s.firstStartErr != nil {
+						return fmt.Errorf("process first start: %w", s.firstStartErr)
 					}
 				}
 
 				if !s.Healthy() {
 					s.logger.Debug("process is unhealthy, cannot handle task, restarting...")
-					err := s.runWithDeadline(ctx, func() error {
-						return s.restart()
-					})
+					err := s.doRestart(ctx)
 					if err != nil {
 						return fmt.Errorf("process restart before task: %w", err)
 					}
@@ -232,19 +246,21 @@ func (s *processSupervisor) Run(ctx context.Context, logger *zap.Logger, task fu
 				err := s.runWithDeadline(ctx, task)
 
 				if s.maxReqLimit > 0 && s.reqCounter.Load() >= s.maxReqLimit {
-					s.logger.Debug("max request limit reached, restarting eagerly...")
-					releaseMutexChan = false
+					// Only one goroutine should trigger the restart.
+					if s.restartMutex.TryLock() {
+						s.logger.Debug("max request limit reached, restarting eagerly...")
+						releaseSemaphore = false
 
-					go func() {
-						err := s.runWithDeadline(context.Background(), func() error {
-							return s.restart()
-						})
-						if err != nil {
-							s.logger.Error(fmt.Sprintf("process restart after task: %v", err))
-						}
-						logger.Debug("process lock released")
-						<-s.mutexChan
-					}()
+						go func() {
+							restartErr := s.doRestartLocked(context.Background())
+							s.restartMutex.Unlock()
+							if restartErr != nil {
+								s.logger.Error(fmt.Sprintf("process restart after task: %v", restartErr))
+							}
+							logger.Debug("process lock released")
+							<-s.semaphore
+						}()
+					}
 				}
 
 				// Note: no error wrapping because it leaks on Chromium console exceptions output.
@@ -259,7 +275,6 @@ func (s *processSupervisor) Run(ctx context.Context, logger *zap.Logger, task fu
 
 		if errors.Is(err, ErrProcessAlreadyRestarting) {
 			logger.Debug("process is already restarting, trying to acquire process lock again...")
-			s.reqQueueSize.Add(1)
 			continue
 		}
 
@@ -268,6 +283,47 @@ func (s *processSupervisor) Run(ctx context.Context, logger *zap.Logger, task fu
 	}
 }
 
+// doRestart coordinates a process restart, draining all active concurrent
+// tasks before stopping and restarting the process.
+func (s *processSupervisor) doRestart(ctx context.Context) error {
+	s.restartMutex.Lock()
+	defer s.restartMutex.Unlock()
+
+	return s.doRestartLocked(ctx)
+}
+
+// doRestartLocked performs the restart drain logic. The caller must hold restartMutex.
+func (s *processSupervisor) doRestartLocked(ctx context.Context) error {
+	s.isRestarting.Store(true)
+	defer s.isRestarting.Store(false)
+
+	// Drain all other active semaphore slots so no other tasks are running during the restart.
+	slotsToAcquire := s.maxConcurrency - 1
+	acquired := make([]struct{}, 0, slotsToAcquire)
+
+	for range slotsToAcquire {
+		select {
+		case s.semaphore <- struct{}{}:
+			acquired = append(acquired, struct{}{})
+		case <-ctx.Done():
+			for range acquired {
+				<-s.semaphore
+			}
+			return fmt.Errorf("drain active tasks before restart: %w", ctx.Err())
+		}
+	}
+
+	err := s.runWithDeadline(ctx, func() error {
+		return s.restart()
+	})
+
+	for range acquired {
+		<-s.semaphore
+	}
+
+	return err
+}
+
 func (s *processSupervisor) runWithDeadline(ctx context.Context, task func() error) error {
 	runChan := make(chan error, 1)
 	go func() {

pkg/gotenberg/supervisor_test.go

@@ -46,7 +46,7 @@ func TestProcessSupervisor_Launch(t *testing.T) {
 				},
 			}
 
-			ps := NewProcessSupervisor(logger, process, 5, 0).(*processSupervisor)
+			ps := NewProcessSupervisor(logger, process, 5, 0, 1).(*processSupervisor)
 			if tc.firstStartSet {
 				ps.firstStart.Store(true)
 			}
@@ -94,7 +94,7 @@ func TestProcessSupervisor_Shutdown(t *testing.T) {
 				},
 			}
 
-			ps := NewProcessSupervisor(logger, process, 5, 0)
+			ps := NewProcessSupervisor(logger, process, 5, 0, 1)
 			err := ps.Shutdown()
 
 			if !tc.expectError && err != nil {
@@ -110,19 +110,11 @@ func TestProcessSupervisor_Shutdown(t *testing.T) {
 
 func TestProcessSupervisor_restart(t *testing.T) {
 	for _, tc := range []struct {
-		scenario            string
-		initiallyRestarting bool
-		startError          error
-		stopError           error
-		expectError         bool
-		expectedError       error
+		scenario    string
+		startError  error
+		stopError   error
+		expectError bool
 	}{
-		{
-			scenario:            "already restarting",
-			initiallyRestarting: true,
-			expectError:         true,
-			expectedError:       ErrProcessAlreadyRestarting,
-		},
 		{
 			scenario:    "successful restart",
 			startError:  nil,
@@ -154,10 +146,7 @@ func TestProcessSupervisor_restart(t *testing.T) {
 				},
 			}
 
-			ps := NewProcessSupervisor(logger, process, 5, 0).(*processSupervisor)
-			if tc.initiallyRestarting {
-				ps.isRestarting.Store(true)
-			}
+			ps := NewProcessSupervisor(logger, process, 5, 0, 1).(*processSupervisor)
 
 			err := ps.restart()
 
@@ -168,10 +157,6 @@ func TestProcessSupervisor_restart(t *testing.T) {
 			if tc.expectError && err == nil {
 				t.Fatal("expected error but got none")
 			}
-
-			if tc.expectedError != nil && !errors.Is(err, tc.expectedError) {
-				t.Fatalf("expected error %v but got: %v", tc.expectedError, err)
-			}
 		})
 	}
 }
@@ -217,7 +202,7 @@ func TestProcessSupervisor_Healthy(t *testing.T) {
 				},
 			}
 
-			ps := NewProcessSupervisor(logger, process, 5, 0).(*processSupervisor)
+			ps := NewProcessSupervisor(logger, process, 5, 0, 1).(*processSupervisor)
 			if tc.initiallyStarted {
 				ps.firstStart.Store(true)
 			}
@@ -402,7 +387,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 				},
 			}
 
-			ps := NewProcessSupervisor(logger, process, tc.maxReqLimit, tc.maxQueueSize).(*processSupervisor)
+			ps := NewProcessSupervisor(logger, process, tc.maxReqLimit, tc.maxQueueSize, 1).(*processSupervisor)
 			if tc.initiallyStarted {
 				ps.firstStart.Store(true)
 			}
@@ -452,8 +437,8 @@ func TestProcessSupervisor_Run(t *testing.T) {
 			}
 
 			// Making sure restarts are finished.
-			ps.mutexChan <- struct{}{}
-			<-ps.mutexChan
+			ps.semaphore <- struct{}{}
+			<-ps.semaphore
 
 			if startCalls.Load() != tc.expectedStartCalls {
 				t.Errorf("expected %d process.Start calls, got %d", tc.expectedStartCalls, startCalls.Load())
@@ -488,7 +473,7 @@ func TestProcessSupervisor_runWithDeadline(t *testing.T) {
 		},
 	} {
 		t.Run(tc.scenario, func(t *testing.T) {
-			ps := NewProcessSupervisor(zap.NewNop(), new(ProcessMock), 0, 0).(*processSupervisor)
+			ps := NewProcessSupervisor(zap.NewNop(), new(ProcessMock), 0, 0, 1).(*processSupervisor)
 
 			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 			if tc.ctxDone {
@@ -522,10 +507,10 @@ func TestProcessSupervisor_ReqQueueSize(t *testing.T) {
 			return true
 		},
 	}
-	ps := NewProcessSupervisor(logger, process, 0, 0).(*processSupervisor)
+	ps := NewProcessSupervisor(logger, process, 0, 0, 1).(*processSupervisor)
 
 	// Simulating a lock.
-	ps.mutexChan <- struct{}{}
+	ps.semaphore <- struct{}{}
 
 	if ps.ReqQueueSize() != 0 {
 		t.Fatalf("expected queue size to be 0 but got %d", ps.ReqQueueSize())
@@ -623,7 +608,7 @@ func TestProcessSupervisor_RestartsCount(t *testing.T) {
 				},
 			}
 
-			ps := NewProcessSupervisor(logger, process, 0, 0).(*processSupervisor)
+			ps := NewProcessSupervisor(logger, process, 0, 0, 1).(*processSupervisor)
 			ps.restartsCounter.Store(tc.initialRestartsCount)
 
 			for i := 0; i < tc.restartAttempts; i++ {
@@ -637,3 +622,126 @@ func TestProcessSupervisor_RestartsCount(t *testing.T) {
 		})
 	}
 }
+
+func TestProcessSupervisor_ConcurrentRun(t *testing.T) {
+	logger := zap.NewNop()
+
+	var startCalls atomic.Int64
+	process := &ProcessMock{
+		StartMock: func(logger *zap.Logger) error {
+			startCalls.Add(1)
+			return nil
+		},
+		StopMock: func(logger *zap.Logger) error {
+			return nil
+		},
+		HealthyMock: func(logger *zap.Logger) bool {
+			return true
+		},
+	}
+
+	maxConcurrency := int64(3)
+	ps := NewProcessSupervisor(logger, process, 0, 0, maxConcurrency).(*processSupervisor)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var running atomic.Int64
+	var maxRunning atomic.Int64
+
+	var wg sync.WaitGroup
+	tasks := 6
+
+	for i := 0; i < tasks; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			err := ps.Run(ctx, logger, func() error {
+				cur := running.Add(1)
+				for {
+					old := maxRunning.Load()
+					if cur <= old || maxRunning.CompareAndSwap(old, cur) {
+						break
+					}
+				}
+				time.Sleep(50 * time.Millisecond)
+				running.Add(-1)
+				return nil
+			})
+			if err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	observed := maxRunning.Load()
+	if observed > maxConcurrency {
+		t.Fatalf("expected at most %d concurrent tasks, but observed %d", maxConcurrency, observed)
+	}
+	if observed < 2 {
+		t.Fatalf("expected concurrent execution (at least 2 tasks running simultaneously), but observed max %d", observed)
+	}
+
+	if startCalls.Load() != 1 {
+		t.Errorf("expected 1 start call, got %d", startCalls.Load())
+	}
+}
+
+func TestProcessSupervisor_RestartDrainsAllSlots(t *testing.T) {
+	logger := zap.NewNop()
+
+	process := &ProcessMock{
+		StartMock: func(logger *zap.Logger) error {
+			return nil
+		},
+		StopMock: func(logger *zap.Logger) error {
+			return nil
+		},
+		HealthyMock: func(logger *zap.Logger) bool {
+			return true
+		},
+	}
+
+	maxConcurrency := int64(3)
+	ps := NewProcessSupervisor(logger, process, 3, 0, maxConcurrency).(*processSupervisor)
+	ps.firstStart.Store(true)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	tasks := 3
+
+	for i := 0; i < tasks; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			err := ps.Run(ctx, logger, func() error {
+				time.Sleep(50 * time.Millisecond)
+				return nil
+			})
+			if err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// Wait for the async restart goroutine to complete.
+	deadline := time.After(5 * time.Second)
+	for ps.RestartsCount() < 1 {
+		select {
+		case <-deadline:
+			t.Fatal("timed out waiting for restart to complete")
+		default:
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+
+	if ps.RestartsCount() != 1 {
+		t.Fatalf("expected 1 restart, got %d", ps.RestartsCount())
+	}
+}

pkg/modules/chromium/chromium.go

@@ -86,9 +86,10 @@ var (
 // Chromium is a module that provides both an [Api] and routes for converting
 // an HTML document to PDF.
 type Chromium struct {
-	autoStart     bool
-	disableRoutes bool
-	args          browserArguments
+	autoStart      bool
+	disableRoutes  bool
+	maxConcurrency int64
+	args           browserArguments
 
 	logger     *zap.Logger
 	browser    browser
@@ -391,8 +392,9 @@ func (mod *Chromium) Descriptor() gotenberg.ModuleDescriptor {
 		ID: "chromium",
 		FlagSet: func() *flag.FlagSet {
 			fs := flag.NewFlagSet("chromium", flag.ExitOnError)
-			fs.Int64("chromium-restart-after", 10, "Number of conversions after which Chromium will automatically restart. Set to 0 to disable this feature")
+			fs.Int64("chromium-restart-after", 100, "Number of conversions after which Chromium will automatically restart. Set to 0 to disable this feature")
 			fs.Int64("chromium-max-queue-size", 0, "Maximum request queue size for Chromium. Set to 0 to disable this feature")
+			fs.Int64("chromium-max-concurrency", 6, "Maximum number of concurrent conversions. Chromium supports up to 6")
 			fs.Bool("chromium-auto-start", false, "Automatically launch Chromium upon initialization if set to true; otherwise, Chromium will start at the time of the first conversion")
 			fs.Duration("chromium-start-timeout", time.Duration(20)*time.Second, "Maximum duration to wait for Chromium to start or restart")
 			fs.Bool("chromium-allow-insecure-localhost", false, "Ignore TLS/SSL errors on localhost")
@@ -426,6 +428,7 @@ func (mod *Chromium) Provision(ctx *gotenberg.Context) error {
 	flags := ctx.ParsedFlags()
 	mod.autoStart = flags.MustBool("chromium-auto-start")
 	mod.disableRoutes = flags.MustBool("chromium-disable-routes")
+	mod.maxConcurrency = flags.MustInt64("chromium-max-concurrency")
 
 	binPath, ok := os.LookupEnv("CHROMIUM_BIN_PATH")
 	if !ok {
@@ -468,7 +471,7 @@ func (mod *Chromium) Provision(ctx *gotenberg.Context) error {
 
 	// Process.
 	mod.browser = newChromiumBrowser(mod.args)
-	mod.supervisor = gotenberg.NewProcessSupervisor(mod.logger, mod.browser, flags.MustInt64("chromium-restart-after"), flags.MustInt64("chromium-max-queue-size"))
+	mod.supervisor = gotenberg.NewProcessSupervisor(mod.logger, mod.browser, flags.MustInt64("chromium-restart-after"), flags.MustInt64("chromium-max-queue-size"), mod.maxConcurrency)
 
 	// PDF Engine.
 	provider, err := ctx.Module(new(gotenberg.PdfEngineProvider))
@@ -486,6 +489,10 @@ func (mod *Chromium) Provision(ctx *gotenberg.Context) error {
 
 // Validate validates the module properties.
 func (mod *Chromium) Validate() error {
+	if mod.maxConcurrency < 1 || mod.maxConcurrency > 6 {
+		return fmt.Errorf("chromium-max-concurrency must be between 1 and 6, got %d", mod.maxConcurrency)
+	}
+
 	_, err := os.Stat(mod.args.binPath)
 	if os.IsNotExist(err) {
 		return fmt.Errorf("chromium binary path does not exist: %w", err)

pkg/modules/libreoffice/api/api.go

@@ -253,7 +253,7 @@ func (a *Api) Provision(ctx *gotenberg.Context) error {
 
 	// Process.
 	a.libreOffice = newLibreOfficeProcess(a.args)
-	a.supervisor = gotenberg.NewProcessSupervisor(a.logger, a.libreOffice, flags.MustInt64("libreoffice-restart-after"), flags.MustInt64("libreoffice-max-queue-size"))
+	a.supervisor = gotenberg.NewProcessSupervisor(a.logger, a.libreOffice, flags.MustInt64("libreoffice-restart-after"), flags.MustInt64("libreoffice-max-queue-size"), 1)
 
 	return nil
 }