Fallback relay commands after broker dispatch failures

This commit is contained in:
Jeppe Bundgaard
2026-06-30 12:32:10 +02:00
parent acc80920f9
commit d902202fe9
2 changed files with 55 additions and 1 deletions
@@ -5638,6 +5638,7 @@ BASH;
$delivery = (array)($job->delivery_json->value() ?? []); $delivery = (array)($job->delivery_json->value() ?? []);
$preferredChannel = (string)($delivery['preferred_channel'] ?? self::DELIVERY_CHANNEL_API); $preferredChannel = (string)($delivery['preferred_channel'] ?? self::DELIVERY_CHANNEL_API);
$requireFastPath = !empty($delivery['require_fast_path']); $requireFastPath = !empty($delivery['require_fast_path']);
$waitTimeoutSeconds = self::COMMAND_WAIT_TIMEOUT_SECONDS;
$effectiveStatus = self::resolveGatewayStatus( $effectiveStatus = self::resolveGatewayStatus(
$gateway->status->value() === null ? null : (string)$gateway->status->value(), $gateway->status->value() === null ? null : (string)$gateway->status->value(),
@@ -5673,6 +5674,24 @@ BASH;
); );
throw $exception; throw $exception;
} }
if ($this->commandDisallowsAutomaticRetry($job)) {
$this->finalizeCommandJob(
$job,
false,
[],
$exception->getMessage(),
$gateway,
$this->isCommandTimeoutError($exception->getMessage()) ? 'TIMED_OUT' : 'FAILED'
);
throw $exception;
}
$this->requeueCommandForApiPolling($job, $exception->getMessage());
$waitTimeoutSeconds = max(
self::COMMAND_WAIT_TIMEOUT_SECONDS,
self::COMMAND_POLL_TIMEOUT_SECONDS + 5
);
} }
} }
} }
@@ -5682,7 +5701,35 @@ BASH;
throw new Exception('Edge broker fast path is unavailable'); throw new Exception('Edge broker fast path is unavailable');
} }
return $this->waitForCommandResult((int)$job->id); return $this->waitForCommandResult((int)$job->id, $waitTimeoutSeconds);
}
private function commandDisallowsAutomaticRetry(edge_gateway_command_jobs_o $job): bool
{
$delivery = (array)($job->delivery_json->value() ?? []);
return filter_var($delivery['no_auto_retry'] ?? false, FILTER_VALIDATE_BOOLEAN);
}
private function requeueCommandForApiPolling(edge_gateway_command_jobs_o $job, string $errorMessage): void
{
$delivery = $this->buildDeliveryMetadata(
array_merge(
(array)($job->delivery_json->value() ?? []),
[
'preferred_channel' => self::DELIVERY_CHANNEL_API,
'delivery_channel' => null,
'fallback_reason' => 'broker_dispatch_failed',
'last_dispatch_error' => $errorMessage,
]
),
self::COMMAND_EXPIRES_AFTER_SECONDS
);
$job->status->set('PENDING');
$job->response_json->set([]);
$job->completed_at->set(null);
$job->error_message->set(null);
$job->delivery_json->set($delivery);
} }
private function isCommandTimeoutError(string $errorMessage): bool private function isCommandTimeoutError(string $errorMessage): bool
@@ -42,6 +42,13 @@ it('keeps relay dispatch and discovery queueing on the edge gateway manager', fu
expect($managerSource)->toContain("'batch_dedupe_key' => \$batchDedupeKey"); expect($managerSource)->toContain("'batch_dedupe_key' => \$batchDedupeKey");
expect($managerSource)->toContain("'idempotency_key' => \$batchId . ':' . (string)\$command['target']"); expect($managerSource)->toContain("'idempotency_key' => \$batchId . ':' . (string)\$command['target']");
expect($managerSource)->toContain("'no_auto_retry' => \$isGatePulse"); expect($managerSource)->toContain("'no_auto_retry' => \$isGatePulse");
expect($managerSource)->toContain('private function commandDisallowsAutomaticRetry');
expect($managerSource)->toContain('private function requeueCommandForApiPolling');
expect($managerSource)->toContain("'preferred_channel' => self::DELIVERY_CHANNEL_API");
expect($managerSource)->toContain("'fallback_reason' => 'broker_dispatch_failed'");
expect($managerSource)->toContain("\$this->requeueCommandForApiPolling(\$job, \$exception->getMessage());");
expect($managerSource)->toContain('self::COMMAND_POLL_TIMEOUT_SECONDS + 5');
expect($managerSource)->toContain('return $this->waitForCommandResult((int)$job->id, $waitTimeoutSeconds);');
expect($managerSource)->toContain('private function buildRelayBatchDedupeKey'); expect($managerSource)->toContain('private function buildRelayBatchDedupeKey');
expect($managerSource)->toContain('private function findRecentRelayBatchByDedupeKey'); expect($managerSource)->toContain('private function findRecentRelayBatchByDedupeKey');
expect($managerSource)->toContain('private function enforceRelayCommandRateLimit'); expect($managerSource)->toContain('private function enforceRelayCommandRateLimit');