From eb21405a3dd638a026d7e39d43e5e92331ebf7c1 Mon Sep 17 00:00:00 2001
From: Jeppe Bundgaard <jb@truckwash.dk>
Date: Tue, 30 Jun 2026 16:16:08 +0200
Subject: [PATCH] Add outbox replay handling with configurable limits and
 timeouts

---
 .../resources/edge-gateway-agent/agent.php    | 39 +++++++++++++++----
 .../EdgeGatewayUpdateLifecycleTest.php        |  2 +-
 2 files changed, 32 insertions(+), 9 deletions(-)

diff --git a/services/nginx/app/resources/edge-gateway-agent/agent.php b/services/nginx/app/resources/edge-gateway-agent/agent.php
index 6589bc5c..07d24f1c 100644
--- a/services/nginx/app/resources/edge-gateway-agent/agent.php
+++ b/services/nginx/app/resources/edge-gateway-agent/agent.php
@@ -997,6 +997,11 @@ final class TruckwashEdgeAgent
     private const DEFAULT_WORKER_BASE_URL = 'http://lan-worker:8090';
     private const DEFAULT_UPDATE_WINDOW = '02:00-04:00';
     private const OPERATION_COMPLETE_TIMEOUT_SECONDS = 120;
+    private const OUTBOX_REPLAY_BATCH_LIMIT = 3;
+    private const OUTBOX_REPLAY_TIMEOUT_SECONDS = 3;
+    private const OUTBOX_OPERATION_COMPLETE_REPLAY_TIMEOUT_SECONDS = 10;
+    private const OUTBOX_REPLAY_FAILURE_COOLDOWN_SECONDS = 15;
+    private const MACHINE_SIGNAL_TIMEOUT_SECONDS = 3;
     private const BROKER_MESSAGE_PUMP_LIMIT = 12;
     private const LOOP_STALE_AFTER_SECONDS = 30;
     private const CONTROL_PLANE_SYNC_STALE_AFTER_SECONDS = 90;
@@ -1019,6 +1024,7 @@ final class TruckwashEdgeAgent
     private int $lastHeartbeatAt = 0;
     private int $lastMachineSignalPollAt = 0;
     private int $lastMachineSignalMonitorRefreshAt = 0;
+    private int $lastOutboxFailureAt = 0;
     private ?array $lastControlPlaneResponse = null;
     private string $agentInstanceId;
 
@@ -1062,7 +1068,6 @@ final class TruckwashEdgeAgent
                 $this->reloadConfigFromDisk();
                 $this->ensureClaimed();
                 $this->configureBrokerClient();
-                $this->flushOutbox();
                 $this->pumpBrokerTransport();
                 $this->heartbeat();
                 $this->pollMachineStartSignals();
@@ -1357,7 +1362,8 @@ final class TruckwashEdgeAgent
                     $this->sendControlPlaneEvent(
                         '/edge-agent/gateways/' . $gatewayId . '/selfserve/machine-signal',
                         $payload,
-                        'machine_signal'
+                        'machine_signal',
+                        self::MACHINE_SIGNAL_TIMEOUT_SECONDS
                     );
                 }
 
@@ -2452,12 +2458,14 @@ final class TruckwashEdgeAgent
 
     private function flushOutbox(): void
     {
-        $items = $this->stateStore->queuedItems(25);
+        if ($this->shouldSkipOutboxReplay()) {
+            return;
+        }
+
+        $items = $this->stateStore->queuedItems(self::OUTBOX_REPLAY_BATCH_LIMIT);
         foreach ($items as $item) {
             try {
-                $timeoutSeconds = (string)($item['type'] ?? '') === 'operation_complete'
-                    ? self::OPERATION_COMPLETE_TIMEOUT_SECONDS
-                    : 20;
+                $timeoutSeconds = $this->outboxReplayTimeoutSeconds((string)($item['type'] ?? ''));
                 $endpoint = (string)$item['endpoint'];
                 $payload = is_array($item['payload'] ?? null) ? (array)$item['payload'] : [];
                 $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
@@ -2465,8 +2473,10 @@ final class TruckwashEdgeAgent
                     $this->http->post($endpoint, $payload, $timeoutSeconds);
                 }
                 $this->stateStore->removeOutboxItem((int)$item['id']);
+                $this->lastOutboxFailureAt = 0;
                 $this->recordSuccessfulSync();
             } catch (Throwable $throwable) {
+                $this->lastOutboxFailureAt = time();
                 $this->recordTransportFailure(
                     'Outbox replay blocked on ' . (string)$item['type'] . ' for ' . (string)$item['endpoint'],
                     $throwable
@@ -2476,6 +2486,19 @@ final class TruckwashEdgeAgent
         }
     }
 
+    private function shouldSkipOutboxReplay(): bool
+    {
+        return $this->lastOutboxFailureAt > 0
+            && (time() - $this->lastOutboxFailureAt) < self::OUTBOX_REPLAY_FAILURE_COOLDOWN_SECONDS;
+    }
+
+    private function outboxReplayTimeoutSeconds(string $type): int
+    {
+        return $type === 'operation_complete'
+            ? self::OUTBOX_OPERATION_COMPLETE_REPLAY_TIMEOUT_SECONDS
+            : self::OUTBOX_REPLAY_TIMEOUT_SECONDS;
+    }
+
     private function probeWorkerHealth(): array
     {
         try {
@@ -2564,7 +2587,7 @@ final class TruckwashEdgeAgent
         ];
     }
 
-    private function sendControlPlaneEvent(string $endpoint, array $payload, string $type): bool
+    private function sendControlPlaneEvent(string $endpoint, array $payload, string $type, int $timeoutSeconds = 20): bool
     {
         $this->lastControlPlaneResponse = null;
         $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
@@ -2573,7 +2596,7 @@ final class TruckwashEdgeAgent
         }
 
         try {
-            $response = $this->http->post($endpoint, $payload, 20);
+            $response = $this->http->post($endpoint, $payload, $timeoutSeconds);
             $this->lastControlPlaneResponse = is_array($response) ? $response : null;
             $this->recordSuccessfulSync();
             return true;
diff --git a/services/nginx/app/tests/Unit/Selfserve/EdgeGatewayUpdateLifecycleTest.php b/services/nginx/app/tests/Unit/Selfserve/EdgeGatewayUpdateLifecycleTest.php
index 5087d81d..28b06195 100644
--- a/services/nginx/app/tests/Unit/Selfserve/EdgeGatewayUpdateLifecycleTest.php
+++ b/services/nginx/app/tests/Unit/Selfserve/EdgeGatewayUpdateLifecycleTest.php
@@ -205,7 +205,7 @@ it('exposes update payload, credential rotation, cancel endpoints, and operation
     expect($agentSource)->toContain('private function dispatchOperationCompletion(array $completion, bool $queueOnFailure): bool');
     expect($agentSource)->toContain("\$state['status'] = 'COMPLETION_PENDING';");
     expect($agentSource)->toContain("\$state['stage'] = 'awaiting_completion_ack';");
-    expect($agentSource)->toContain("? self::OPERATION_COMPLETE_TIMEOUT_SECONDS");
+    expect($agentSource)->toContain("? self::OUTBOX_OPERATION_COMPLETE_REPLAY_TIMEOUT_SECONDS");
     expect($agentSource)->toContain("unset(\$state['completion']);");
     expect($agentSource)->toContain('$services[] = $this->probeTcpService(\'redis\', \'redis\', 6379);');
     expect($agentSource)->toContain('final class HttpRequestTimeoutException extends RuntimeException');