Add outbox replay handling with configurable limits and timeouts

This commit is contained in:
Jeppe Bundgaard
2026-06-30 16:16:08 +02:00
parent 8ea10ef808
commit eb21405a3d
2 changed files with 32 additions and 9 deletions
@@ -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;
@@ -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');