diff --git a/Makefile b/Makefile index 3df44c3d..8a186041 100644 --- a/Makefile +++ b/Makefile @@ -31,8 +31,9 @@ API-DOWNLOAD-FROM-FROM-MAX-RETRY=4 API-DISABLE-DOWNLOAD-FROM=false API_DISABLE_HEALTH_CHECK_LOGGING=false API_ENABLE_DEBUG_ROUTE=false -CHROMIUM_RESTART_AFTER=10 +CHROMIUM_RESTART_AFTER=100 CHROMIUM_MAX_QUEUE_SIZE=0 +CHROMIUM_MAX_CONCURRENCY=6 CHROMIUM_AUTO_START=false CHROMIUM_START_TIMEOUT=20s CHROMIUM_ALLOW_INSECURE_LOCALHOST=false @@ -111,6 +112,7 @@ run: ## Start a Gotenberg container --chromium-restart-after=$(CHROMIUM_RESTART_AFTER) \ --chromium-auto-start=$(CHROMIUM_AUTO_START) \ --chromium-max-queue-size=$(CHROMIUM_MAX_QUEUE_SIZE) \ + --chromium-max-concurrency=$(CHROMIUM_MAX_CONCURRENCY) \ --chromium-start-timeout=$(CHROMIUM_START_TIMEOUT) \ --chromium-allow-insecure-localhost=$(CHROMIUM_ALLOW_INSECURE_LOCALHOST) \ --chromium-ignore-certificate-errors=$(CHROMIUM_IGNORE_CERTIFICATE_ERRORS) \ @@ -166,6 +168,7 @@ PLATFORM= NO_CONCURRENCY=false # Available tags: # chromium +# chromium-concurrent # chromium-convert-html # chromium-convert-markdown # chromium-convert-url diff --git a/pkg/gotenberg/supervisor.go b/pkg/gotenberg/supervisor.go index 66c2ce62..63998fe7 100644 --- a/pkg/gotenberg/supervisor.go +++ b/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() { diff --git a/pkg/gotenberg/supervisor_test.go b/pkg/gotenberg/supervisor_test.go index 56368418..123a9160 100644 --- a/pkg/gotenberg/supervisor_test.go +++ b/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()) + } +} diff --git a/pkg/modules/chromium/chromium.go b/pkg/modules/chromium/chromium.go index 4e5a6a0f..2645eefb 100644 --- a/pkg/modules/chromium/chromium.go +++ b/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) diff --git a/pkg/modules/libreoffice/api/api.go b/pkg/modules/libreoffice/api/api.go index 2e0529b4..aeb6e1bf 100644 --- a/pkg/modules/libreoffice/api/api.go +++ b/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 } diff --git a/test/integration/features/chromium_concurrent.feature b/test/integration/features/chromium_concurrent.feature new file mode 100644 index 00000000..572b5846 --- /dev/null +++ b/test/integration/features/chromium_concurrent.feature @@ -0,0 +1,20 @@ +@chromium +@chromium-concurrent +Feature: Chromium concurrent conversions + + Scenario: Concurrent HTML to PDF conversions with max concurrency 3 + Given I have a Gotenberg container with the following environment variable(s): + | CHROMIUM_MAX_CONCURRENCY | 3 | + When I make 3 concurrent "POST" requests to Gotenberg at the "/forms/chromium/convert/html" endpoint with the following form data and header(s): + | files | testdata/page-1-html/index.html | file | + Then all concurrent response status codes should be 200 + Then all concurrent responses should have 1 PDF(s) + + Scenario: Concurrent conversions exceeding restart-after limit + Given I have a Gotenberg container with the following environment variable(s): + | CHROMIUM_MAX_CONCURRENCY | 3 | + | CHROMIUM_RESTART_AFTER | 5 | + When I make 10 concurrent "POST" requests to Gotenberg at the "/forms/chromium/convert/html" endpoint with the following form data and header(s): + | files | testdata/page-1-html/index.html | file | + Then all concurrent response status codes should be 200 + Then all concurrent responses should have 1 PDF(s) diff --git a/test/integration/features/debug.feature b/test/integration/features/debug.feature index 6a4dce55..b978fea3 100644 --- a/test/integration/features/debug.feature +++ b/test/integration/features/debug.feature @@ -204,8 +204,9 @@ Feature: /debug "chromium-ignore-certificate-errors": "false", "chromium-incognito": "false", "chromium-max-queue-size": "0", + "chromium-max-concurrency": "6", "chromium-proxy-server": "", - "chromium-restart-after": "10", + "chromium-restart-after": "100", "chromium-start-timeout": "20s", "gotenberg-build-debug-data": "true", "gotenberg-graceful-shutdown-duration": "30s", diff --git a/test/integration/scenario/scenario.go b/test/integration/scenario/scenario.go index 32336f90..0599bcfa 100644 --- a/test/integration/scenario/scenario.go +++ b/test/integration/scenario/scenario.go @@ -14,6 +14,7 @@ import ( "regexp" "strconv" "strings" + "sync" "time" "github.com/cucumber/godog" @@ -24,6 +25,7 @@ import ( type scenario struct { resp *httptest.ResponseRecorder + concurrentResps []*httptest.ResponseRecorder workdir string gotenbergContainer testcontainers.Container gotenbergContainerNetwork *testcontainers.DockerNetwork @@ -33,6 +35,7 @@ type scenario struct { func (s *scenario) reset(ctx context.Context) error { s.resp = httptest.NewRecorder() + s.concurrentResps = nil err := os.RemoveAll(s.workdir) if err != nil { @@ -281,6 +284,171 @@ func (s *scenario) iMakeARequestToGotenbergWithTheFollowingFormDataAndHeaders(ct return nil } +func (s *scenario) iMakeConcurrentRequestsToGotenberg(ctx context.Context, count int, method, endpoint string, dataTable *godog.Table) error { + if s.gotenbergContainer == nil { + return errors.New("no Gotenberg container") + } + + fields := make(map[string]string) + files := make(map[string][]string) + headers := make(map[string]string) + + for _, row := range dataTable.Rows { + name := row.Cells[0].Value + value := row.Cells[1].Value + kind := row.Cells[2].Value + + switch kind { + case "field": + fields[name] = value + case "file": + wd, err := os.Getwd() + if err != nil { + return fmt.Errorf("get current directory: %w", err) + } + value = fmt.Sprintf("%s/%s", wd, value) + files[name] = append(files[name], value) + case "header": + headers[name] = value + default: + return fmt.Errorf("unexpected %q %q", kind, value) + } + } + + base, err := containerHttpEndpoint(ctx, s.gotenbergContainer, "3000") + if err != nil { + return fmt.Errorf("get container HTTP endpoint: %w", err) + } + + var ( + mu sync.Mutex + wg sync.WaitGroup + ) + s.concurrentResps = make([]*httptest.ResponseRecorder, 0, count) + errs := make([]error, 0) + + for i := 0; i < count; i++ { + wg.Add(1) + go func() { + defer wg.Done() + + resp, reqErr := doFormDataRequest(method, fmt.Sprintf("%s%s", base, endpoint), fields, files, headers) + if reqErr != nil { + mu.Lock() + errs = append(errs, fmt.Errorf("do request: %w", reqErr)) + mu.Unlock() + return + } + defer resp.Body.Close() + + body, reqErr := io.ReadAll(resp.Body) + if reqErr != nil { + mu.Lock() + errs = append(errs, fmt.Errorf("read response body: %w", reqErr)) + mu.Unlock() + return + } + + rec := httptest.NewRecorder() + rec.Code = resp.StatusCode + for key, values := range resp.Header { + for _, v := range values { + rec.Header().Add(key, v) + } + } + _, _ = rec.Body.Write(body) + + if resp.StatusCode == http.StatusOK { + cd := resp.Header.Get("Content-Disposition") + if cd != "" { + _, params, parseErr := mime.ParseMediaType(cd) + if parseErr == nil { + if filename, ok := params["filename"]; ok { + traceID := resp.Header.Get("Gotenberg-Trace") + dirPath := fmt.Sprintf("%s/%s", s.workdir, traceID) + + mu.Lock() + mkErr := os.MkdirAll(dirPath, 0o755) + mu.Unlock() + + if mkErr == nil { + fpath := fmt.Sprintf("%s/%s", dirPath, filename) + f, fErr := os.Create(fpath) + if fErr == nil { + _, _ = f.Write(body) + f.Close() + } + } + } + } + } + } + + mu.Lock() + s.concurrentResps = append(s.concurrentResps, rec) + mu.Unlock() + }() + } + + wg.Wait() + + if len(errs) > 0 { + return fmt.Errorf("concurrent requests failed: %v", errs) + } + + return nil +} + +func (s *scenario) allConcurrentResponseStatusCodesShouldBe(expected int) error { + if len(s.concurrentResps) == 0 { + return errors.New("no concurrent responses recorded") + } + + for i, resp := range s.concurrentResps { + if resp.Code != expected { + return fmt.Errorf("concurrent response %d: expected status %d, got %d %q", i+1, expected, resp.Code, resp.Body.String()) + } + } + + return nil +} + +func (s *scenario) allConcurrentResponsesShouldHavePdfs(expected int) error { + if len(s.concurrentResps) == 0 { + return errors.New("no concurrent responses recorded") + } + + for i, resp := range s.concurrentResps { + traceID := resp.Header().Get("Gotenberg-Trace") + dirPath := fmt.Sprintf("%s/%s", s.workdir, traceID) + + _, err := os.Stat(dirPath) + if os.IsNotExist(err) { + return fmt.Errorf("concurrent response %d: directory %q does not exist", i+1, dirPath) + } + + var paths []string + err = filepath.Walk(dirPath, func(path string, info os.FileInfo, pathErr error) error { + if pathErr != nil { + return pathErr + } + if strings.EqualFold(filepath.Ext(info.Name()), ".pdf") { + paths = append(paths, path) + } + return nil + }) + if err != nil { + return fmt.Errorf("concurrent response %d: walk %q: %w", i+1, dirPath, err) + } + + if len(paths) != expected { + return fmt.Errorf("concurrent response %d: expected %d PDF(s), got %d", i+1, expected, len(paths)) + } + } + + return nil +} + func (s *scenario) iWaitForTheAsynchronousRequestToWebhook(ctx context.Context) error { if s.server == nil { return errors.New("server not initialized") @@ -965,9 +1133,12 @@ func InitializeScenario(ctx *godog.ScenarioContext) { ctx.When(`^I make a "(GET|HEAD)" request to Gotenberg at the "([^"]*)" endpoint$`, s.iMakeARequestToGotenberg) ctx.When(`^I make a "(GET|HEAD)" request to Gotenberg at the "([^"]*)" endpoint with the following header\(s\):$`, s.iMakeARequestToGotenbergWithTheFollowingHeaders) ctx.When(`^I make a "(POST)" request to Gotenberg at the "([^"]*)" endpoint with the following form data and header\(s\):$`, s.iMakeARequestToGotenbergWithTheFollowingFormDataAndHeaders) + ctx.When(`^I make (\d+) concurrent "(POST)" requests to Gotenberg at the "([^"]*)" endpoint with the following form data and header\(s\):$`, s.iMakeConcurrentRequestsToGotenberg) ctx.When(`^I wait for the asynchronous request to the webhook$`, s.iWaitForTheAsynchronousRequestToWebhook) ctx.Then(`^the Gotenberg container (should|should NOT) log the following entries:$`, s.theGotenbergContainerShouldLogTheFollowingEntries) ctx.Then(`^the response status code should be (\d+)$`, s.theResponseStatusCodeShouldBe) + ctx.Then(`^all concurrent response status codes should be (\d+)$`, s.allConcurrentResponseStatusCodesShouldBe) + ctx.Then(`^all concurrent responses should have (\d+) PDF\(s\)$`, s.allConcurrentResponsesShouldHavePdfs) ctx.Then(`^the (response|webhook request|file request|server request) header "([^"]*)" should be "([^"]*)"$`, s.theHeaderValueShouldBe) ctx.Then(`^the (response|webhook request|file request|server request) cookie "([^"]*)" should be "([^"]*)"$`, s.theCookieValueShouldBe) ctx.Then(`^the (response|webhook request) body should match string:$`, s.theBodyShouldMatchString)