fix(supervisor): if process is already restarting, requeue a task (#754)

pkg/gotenberg/supervisor.go

@@ -2,12 +2,17 @@ package gotenberg
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync/atomic"
 
 	"go.uber.org/zap"
 )
 
+// ErrProcessAlreadyRestarting happens if the [ProcessSupervisor] is trying
+// to restart an already restarting [Process].
+var ErrProcessAlreadyRestarting = errors.New("process already restarting")
+
 // Process is an interface that represents an abstract process
 // and provides methods for starting, stopping, and checking the health of the
 // process.
@@ -122,7 +127,7 @@ func (s *processSupervisor) restart() error {
 	if s.isRestarting.Load() {
 		s.logger.Debug("process already restarting, skip restart")
 
-		return nil
+		return ErrProcessAlreadyRestarting
 	}
 
 	s.logger.Debug("restart process")
@@ -164,53 +169,66 @@ func (s *processSupervisor) Healthy() bool {
 func (s *processSupervisor) Run(ctx context.Context, logger *zap.Logger, task func() error) error {
 	s.reqQueueSize.Add(1)
 
-	select {
+	for {
-	case s.mutexChan <- struct{}{}:
+		err := func() error {
-		logger.Debug("process lock acquired")
+			select {
-		s.reqQueueSize.Add(-1)
+			case s.mutexChan <- struct{}{}:
-		s.reqCounter.Add(1)
+				logger.Debug("process lock acquired")
+				s.reqQueueSize.Add(-1)
+				s.reqCounter.Add(1)
 
-		defer func() {
+				defer func() {
-			logger.Debug("process lock released")
+					logger.Debug("process lock released")
-			<-s.mutexChan
+					<-s.mutexChan
+				}()
+
+				if !s.firstStart.Load() {
+					err := s.runWithDeadline(ctx, func() error {
+						return s.Launch()
+					})
+					if err != nil {
+						return fmt.Errorf("process first start: %w", err)
+					}
+				}
+
+				if !s.Healthy() {
+					s.logger.Debug("process is unhealthy, cannot handle task, restarting...")
+					err := s.runWithDeadline(ctx, func() error {
+						return s.restart()
+					})
+					if err != nil {
+						return fmt.Errorf("process restart before task: %w", err)
+					}
+				}
+
+				if s.maxReqLimit > 0 && s.reqCounter.Load() >= s.maxReqLimit {
+					s.logger.Debug("max request limit reached, restarting...")
+					err := s.runWithDeadline(ctx, func() error {
+						return s.restart()
+					})
+					if err != nil {
+						return fmt.Errorf("process restart before task: %w", err)
+					}
+				}
+
+				// Note: no error wrapping because it leaks on Chromium console exceptions output.
+				return s.runWithDeadline(ctx, task)
+			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())
+			}
 		}()
 
-		if !s.firstStart.Load() {
+		if errors.Is(err, ErrProcessAlreadyRestarting) {
-			err := s.runWithDeadline(ctx, func() error {
+			logger.Debug("process is already restarting, trying to acquire process lock again...")
-				return s.Launch()
+			s.reqQueueSize.Add(1)
-			})
+			continue
-			if err != nil {
-				return fmt.Errorf("process first start: %w", err)
-			}
 		}
 
-		if !s.Healthy() {
+		// Note: no error wrapping because it leaks on Chromium console exceptions output.
-			s.logger.Debug("process is unhealthy, cannot handle task, restarting...")
+		return err
-			err := s.runWithDeadline(ctx, func() error {
-				return s.restart()
-			})
-			if err != nil {
-				return fmt.Errorf("process restart before task: %w", err)
-			}
-		}
-
-		if s.maxReqLimit > 0 && s.reqCounter.Load() >= s.maxReqLimit {
-			s.logger.Debug("max request limit reached, restarting...")
-			err := s.runWithDeadline(ctx, func() error {
-				return s.restart()
-			})
-			if err != nil {
-				return fmt.Errorf("process restart before task: %w", err)
-			}
-		}
-
-		// FIXME: no error wrapping because it leaks on Chromium console exceptions output.
-		return s.runWithDeadline(ctx, task)
-	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

@@ -115,11 +115,13 @@ func TestProcessSupervisor_restart(t *testing.T) {
 		startError          error
 		stopError           error
 		expectError         bool
+		expectedError       error
 	}{
 		{
 			scenario:            "already restarting",
 			initiallyRestarting: true,
-			expectError:         false,
+			expectError:         true,
+			expectedError:       ErrProcessAlreadyRestarting,
 		},
 		{
 			scenario:    "successful restart",
@@ -166,6 +168,10 @@ 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)
+			}
 		})
 	}
 }
@@ -232,12 +238,14 @@ func TestProcessSupervisor_Run(t *testing.T) {
 	for _, tc := range []struct {
 		scenario             string
 		initiallyStarted     bool
+		isRestarting         bool
 		startError           error
 		processHealthy       bool
 		maxReqLimit          int64
 		tasksToRun           int
 		taskError            error
 		expectError          bool
+		skipCallsCheck       bool
 		expectedStartCalls   int64
 		expectedHealthyCalls int64
 		expectedStopCalls    int64
@@ -245,6 +253,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 		{
 			scenario:             "successfully run task on non-started process",
 			initiallyStarted:     false,
+			isRestarting:         false,
 			processHealthy:       true,
 			maxReqLimit:          2,
 			tasksToRun:           1,
@@ -256,6 +265,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 		{
 			scenario:             "cannot launch non-started process",
 			initiallyStarted:     false,
+			isRestarting:         false,
 			startError:           errors.New("launch error"),
 			processHealthy:       true,
 			maxReqLimit:          2,
@@ -268,6 +278,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 		{
 			scenario:             "run task with unhealthy process causing restart",
 			initiallyStarted:     true,
+			isRestarting:         false,
 			processHealthy:       false,
 			maxReqLimit:          2,
 			tasksToRun:           1,
@@ -280,6 +291,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 			scenario:             "cannot restart unhealthy process",
 			startError:           errors.New("start error"),
 			initiallyStarted:     true,
+			isRestarting:         false,
 			processHealthy:       false,
 			maxReqLimit:          2,
 			tasksToRun:           1,
@@ -288,9 +300,20 @@ func TestProcessSupervisor_Run(t *testing.T) {
 			expectedHealthyCalls: 1,
 			expectedStopCalls:    1,
 		},
+		{
+			scenario:         "ErrProcessAlreadyRestarting",
+			initiallyStarted: true,
+			isRestarting:     true,
+			processHealthy:   false,
+			maxReqLimit:      1,
+			tasksToRun:       1,
+			expectError:      true,
+			skipCallsCheck:   true,
+		},
 		{
 			scenario:             "run tasks reaching max request limit causing restart",
 			initiallyStarted:     true,
+			isRestarting:         false,
 			processHealthy:       true,
 			maxReqLimit:          2,
 			tasksToRun:           3,
@@ -303,6 +326,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 			scenario:             "cannot restart after reaching max request limit",
 			startError:           errors.New("start error"),
 			initiallyStarted:     true,
+			isRestarting:         false,
 			processHealthy:       true,
 			maxReqLimit:          2,
 			tasksToRun:           2,
@@ -314,6 +338,7 @@ func TestProcessSupervisor_Run(t *testing.T) {
 		{
 			scenario:             "task error",
 			initiallyStarted:     true,
+			isRestarting:         false,
 			processHealthy:       true,
 			maxReqLimit:          0,
 			tasksToRun:           1,
@@ -351,6 +376,9 @@ func TestProcessSupervisor_Run(t *testing.T) {
 			if tc.initiallyStarted {
 				ps.firstStart.Store(true)
 			}
+			if tc.isRestarting {
+				ps.isRestarting.Store(true)
+			}
 
 			task := func() error {
 				return tc.taskError
@@ -386,6 +414,10 @@ func TestProcessSupervisor_Run(t *testing.T) {
 				}
 			}
 
+			if tc.skipCallsCheck {
+				return
+			}
+
 			if startCalls.Load() != tc.expectedStartCalls {
 				t.Errorf("expected %d process.Start calls, got %d", tc.expectedStartCalls, startCalls.Load())
 			}