Handle relay command job timeouts for edge gateways

- Introduce `expireTimedOutRelayStatusCommandJobs` to clean up long-pending relay status command jobs. - Add `TIMED_OUT` status for relay command jobs and incorporate it into job status evaluations. - Refactor command job finalization to support timeout-specific error messaging. - Improve handling of fast-path failures in edge broker commands.
2026-04-27 16:47:09 +02:00
parent 86eec9a51e
commit feb9aac4f7
16 changed files with 636 additions and 108 deletions
@@ -497,6 +497,7 @@ class edge_gateway_manager
        $operations = new edge_gateway_operation_service($this);
        $data['metadata']['broker_presence'] = $this->readBrokerPresence((int)$gateway->id);
        $gatewayId = (int)$gateway->id;
+        $this->expireTimedOutRelayStatusCommandJobs($gatewayId);
        if ($includeDetail) {
            $data['inventory'] = $this->listInventory($gatewayId);
            $data['bindings'] = $this->listBindings($gatewayId);
@@ -801,7 +802,7 @@ class edge_gateway_manager
        }

        $status = (string)$job->status->value();
-        if (in_array($status, ['COMPLETED', 'FAILED'], true)) {
+        if (in_array($status, ['COMPLETED', 'FAILED', 'TIMED_OUT'], true)) {
            return [
                'acknowledged' => true,
                'job' => $job->asArray(),
@@ -2765,6 +2766,41 @@ BASH;
        return $jobObject->select($jobId);
    }

+    private function expireTimedOutRelayStatusCommandJobs(int $gatewayId): void
+    {
+        if ($gatewayId <= 0) {
+            return;
+        }
+
+        $statement = db::getPDO()->prepare(
+            "SELECT id
+             FROM edge_gateway_command_jobs
+             WHERE gateway_id = :gateway_id
+               AND deleted_at IS NULL
+               AND command_type = 'GET_RELAY_STATUS'
+               AND status IN ('PENDING', 'DISPATCHING')
+               AND requested_at <= :cutoff
+             ORDER BY requested_at ASC, id ASC
+             LIMIT 200"
+        );
+        $statement->execute([
+            ':gateway_id' => $gatewayId,
+            ':cutoff' => $this->formatDateTime(time() - self::COMMAND_WAIT_TIMEOUT_SECONDS),
+        ]);
+
+        foreach ($statement->fetchAll() ?: [] as $row) {
+            $jobId = (int)($row['id'] ?? 0);
+            if ($jobId <= 0) {
+                continue;
+            }
+
+            $job = (new edge_gateway_command_jobs_o())->select($jobId);
+            if ($job->exists()) {
+                $this->finalizeCommandJob($job, false, [], 'Edge gateway command timed out', null, 'TIMED_OUT');
+            }
+        }
+    }
+
    /**
     * @throws Exception
     */
@@ -2788,6 +2824,10 @@ BASH;
                $errorMessage = trim((string)($job->error_message->value() ?? ''));
                throw new Exception($errorMessage !== '' ? $errorMessage : 'Edge gateway command failed');
            }
+            if ($status === 'TIMED_OUT') {
+                $errorMessage = trim((string)($job->error_message->value() ?? ''));
+                throw new Exception($errorMessage !== '' ? $errorMessage : 'Edge gateway command timed out');
+            }

            if (microtime(true) >= $deadline) {
                break;
@@ -2798,13 +2838,21 @@ BASH;

        $job = (new edge_gateway_command_jobs_o())->select($jobId);
        if ($job->exists()) {
-            $job->delivery_json->set($this->buildDeliveryMetadata(
-                array_merge(
-                    (array)($job->delivery_json->value() ?? []),
-                    ['last_dispatch_error' => 'Edge gateway command timed out']
-                ),
-                self::COMMAND_EXPIRES_AFTER_SECONDS
-            ));
+            $status = (string)$job->status->value();
+            if ($status === 'COMPLETED') {
+                $response = (array)($job->response_json->value() ?? []);
+                return (array)($response['payload'] ?? []);
+            }
+            if ($status === 'FAILED') {
+                $errorMessage = trim((string)($job->error_message->value() ?? ''));
+                throw new Exception($errorMessage !== '' ? $errorMessage : 'Edge gateway command failed');
+            }
+            if ($status === 'TIMED_OUT') {
+                $errorMessage = trim((string)($job->error_message->value() ?? ''));
+                throw new Exception($errorMessage !== '' ? $errorMessage : 'Edge gateway command timed out');
+            }
+
+            $this->finalizeCommandJob($job, false, [], 'Edge gateway command timed out', null, 'TIMED_OUT');
        }

        throw new Exception('Edge gateway command timed out');
@@ -2916,7 +2964,8 @@ BASH;
        bool $ok,
        array $payload = [],
        ?string $errorMessage = null,
-        ?edge_gateways_o $gateway = null
+        ?edge_gateways_o $gateway = null,
+        ?string $terminalStatus = null
    ): void {
        $gatewayObject = $gateway ?? $this->requireGateway((int)$job->gateway_id->value());
        $response = [
@@ -2940,11 +2989,17 @@ BASH;
        ));
        $job->completed_at->set($this->now());
        $job->error_message->set($ok ? null : $errorMessage);
-        $job->status->set($ok ? 'COMPLETED' : 'FAILED');
+        $job->status->set($ok ? 'COMPLETED' : $this->normalizeCommandFailureStatus($terminalStatus));

        $this->applyCommandResult($gatewayObject, $job, $ok, $payload, $errorMessage);
    }

+    private function normalizeCommandFailureStatus(?string $status): string
+    {
+        $normalized = strtoupper(trim((string)$status));
+        return in_array($normalized, ['FAILED', 'TIMED_OUT'], true) ? $normalized : 'FAILED';
+    }
+
    private function applyCommandResult(
        edge_gateways_o $gateway,
        edge_gateway_command_jobs_o $job,
@@ -3623,6 +3678,7 @@ BASH;
            if ($this->buildBrokerInternalUrl() === null) {
                $this->markCommandDeliveryFailure($job, self::DELIVERY_CHANNEL_BROKER, 'Edge broker is not configured', 'broker_not_configured');
                if ($requireFastPath) {
+                    $this->finalizeCommandJob($job, false, [], 'Edge broker is not configured', $gateway);
                    throw new Exception('Edge broker is not configured');
                }
            } else {
@@ -3634,6 +3690,14 @@ BASH;
                } catch (Exception $exception) {
                    $this->markCommandDeliveryFailure($job, self::DELIVERY_CHANNEL_BROKER, $exception->getMessage(), 'broker_dispatch_failed');
                    if ($requireFastPath) {
+                        $this->finalizeCommandJob(
+                            $job,
+                            false,
+                            [],
+                            $exception->getMessage(),
+                            $gateway,
+                            $this->isCommandTimeoutError($exception->getMessage()) ? 'TIMED_OUT' : 'FAILED'
+                        );
                        throw $exception;
                    }
                }
@@ -3641,12 +3705,18 @@ BASH;
        }

        if ($requireFastPath) {
+            $this->finalizeCommandJob($job, false, [], 'Edge broker fast path is unavailable', $gateway);
            throw new Exception('Edge broker fast path is unavailable');
        }

        return $this->waitForCommandResult((int)$job->id);
    }

+    private function isCommandTimeoutError(string $errorMessage): bool
+    {
+        return str_contains(strtolower(trim($errorMessage)), 'timed out');
+    }
+
    private function markCommandJobDispatching(
        edge_gateway_command_jobs_o $job,
        string $channel,
@@ -368,6 +368,14 @@ Purpose: operational lane control and relay management.
 | `POST /modules/self-serve/lane/force/machine/enable` | `lane_id`, optional `duration`, optional `license_plate` | `modules_selfserve_lane_force_machine_enable` | Bypasses service gating and marks the lane as in wash. |
 | `POST /modules/self-serve/lane/force/machine/disable` | `lane_id`, optional `license_plate` | `modules_selfserve_lane_force_machine_disable` | Keeps the lane in wash but turns the machine relay off. |

+Shelly transport behavior:
+
+- The department variable `shelly_transport_mode` controls the default relay path. Missing or `cloud` keeps Shelly cloud as the default. `gateway` routes self-serve relay status, set, manual enable, force, START/STOP side effects, and gate relay operations through the edge gateway/local edge agent.
+- Local edge operation requires an active edge gateway plus an `edge_gateway_relay_bindings` row for each logical Shelly relay id. Each binding must resolve a bound `device_id`, `local_ip`, and `channel`.
+- Binding fallback stays unchanged: `PREFER_LOCAL` uses the local edge agent first and may fall back to cloud, `LOCAL_ONLY` fails instead of falling back, and `CLOUD_ONLY` bypasses local dispatch.
+- Operator diagnostics can add `transport=local` or `transport=gateway` to the `/modules/self-serve/lane/*` relay, gate, command, and allowed-services endpoints to force local-only dispatch. `transport=cloud` forces Shelly cloud. Regular customer flows do not send these overrides.
+- Relay status and set responses keep `relay_id`, `online`, `on`, and `status.switch:0.output` stable. Gateway responses may also include `binding`, `execution`, and `raw` metadata for diagnostics.
+
 STOP flow details:

 - `selfserve_lane_command::STOP` requires lane status `OCCUPIED`.
@@ -1025,7 +1033,7 @@ The STOP flow tries not to let relay or session-completion errors block the lane

 - Runtime schema changes are additive and lazy through `classes/selfserve_schema_bootstrap.php`.
 - The session tables are safe to create idempotently from runtime flows because the project does not use a centralized migration runner.
- Machine relay control is delegated to the Shelly module through `selfserve_lane_relay_controller_t`.
+- Machine relay control is delegated to `selfserve_lane_relay_controller_t`, which resolves Shelly cloud vs. edge gateway transport per department.
 - Manual relay enable is still gated by the lane cache, while force enable and force disable bypass that gate.

 ## Suggested Usage Pattern
@@ -35,7 +35,7 @@ trait selfserve_lane_relay_controller_t
        }

        if (!in_array($normalized, ['cloud', 'gateway', 'local'], true)) {
-            throw new \Exception('Invalid Shelly transport override. Expected local or cloud.');
+            throw new \Exception('Invalid Shelly transport override. Expected cloud, gateway, or local.');
        }

        $this->shelly_transport_override = $normalized;
@@ -44,7 +44,7 @@ trait selfserve_lane_relay_controller_t

    /**
     * Get current MACHINE relay status from Shelly.
-     * @return array{relay_id: string, online: bool, on: bool}
+     * @return array{relay_id: string, online: bool, on: bool, status: array<string,array<string,bool>>, binding?: array, execution?: array, raw?: array}
     * @throws \Exception
     */
    public function getMachineRelayStatus(): array
@@ -54,7 +54,7 @@ trait selfserve_lane_relay_controller_t

    /**
     * Get current MACHINE_PROGRAM_PICKER relay status from Shelly.
-     * @return array{relay_id: string, online: bool, on: bool}
+     * @return array{relay_id: string, online: bool, on: bool, status: array<string,array<string,bool>>, binding?: array, execution?: array, raw?: array}
     * @throws \Exception
     */
    public function getMachineProgramPickerRelayStatus(): array
@@ -64,7 +64,7 @@ trait selfserve_lane_relay_controller_t

    /**
     * Get current MACHINE_CLEANER relay status from Shelly.
-     * @return array{relay_id: string, online: bool, on: bool}
+     * @return array{relay_id: string, online: bool, on: bool, status: array<string,array<string,bool>>, binding?: array, execution?: array, raw?: array}
     * @throws \Exception
     */
    public function getMachineCleanerRelayStatus(): array
@@ -75,7 +75,7 @@ trait selfserve_lane_relay_controller_t
    /**
     * Get current relay status from Shelly for a specific relay type.
     * @param selfserve_lane_relay $relay
-     * @return array{relay_id: string, online: bool, on: bool}
+     * @return array{relay_id: string, online: bool, on: bool, status: array<string,array<string,bool>>, binding?: array, execution?: array, raw?: array}
     * @throws \Exception
     */
    public function getRelayStatus(selfserve_lane_relay $relay): array
@@ -87,11 +87,22 @@ trait selfserve_lane_relay_controller_t
        }

        $device = $snapshot[$relay_id];
-        return [
+        $on = $this->extractRelayOnState($device, $relay);
+        $result = [
            'relay_id' => $relay_id,
            'online' => $this->extractRelayOnlineState($device),
-            'on' => $this->extractRelayOnState($device, $relay),
+            'on' => $on,
+            'status' => ['switch:0' => ['output' => $on]],
        ];
+
+        foreach (['binding', 'execution', 'raw'] as $diagnostic_key) {
+            $diagnostic_value = $this->getPayloadValue($device, $diagnostic_key);
+            if (is_array($diagnostic_value) && $diagnostic_value !== []) {
+                $result[$diagnostic_key] = $diagnostic_value;
+            }
+        }
+
+        return $result;
    }

    /**
@@ -870,8 +881,9 @@ trait selfserve_lane_relay_controller_t
            'channel' => self::SHELLY_DEFAULT_CHANNEL,
            'on' => $on,
        ];
-        // Keep duration parameter for route compatibility, but do not pass toggle_after.
-        // Some Shelly firmware variants interpret toggle_after=0 as immediate toggle.
+        if ($duration !== null && $duration > 0) {
+            $payload['toggle_after'] = $duration;
+        }

        $response = $this->isDemoRelayId($relay_id)
            ? [$this->buildDemoRelaySnapshotEntry($relay_id, $on)]