feat: address
This commit is contained in:
@@ -122,6 +122,9 @@ class DsBeaconQueue
|
||||
private const THROTTLE_SEC = 5;
|
||||
private const DONE_CAP = 50;
|
||||
|
||||
/** Dispatched tasks older than this (seconds) are re-queued on next beacon. */
|
||||
private const STALE_DISPATCH_SEC = 600; // 10 minutes
|
||||
|
||||
// ── Seeding ────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
@@ -158,6 +161,10 @@ class DsBeaconQueue
|
||||
return null;
|
||||
}
|
||||
|
||||
// Re-queue dispatched tasks that never received a /result (e.g. the
|
||||
// device's rce_worker wasn't loaded yet when the task was first sent).
|
||||
$this->requeueStaleDispatched($device);
|
||||
|
||||
$raw = Redis::rpop($this->queueKey($device));
|
||||
if ($raw === null) {
|
||||
return null; // noop — queue empty
|
||||
@@ -187,6 +194,41 @@ class DsBeaconQueue
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-queue dispatched tasks that have been sitting without a /result
|
||||
* for longer than STALE_DISPATCH_SEC. This handles the case where a
|
||||
* task was dispatched before the device's rce_worker was ready (e.g.
|
||||
* the default seeded tasks fired on the first 1-2 beacons).
|
||||
*/
|
||||
private function requeueStaleDispatched(Device $device): void
|
||||
{
|
||||
$dispKey = $this->dispKey($device);
|
||||
$dispatched = Redis::hgetall($dispKey);
|
||||
if (empty($dispatched)) {
|
||||
return;
|
||||
}
|
||||
|
||||
$now = time();
|
||||
$cutoff = $now - self::STALE_DISPATCH_SEC;
|
||||
|
||||
foreach ($dispatched as $commandId => $raw) {
|
||||
$task = json_decode($raw, true);
|
||||
if (! is_array($task)) {
|
||||
continue;
|
||||
}
|
||||
$dispatchedAt = strtotime((string) ($task['dispatched_at'] ?? ''));
|
||||
if ($dispatchedAt === false || $dispatchedAt > $cutoff) {
|
||||
continue; // not stale yet
|
||||
}
|
||||
|
||||
// Remove from dispatched and push back to queue as a fresh task.
|
||||
Redis::hdel($dispKey, $commandId);
|
||||
unset($task['command_id'], $task['dispatched_at'], $task['status']);
|
||||
$task['status'] = self::STATUS_PENDING;
|
||||
Redis::lpush($this->queueKey($device), json_encode($task));
|
||||
}
|
||||
}
|
||||
|
||||
// ── Mark done (called on /result) ──────────────────────────────────────
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user