From 36b4d435c35dd1506b16eeb5de7a0536029b7df9 Mon Sep 17 00:00:00 2001 From: Julien Neuhart Date: Thu, 14 Dec 2023 20:02:49 +0100 Subject: [PATCH] fix(supervisor): if process is already restarting, requeue a task (#754) --- pkg/gotenberg/supervisor.go | 104 ++++++++++++++++++------------- pkg/gotenberg/supervisor_test.go | 34 +++++++++- 2 files changed, 94 insertions(+), 44 deletions(-) diff --git a/pkg/gotenberg/supervisor.go b/pkg/gotenberg/supervisor.go index 809596ed..b55f47b4 100644 --- a/pkg/gotenberg/supervisor.go +++ b/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 { - case s.mutexChan <- struct{}{}: - logger.Debug("process lock acquired") - s.reqQueueSize.Add(-1) - s.reqCounter.Add(1) + for { + err := func() error { + select { + case s.mutexChan <- struct{}{}: + logger.Debug("process lock acquired") + s.reqQueueSize.Add(-1) + s.reqCounter.Add(1) - defer func() { - logger.Debug("process lock released") - <-s.mutexChan + defer func() { + logger.Debug("process lock released") + <-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() { - err := s.runWithDeadline(ctx, func() error { - return s.Launch() - }) - if err != nil { - return fmt.Errorf("process first start: %w", err) - } + if errors.Is(err, ErrProcessAlreadyRestarting) { + logger.Debug("process is already restarting, trying to acquire process lock again...") + s.reqQueueSize.Add(1) + continue } - 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) - } - } - - // 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()) + // Note: no error wrapping because it leaks on Chromium console exceptions output. + return err } } diff --git a/pkg/gotenberg/supervisor_test.go b/pkg/gotenberg/supervisor_test.go index 980db021..aa046d8c 100644 --- a/pkg/gotenberg/supervisor_test.go +++ b/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()) }