Default the worker to sending a full batch concurrently (#166)

A worker cycle is bounded by maximumSecondsPerWorkerCycle, but sending a
batch one request at a time costs batchSize * timeoutInSeconds, so the
previous defaults could not deliver a full batch inside their own budget:
the worker would consider itself stuck and exit, leaving the request it
was sending to be recovered minutes later.

The concurrent chunk size now defaults to the batch size, which is the
only value that satisfies the budget regardless of how batchSize is
tuned, and concurrentHttpEnabled becomes redundant next to a chunk size
of 1 - it is still honoured when explicitly disabled. Both settings, and
maximumSecondsPerWorkerCycle, are now readable from the environment.

Per-cycle database reconnects are also no longer the default. Lost
connections are instead detected and reconnected where they surface,
quietly for the first few in a row and reported once they start looking
like an outage rather than an idle timeout.

Author: LauJosefsen

README.md

@@ -19,7 +19,8 @@ own process dies mid-request — webhooks, downstream service calls, third-party
 |-----------------|------------------------|--------|
 | ^1              | ^7.4, ^8.0             | Security and bug fixes only |
 | ^2              | ^8.3                   | Security and bug fixes only |
-| ^3              | ^8.3                   | Active development |
+| ^3              | ^8.3                   | Security and bug fixes only |
+| ^4              | ^8.3                   | Active development |
 
 ## Installation
 
@@ -62,8 +63,30 @@ php artisan process:request-insurances
 ```
 
 Run one or many — workers coordinate through row locking (`SELECT … FOR UPDATE SKIP LOCKED` on
-MySQL 8+), pick requests in priority order, and process them in batches (`batchSize`, optionally
-with concurrent HTTP via `concurrentHttpEnabled`).
+MySQL 8+) and claim requests in priority order, `batchSize` at a time.
+
+### Cycle budget
+
+Each batch is sent concurrently in chunks of `concurrentHttpChunkSize`, which defaults to the whole
+batch. A cycle has `maximumSecondsPerWorkerCycle` (120s) to finish, after which the worker considers
+itself stuck and exits — leaving the chunk it was sending to be recovered by
+`request-insurance:unstuck-processing` some minutes later. The worst case is:
+
+```
+ceil(batchSize / concurrentHttpChunkSize) * timeoutInSeconds
+```
+
+The defaults therefore need a single `timeoutInSeconds` (20s) at worst, well inside the budget, while
+lowering the chunk size means checking that arithmetic against it. A chunk size of `1` sends requests
+strictly one at a time in priority order, which only fits a small batch or a short timeout.
+
+Concurrency also means requests within a batch complete in whatever order the receivers respond, so
+`priority` orders which requests get claimed, not which arrive first. The same goes for running more
+than one worker: a single worker with a chunk size of `1` is the only setup that delivers in strict
+order.
+
+The deprecated `concurrentHttpEnabled` setting is still honoured when set to `false`, which does the
+same thing as a chunk size of `1`.
 
 ### Lifecycle
 

publishable/config/request-insurance.php

@@ -65,16 +65,23 @@
     'batchSize' => env('REQUEST_INSURANCE_BATCH_SIZE', 100),
 
     /*
-    | Determines if concurrent http requests are enabled or not
+    | The maximum number of http requests to send concurrently.
+    |
+    | Null sends the entire batch at once. A value of 1 sends requests one at a time,
+    | in priority order, at the cost of a worker cycle long enough to do so - see
+    | maximumSecondsPerWorkerCycle.
     */
 
-    'concurrentHttpEnabled' => false,
+    'concurrentHttpChunkSize' => env('REQUEST_INSURANCE_CONCURRENT_HTTP_CHUNK_SIZE'),
 
     /*
-    | The maximum number of http requests to send concurrently
+    | Sets how long a single worker cycle may take, before the worker considers itself
+    | stuck and exits. A cycle must be able to send a full batch within the budget:
+    |
+    |     ceil(batchSize / concurrentHttpChunkSize) * timeoutInSeconds
     */
 
-    'concurrentHttpChunkSize' => 5,
+    'maximumSecondsPerWorkerCycle' => env('REQUEST_INSURANCE_MAX_SECONDS_PER_WORKER_CYCLE', 120),
 
     /*
      | Set the concrete implementation for HttpRequest
@@ -106,7 +113,15 @@
     'table_edits'          => null,
     'table_edit_approvals' => null,
 
-    'useDbReconnect' => env('REQUEST_INSURANCE_WORKER_USE_DB_RECONNECT', true),
+    /*
+     | Sets if the worker should reconnect to the database at the start of every cycle.
+     |
+     | Lost connections are detected and recovered from either way, so this is only needed
+     | to stop workers from holding on to a connection they should not keep - such as one
+     | pinned to a single node behind a load balancer.
+     */
+
+    'useDbReconnect' => env('REQUEST_INSURANCE_WORKER_USE_DB_RECONNECT', false),
 
     /*
      | Using skip locked optimizes request insurance to run with multiple worker threads,

src/RequestInsurance/RequestInsuranceWorker.php

@@ -13,14 +13,23 @@
 use Illuminate\Support\Facades\Log;
 use Cego\RequestInsurance\Enums\State;
 use Illuminate\Support\Facades\Config;
+use Illuminate\Database\DetectsLostConnections;
 use Cego\RequestInsurance\Models\RequestInsurance;
 use Cego\RequestInsurance\AsyncRequests\RequestInsuranceClient;
 use Illuminate\Database\Eloquent\Collection as EloquentCollection;
 
 class RequestInsuranceWorker
 {
+    use DetectsLostConnections;
+
     private const TIMEOUT_EXIT_CODE = 124;
 
+    /**
+     * The number of consecutive lost database connections to recover from quietly,
+     * before treating them as an outage worth reporting
+     */
+    private const QUIET_LOST_CONNECTION_RECOVERIES = 3;
+
     /**
      * Holds a hash identifier for the service instance once set
      *
@@ -49,6 +58,8 @@ class RequestInsuranceWorker
 
     protected ?string $currentRequestInsuranceIds = null;
 
+    protected int $consecutiveLostConnections = 0;
+
     /**
      * RequestInsuranceService constructor.
      */
@@ -79,7 +90,7 @@ public function run(bool $runOnlyOnce = false): void
 
                 if (Config::get('request-insurance.useDbReconnect')) {
                     $this->setWorkerPhase('database_reconnect');
-                    DB::reconnect();
+                    $this->reconnectToDatabase();
                 }
 
                 $start = hrtime(true);
@@ -90,6 +101,8 @@ public function run(bool $runOnlyOnce = false): void
                     $this->readyWaitingRequestInsurances();
                 });
 
+                $this->consecutiveLostConnections = 0;
+
                 $executionTimeNs = hrtime(true) - $start;
 
                 $waitTime = (int) max(Config::get('request-insurance.microSecondsToWait') - ($executionTimeNs / 1000), 0);
@@ -99,7 +112,7 @@ public function run(bool $runOnlyOnce = false): void
             } catch (Throwable $throwable) {
                 $this->resetTimeoutHandler(); // We need to reset here before logging the error and sleeping, otherwise the timeout handler might trigger while we are sleeping/logging, which is not desirable.
 
-                Log::error($throwable);
+                $this->handleCycleFailure($throwable);
 
                 if ($runOnlyOnce) {
                     throw $throwable;
@@ -114,6 +127,72 @@ public function run(bool $runOnlyOnce = false): void
         Log::info(sprintf('RequestInsurance Worker (#%s) has gracefully stopped', $this->runningHash));
     }
 
+    /**
+     * Handles a failed worker cycle
+     *
+     * A dropped database connection is expected of a long-lived worker, and is recovered from
+     * without noise, until it keeps happening and starts looking like an outage instead.
+     *
+     * @param Throwable $throwable
+     *
+     * @return void
+     */
+    protected function handleCycleFailure(Throwable $throwable): void
+    {
+        if ( ! $this->wasCausedByLostConnection($throwable)) {
+            $this->consecutiveLostConnections = 0;
+
+            Log::error($throwable);
+
+            return;
+        }
+
+        $this->consecutiveLostConnections++;
+
+        $message = sprintf(
+            'RequestInsurance Worker (#%s) lost its database connection during %s and is reconnecting (%d in a row)',
+            $this->runningHash,
+            $this->currentPhase,
+            $this->consecutiveLostConnections
+        );
+
+        if ($this->consecutiveLostConnections > self::QUIET_LOST_CONNECTION_RECOVERIES) {
+            Log::error($message, ['exception' => $throwable]);
+        } else {
+            Log::debug($message);
+        }
+
+        rescue(fn () => $this->reconnectToDatabase(), null, false);
+    }
+
+    /**
+     * Tells if the given throwable, or anything it wraps, was caused by a lost database connection
+     *
+     * @param Throwable $throwable
+     *
+     * @return bool
+     */
+    protected function wasCausedByLostConnection(Throwable $throwable): bool
+    {
+        for ($exception = $throwable; $exception !== null; $exception = $exception->getPrevious()) {
+            if ($this->causedByLostConnection($exception)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Reconnects the database connection holding the request insurance tables
+     *
+     * @return void
+     */
+    protected function reconnectToDatabase(): void
+    {
+        DB::reconnect(resolve(RequestInsurance::class)->getConnectionName());
+    }
+
     /**
      * Sets up signal handler to make sure that request insurance can shutdown gracefully.
      *
@@ -264,17 +343,21 @@ protected function setStateToProcessingAndIncrementAttempts(EloquentCollection $
     }
 
     /**
-     * Returns the concurrent request chunk size
+     * Returns the number of requests to send concurrently
      *
      * @return int
      */
     protected function getRequestChunkSize(): int
     {
-        if (Config::get('request-insurance.concurrentHttpEnabled', false)) {
-            return Config::get('request-insurance.concurrentHttpChunkSize', 5);
+        // Deprecated flag, still honoured for configs that set it explicitly
+        if (Config::get('request-insurance.concurrentHttpEnabled') === false) {
+            return 1;
         }
 
-        return 1;
+        $chunkSize = Config::get('request-insurance.concurrentHttpChunkSize')
+            ?? Config::get('request-insurance.batchSize', 100);
+
+        return max(1, (int) $chunkSize);
     }
 
     /**
@@ -416,7 +499,7 @@ private function registerTimeoutHandler()
         pcntl_signal(SIGALRM, function () {
             $this->handleTimeoutSignal();
         });
-        pcntl_alarm(Config::integer('request-insurance.maximumSecondsPerWorkerCycle', 120));
+        pcntl_alarm((int) Config::get('request-insurance.maximumSecondsPerWorkerCycle', 120));
     }
 
     private function resetTimeoutHandler(): void

tests/Unit/RequestInsuranceWorkerTest.php

@@ -2,8 +2,11 @@
 
 namespace Tests\Unit;
 
+use Exception;
+use Throwable;
 use Tests\TestCase;
 use GuzzleHttp\Psr7\Request;
+use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Facades\Http;
 use Illuminate\Support\Facades\Crypt;
 use Illuminate\Support\Facades\Event;
@@ -434,4 +437,100 @@ protected function terminateWorkerAfterTimeout(): void
             ['terminate'],
         ], $worker->events);
     }
+
+    public function test_the_request_chunk_size_defaults_to_the_batch_size(): void
+    {
+        Config::set('request-insurance.batchSize', 42);
+        Config::set('request-insurance.concurrentHttpChunkSize', null);
+
+        $this->assertSame(42, $this->getWorkerProbe()->exposeGetRequestChunkSize());
+    }
+
+    public function test_the_request_chunk_size_can_be_set_below_the_batch_size(): void
+    {
+        Config::set('request-insurance.batchSize', 42);
+        Config::set('request-insurance.concurrentHttpChunkSize', 7);
+
+        $this->assertSame(7, $this->getWorkerProbe()->exposeGetRequestChunkSize());
+    }
+
+    public function test_the_deprecated_concurrent_http_enabled_flag_still_disables_concurrency(): void
+    {
+        Config::set('request-insurance.batchSize', 42);
+        Config::set('request-insurance.concurrentHttpEnabled', false);
+
+        $this->assertSame(1, $this->getWorkerProbe()->exposeGetRequestChunkSize());
+    }
+
+    public function test_it_recovers_quietly_from_a_lost_database_connection(): void
+    {
+        $worker = $this->getWorkerProbe();
+
+        Log::shouldReceive('debug')
+            ->once()
+            ->with('RequestInsurance Worker (#' . $worker->exposeRunningHash() . ') lost its database connection during idle and is reconnecting (1 in a row)');
+
+        Log::shouldReceive('error')->never();
+
+        $worker->exposeReportCycleFailure(new Exception('Cycle failed', 0, new Exception('MySQL server has gone away')));
+
+        $this->assertSame(1, $worker->reconnects);
+    }
+
+    public function test_it_reports_lost_database_connections_that_keep_happening(): void
+    {
+        $worker = $this->getWorkerProbe();
+
+        Log::shouldReceive('debug')->times(3);
+        Log::shouldReceive('error')->twice();
+
+        foreach (range(1, 5) as $ignored) {
+            $worker->exposeReportCycleFailure(new Exception('Lost connection to server'));
+        }
+
+        $this->assertSame(5, $worker->reconnects);
+    }
+
+    public function test_it_reports_cycle_failures_that_are_not_lost_connections(): void
+    {
+        $worker = $this->getWorkerProbe();
+        $throwable = new Exception('Something else went wrong');
+
+        Log::shouldReceive('error')->once()->with($throwable);
+        Log::shouldReceive('debug')->never();
+
+        $worker->exposeReportCycleFailure($throwable);
+
+        $this->assertSame(0, $worker->reconnects);
+    }
+
+    /**
+     * Returns a worker exposing the internals needed to assert on cycle failure handling
+     */
+    private function getWorkerProbe(): RequestInsuranceWorker
+    {
+        return new class () extends RequestInsuranceWorker {
+            public int $reconnects = 0;
+
+            public function exposeGetRequestChunkSize(): int
+            {
+                return $this->getRequestChunkSize();
+            }
+
+            public function exposeReportCycleFailure(Throwable $throwable): void
+            {
+                $this->handleCycleFailure($throwable);
+            }
+
+            public function exposeRunningHash(): ?string
+            {
+                return $this->runningHash;
+            }
+
+            protected function reconnectToDatabase(): void
+            {
+                $this->reconnects++;
+            }
+        };
+    }
 }