{

public const HEARTBEAT_INTERVAL = 30;

public const PRUNE_TIMEOUT = 300;

public const PRUNE_CHECK_INTERVAL = 60;

public const TASK_RESULT_RETENTION = 86400;

public const TASK_HISTORY_RETENTION = 86400; // 24 hours (configurable)

public const TASK_TTL = 86400; // 24 hours TTL for all tasks

public const DEFAULT_STREAM_TIMEOUT = 300; // 5 minutes default stream timeout

public const SCAN_BATCH = 500; // keys per SCAN round-trip

/**

public const BLPOP_TIMEOUT = 1; // seconds Redis holds the blocking pop

public const CONNECT_TIMEOUT = 5; // seconds to establish a connection

public const READ_TIMEOUT = 10; // seconds to wait for any single reply

public const TCP_KEEPALIVE = 60; // seconds idle before the kernel probes

private Redis $redis;

private string $serverId;

private array $allowedTasks = [];

private array $taskHandlers = [];

private array $taskOptions = [];

private ?Middleware $middleware = null;

private ?string $webhookUrl = null;

private ?int $requeueThreshold = null;

private int $delaySeconds;

private int $taskHistoryRetention;

private int $taskTtl;

private LoggerInterface $logger;

private bool $running = false;

public function __construct(

?Redis $redisClient = null,

string $host = '127.0.0.1',

int $port = 6379,

int $db = 0,

?string $password = null,

?string $serverId = null,

?string $webhookUrl = null,

?int $requeueThreshold = null,

int $delaySeconds = 30,

?int $taskHistoryRetention = null,

?int $taskTtl = null,

?LoggerInterface $logger = null

) {

if ($redisClient) {

$this->redis = $redisClient;

} else {

$this->redis = new Redis();

// Bounded connect and read. Without the read timeout phpredis waits

// forever for a reply, so a half-open socket strands the caller

// instead of raising — see the constants above.

$this->redis->connect(

$host,

$port,

self::CONNECT_TIMEOUT,

null,

0,

self::READ_TIMEOUT

);

// Let the kernel notice a peer that has gone away while we are idle.

if (defined('Redis::OPT_TCP_KEEPALIVE')) {

$this->redis->setOption(Redis::OPT_TCP_KEEPALIVE, self::TCP_KEEPALIVE);

}

if ($password) {

$this->redis->auth($password);

}

$this->redis->select($db);

}

$this->serverId = $serverId ?? gethostname();

$this->webhookUrl = $webhookUrl;

$this->requeueThreshold = $requeueThreshold;

$this->delaySeconds = $delaySeconds;

$this->taskHistoryRetention = $taskHistoryRetention ?? self::TASK_HISTORY_RETENTION;

$this->taskTtl = $taskTtl ?? self::TASK_TTL;

$this->logger = $logger ?? new NullLogger();

$this->registerServer();

}

/**

public function getRedisClient(): Redis

{

return $this->redis;

}

/**

public function getAllowedTasks(): array

{

return array_keys($this->allowedTasks);

}

/**

public function setMiddleware(Middleware $middleware): self

{

$this->middleware = $middleware;

return $this;

}

/**

public function registerServer(): void

{

$serverData = [

'allowed_tasks' => array_keys($this->allowedTasks),

'status' => 'idle',

'last_heartbeat' => microtime(true),

];

$this->redis->hSet('servers', $this->serverId, json_encode($serverData));

}

/**

public function task(string $taskName, callable $handler, array $options = []): self

{

$this->allowedTasks[$taskName] = true;

$this->taskHandlers[$taskName] = $handler;

$this->taskOptions[$taskName] = array_merge([

'timeout' => null,

'stream' => false,

'retries' => 0,

], $options);

$this->registerServer();

return $this;

}

/**

public function enqueue(string $taskName, array $data = [], ?string $taskId = null, ?array $additionalParams = null): Task

{

$options = $this->taskOptions[$taskName] ?? [];

$payload = [

'data' => [

"args" => [$data],

],

'timeout' => $options['timeout'] ?? null,

'stream' => $options['stream'] ?? false,

'retries' => $options['retries'] ?? 0,

];

$task = new Task(

taskName: $taskName,

payload: $payload,

taskId: $taskId,

additionalParams: $additionalParams

);

if ($options['stream'] ?? false) {

$task->stream = true;

}

$taskDict = $task->toArray();

$now = microtime(true);

$taskDict['created_at'] = $now;

$taskDict['queued_at'] = $now;

$this->enqueueTask($taskDict, $payload);

return $task;

}

/**

private function enqueueTask(array $taskData, array $payload): void

{

$taskData['status'] = 'queued';

$this->checkMiddleware('before_enqueue');

if (!isset($taskData['queued_at'])) {

$taskData['queued_at'] = microtime(true);

}

$taskId = $taskData['task_id'];

$encoded = json_encode($taskData);

$historyScore = $taskData['created_at'] ?? $taskData['queued_at'];

$this->redis->multi(Redis::PIPELINE)

->setex("task:{$taskId}", self::TASK_TTL, $encoded)

->zAdd('task_history', $historyScore, $taskId)

->setex(

"task_history:{$taskId}",

self::TASK_HISTORY_RETENTION,

json_encode($this->withoutPayload($taskData))

)

->rPush('ml_tasks', $encoded)

->zAdd('queued_requests', $taskData['queued_at'], $taskId)

->exec();

$this->checkMiddleware('after_enqueue');

}

/**

public function enqueueDelayedTask(array $taskDict, int $delaySeconds): void

{

$runAt = microtime(true) + $delaySeconds;

$taskJson = json_encode($taskDict);

$this->redis->zAdd('delayed_tasks', $runAt, $taskJson);

$this->logger->info("Delayed task {$taskDict['task_id']} by {$delaySeconds} seconds.");

}

/**

public function startWorkers(int $workers = 1): void

{

$this->running = true;

$this->checkMiddleware('before_worker_boot');

$this->checkMiddleware('after_worker_boot');

$lastHeartbeat = 0.0;

$lastPrune = 0.0;

$lastDelayedCheck = 0.0;

$this->logger->info("ModelQ workers started. Registered tasks: " . implode(', ', array_keys($this->allowedTasks)));

while ($this->running) {

$now = microtime(true);

// Heartbeat

if ($now - $lastHeartbeat >= self::HEARTBEAT_INTERVAL) {

$this->heartbeat();

$lastHeartbeat = $now;

}

// Pruning

if ($now - $lastPrune >= self::PRUNE_CHECK_INTERVAL) {

$this->pruneInactiveServers();

$this->requeueStuckProcessingTasks();

// pruneOldTaskResults() is deliberately NOT called here. Every

// task_result key is written with a TTL, so Redis expires it on

// its own; running the scan every PRUNE_CHECK_INTERVAL only

// re-read the whole keyspace. Measured on a production shard

// over 26 days: 196.7M SCAN + 228.8M GET to issue 2 DELs.

$lastPrune = $now;

}

// Check delayed tasks

if ($now - $lastDelayedCheck >= 1) {

$this->requeueDelayedTasks();

$lastDelayedCheck = $now;

}

// Process tasks

$this->updateServerStatus('idle');

$taskData = $this->redis->blPop(['ml_tasks'], self::BLPOP_TIMEOUT);

if (!$taskData) {

continue;

}

$this->updateServerStatus('busy');

$taskJson = $taskData[1];

$taskDict = json_decode($taskJson, true);

$task = Task::fromArray($taskDict);

$added = $this->redis->sAdd('processing_tasks', $task->taskId);

if ($added === 0) {

$this->logger->warning("Task {$task->taskId} is already being processed. Skipping duplicate.");

continue;

}

$now = microtime(true);

$task->status = 'processing';

$task->startedAt = $now;

$taskDict['status'] = 'processing';

$taskDict['started_at'] = $now;

$this->redis->setex("task:{$task->taskId}", 86400, json_encode($taskDict));

if (isset($this->allowedTasks[$task->taskName])) {

try {

$this->logger->info("Started processing: {$task->taskName}");

$startTime = microtime(true);

$this->processTask($task);

$endTime = microtime(true);

$this->logger->info(sprintf(

"Finished %s in %.2f seconds",

$task->taskName,

$endTime - $startTime

));

} catch (TaskProcessingException $e) {

$this->logger->error("TaskProcessingError: " . $e->getMessage());

$this->handleRetry($task, $taskDict);

} catch (Throwable $e) {

$this->logger->error("Unexpected error: " . $e->getMessage());

$this->handleRetry($task, $taskDict);

}

} else {

$this->logger->warning("Cannot process task {$task->taskName}, re-queueing...");

$this->redis->rPush('ml_tasks', $taskJson);

}

}

$this->checkMiddleware('before_worker_shutdown');

$this->checkMiddleware('after_worker_shutdown');

}

/**

public function stop(): void

{

$this->running = false;

}

/**

private function handleRetry(Task $task, array $taskDict): void

{

$retries = $task->payload['retries'] ?? 0;

if ($retries > 0) {

$newTaskDict = $task->toArray();

$newTaskDict['payload'] = $task->originalPayload;

$newTaskDict['payload']['retries'] = $retries - 1;

$this->enqueueDelayedTask($newTaskDict, $this->delaySeconds);

}

}

/**

private function processTask(Task $task): void

{

try {

if (!isset($this->allowedTasks[$task->taskName])) {

$task->status = 'failed';

$task->result = 'Task not allowed on this server.';

$this->storeFinalTaskState($task, false);

$this->logger->error("Task {$task->taskName} is not allowed on this server.");

throw new TaskProcessingException($task->taskName, 'Task not allowed');

}

$handler = $this->taskHandlers[$task->taskName] ?? null;

if (!$handler) {

$task->status = 'failed';

$task->result = 'Task handler not found';

$this->storeFinalTaskState($task, false);

$this->logger->error("Task {$task->taskName} failed - handler not found.");

throw new TaskProcessingException($task->taskName, 'Task handler not found');

}

$data = $task->payload['data'] ?? [];

$timeout = $task->payload['timeout'] ?? null;

$stream = $task->payload['stream'] ?? false;

$this->logger->info("Processing task: {$task->taskName} with data: " . json_encode($data));

if ($stream) {

// Stream results using Redis Streams

$generator = $handler($data);

if ($generator instanceof \Generator) {

foreach ($generator as $result) {

$task->status = 'in_progress';

$this->redis->xAdd(

"task_stream:{$task->taskId}",

'*',

['result' => json_encode($result)]

);

}

}

$task->status = 'completed';

$this->redis->expire("task_stream:{$task->taskId}", 3600);

$this->storeFinalTaskState($task, true);

} else {

// Standard execution

if ($timeout) {

$result = $this->runWithTimeout($handler, $timeout, $data);

} else {

$result = $handler($data);

}

$task->result = $result;

$task->status = 'completed';

$this->storeFinalTaskState($task, true);

}

$this->logger->info("Task {$task->taskName} completed successfully.");

} catch (RetryTaskException $e) {

$this->logger->warning("Task {$task->taskName} requested retry: " . $e->getMessage());

$newTaskDict = $task->toArray();

$newTaskDict['payload'] = $task->originalPayload;

$this->enqueueDelayedTask($newTaskDict, $this->delaySeconds);

} catch (Throwable $e) {

$task->status = 'failed';

$task->result = $e->getMessage();

$this->storeFinalTaskState($task, false, $e);

$this->logTaskErrorToFile($task, $e);

$this->checkMiddleware('on_error', $task, $e);

$this->postErrorToWebhook($task, $e);

$this->logger->error("Task {$task->taskName} failed with error: " . $e->getMessage());

throw new TaskProcessingException($task->taskName, $e->getMessage(), 0, $e);

} finally {

$this->redis->sRem('processing_tasks', $task->taskId);

}

}

/**

private function runWithTimeout(callable $handler, int $timeout, array $data): mixed

{

$startTime = microtime(true);

// Set alarm if pcntl is available

if (function_exists('pcntl_alarm')) {

pcntl_alarm($timeout);

try {

$result = $handler($data);

pcntl_alarm(0);

return $result;

} catch (\Throwable $e) {

pcntl_alarm(0);

throw $e;

}

}

// Fallback: just run and check time after

$result = $handler($data);

$elapsed = microtime(true) - $startTime;

if ($elapsed > $timeout) {

throw new TaskTimeoutException(

'timeout',

"Task exceeded timeout of {$timeout} seconds"

);

}

return $result;

}

/**

private function storeFinalTaskState(Task $task, bool $success, ?Throwable $error = null): void

{

$taskDict = $task->toArray();

$taskDict['finished_at'] = microtime(true);

// Add error details if failed

if (!$success && $error) {

$taskDict['error'] = [

'message' => $error->getMessage(),

'type' => get_class($error),

'file' => $error->getFile(),

'line' => $error->getLine(),

'trace' => $error->getTraceAsString(),

];

}

$this->redis->setex("task_result:{$task->taskId}", 3600, json_encode($taskDict));

$this->redis->setex("task:{$task->taskId}", 86400, json_encode($taskDict));

$this->redis->zRem('queued_requests', $task->taskId);

// Update task history

$this->updateTaskHistory($task->taskId, $taskDict);

}

/**

private function heartbeat(): void

{

$rawData = $this->redis->hGet('servers', $this->serverId);

if (!$rawData) {

$this->registerServer();

return;

}

$data = json_decode($rawData, true);

$data['last_heartbeat'] = microtime(true);

$this->redis->hSet('servers', $this->serverId, json_encode($data));

}

/**

private function updateServerStatus(string $status): void

{

$rawData = $this->redis->hGet('servers', $this->serverId);

if (!$rawData) {

$this->registerServer();

return;

}

$serverData = json_decode($rawData, true);

$serverData['status'] = $status;

$serverData['last_heartbeat'] = microtime(true);

$this->redis->hSet('servers', $this->serverId, json_encode($serverData));

}

/**

private function checkMiddleware(string $event, ?Task $task = null, ?Throwable $error = null): void

{

if ($this->middleware) {

$this->middleware->execute($event, $task, $error);

}

}

/**

public function getRegisteredServerIds(): array

{

$keys = $this->redis->hKeys('servers');

return $keys ?: [];

}

/**

public function getWorkers(): array

{

$workers = [];

$allServers = $this->redis->hGetAll('servers');

foreach ($allServers as $serverId => $dataJson) {

try {

$data = json_decode($dataJson, true);

$workers[$serverId] = [

'worker_id' => $serverId,

'status' => $data['status'] ?? 'unknown',

'allowed_tasks' => $data['allowed_tasks'] ?? [],

'last_heartbeat' => array_key_exists('last_heartbeat', $data) ? $data['last_heartbeat'] : null,

'system_info' => array_key_exists('system_info', $data) ? $data['system_info'] : null,

'hostname' => array_key_exists('hostname', $data) ? $data['hostname'] : null,

'os' => array_key_exists('os', $data) ? $data['os'] : null,

'python_version' => array_key_exists('python_version', $data) ? $data['python_version'] : null,

'php_version' => array_key_exists('php_version', $data) ? $data['php_version'] : null,

];

} catch (Throwable $e) {

$this->logger->warning("Could not parse worker data for {$serverId}: " . $e->getMessage());

}

}

return $workers;

}

/**

public function getWorker(string $workerId): ?array

{

$dataJson = $this->redis->hGet('servers', $workerId);

if (!$dataJson) {

return null;

}

try {

$data = json_decode($dataJson, true);

return [

'worker_id' => $workerId,

'status' => $data['status'] ?? 'unknown',

'allowed_tasks' => $data['allowed_tasks'] ?? [],

'last_heartbeat' => array_key_exists('last_heartbeat', $data) ? $data['last_heartbeat'] : null,

'system_info' => array_key_exists('system_info', $data) ? $data['system_info'] : null,

'hostname' => array_key_exists('hostname', $data) ? $data['hostname'] : null,

'os' => array_key_exists('os', $data) ? $data['os'] : null,

'python_version' => array_key_exists('python_version', $data) ? $data['python_version'] : null,

'php_version' => array_key_exists('php_version', $data) ? $data['php_version'] : null,

];

} catch (Throwable $e) {

$this->logger->warning("Could not parse worker data for {$workerId}: " . $e->getMessage());

return null;

}

}

/**

public function getAllQueuedTasks(): array

{

$queuedTasks = [];

$tasksInList = $this->redis->lRange('ml_tasks', 0, -1);

foreach ($tasksInList as $taskJson) {

try {

$taskDict = json_decode($taskJson, true);

if (($taskDict['status'] ?? '') === 'queued') {

$queuedTasks[] = $taskDict;

}

} catch (Throwable $e) {

$this->logger->error("Error deserializing task from ml_tasks: " . $e->getMessage());

}

}

return $queuedTasks;

}

/**

public function getQueuedTaskCount(): int

{

return (int) $this->redis->lLen('ml_tasks');

}

/**

public function getTaskStatus(string $taskId): ?string

{

$taskData = $this->redis->get("task:{$taskId}");

if ($taskData) {

$data = json_decode($taskData, true);

return $data['status'] ?? null;

}

return null;

}

// ========================================

// Task History Methods

// ========================================

/**

private function withoutPayload(array $taskData): array

{

unset($taskData['payload']);

return $taskData;

}

/**

private function updateTaskHistory(string $taskId, array $taskData): void

{

$this->redis->setex(

"task_history:{$taskId}",

self::TASK_HISTORY_RETENTION,

json_encode($this->withoutPayload($taskData))

);

}

/**

public function getTaskDetails(string $taskId): ?array

{

// First try task_history (longer retention)

$taskData = $this->redis->get("task:{$taskId}");

if ($taskData) {

return json_decode($taskData, true);

}

return null;

}

/**

public function getTaskHistory(

int $limit = 50,

int $offset = 0,

?string $status = null,

?string $taskName = null

): array {

// Get task IDs from sorted set (newest first)

$taskIds = $this->redis->zRevRange('task_history', $offset, $offset + $limit - 1);

$tasks = [];

foreach ($taskIds as $taskId) {

$taskData = $this->redis->get("task_history:{$taskId}");

if ($taskData) {

$task = json_decode($taskData, true);

// Apply filters

if ($status !== null && ($task['status'] ?? '') !== $status) {

continue;

}

if ($taskName !== null && ($task['task_name'] ?? '') !== $taskName) {

continue;

}

$tasks[] = $task;

}

}

return $tasks;

}

/**

public function getFailedTasks(int $limit = 50): array

{

return $this->getTaskHistory(limit: $limit, status: 'failed');

}

/**

public function getCompletedTasks(int $limit = 50): array

{

return $this->getTaskHistory(limit: $limit, status: 'completed');

}

/**

public function getTasksByName(string $taskName, int $limit = 50): array

{

return $this->getTaskHistory(limit: $limit, taskName: $taskName);

}

/**

public function getTaskStats(): array

{

$taskIds = $this->redis->zRange('task_history', 0, -1);

$stats = [

'total' => count($taskIds),

'by_status' => [

'queued' => 0,

'processing' => 0,

'completed' => 0,

'failed' => 0,

],

'by_task_name' => [],

'failed_tasks' => [],

];

foreach ($taskIds as $taskId) {

$taskData = $this->redis->get("task_history:{$taskId}");

if ($taskData) {

$task = json_decode($taskData, true);

$status = $task['status'] ?? 'unknown';

$taskName = $task['task_name'] ?? 'unknown';

// Count by status

if (isset($stats['by_status'][$status])) {

$stats['by_status'][$status]++;

}

// Count by task name

if (!isset($stats['by_task_name'][$taskName])) {

$stats['by_task_name'][$taskName] = ['total' => 0, 'completed' => 0, 'failed' => 0];

}

$stats['by_task_name'][$taskName]['total']++;

if ($status === 'completed') {

$stats['by_task_name'][$taskName]['completed']++;

} elseif ($status === 'failed') {

$stats['by_task_name'][$taskName]['failed']++;

// Add to failed tasks list (limit to 10 most recent)

if (count($stats['failed_tasks']) < 10) {

$stats['failed_tasks'][] = [

'task_id' => $task['task_id'],

'task_name' => $taskName,

'error' => $task['error']['message'] ?? $task['result'] ?? 'Unknown error',

'finished_at' => $task['finished_at'] ?? null,

];

}

}

}

}

return $stats;

}

/**

public function clearTaskHistory(int $olderThanSeconds = 604800): int

{

$cutoff = microtime(true) - $olderThanSeconds;

// Get old task IDs

$oldTaskIds = $this->redis->zRangeByScore('task_history', '-inf', (string) $cutoff);

$removed = 0;

foreach ($oldTaskIds as $taskId) {

$this->redis->zRem('task_history', $taskId);

$this->redis->del("task_history:{$taskId}");

$removed++;

}

return $removed;

}

/**

public function getTaskHistoryCount(): int

{

return $this->redis->zCard('task_history') ?: 0;

}

/**

public function deleteQueue(): void

{

$this->redis->lTrim('ml_tasks', 1, 0);

}

/**

public function removeTaskFromQueue(string $taskId): bool

{

// Always drop the queue index and processing marker so a cancelled task

// is neither counted (queue_num / queue_time) nor picked up again — even

// if it has already left ml_tasks.

$this->redis->zRem('queued_requests', $taskId);

$this->redis->sRem('processing_tasks', $taskId);

$tasks = $this->redis->lRange('ml_tasks', 0, -1);

$removed = false;

foreach ($tasks as $taskJson) {

try {

$taskDict = json_decode($taskJson, true);

if (($taskDict['task_id'] ?? '') === $taskId) {

$this->redis->lRem('ml_tasks', $taskJson, 1);

$removed = true;

$this->logger->info("Removed task {$taskId} from queue.");

break;

}

} catch (Throwable $e) {

$this->logger->error("Failed to process task while trying to remove: " . $e->getMessage());

}

}

return $removed;

}

/**

private function pruneInactiveServers(?int $timeoutSeconds = null): void

{

$timeoutSeconds = $timeoutSeconds ?? self::PRUNE_TIMEOUT;

$allServers = $this->redis->hGetAll('servers');

$now = microtime(true);

$removedCount = 0;

foreach ($allServers as $serverId => $dataJson) {

try {

$data = json_decode($dataJson, true);

$lastHeartbeat = $data['last_heartbeat'] ?? 0;

if (($now - $lastHeartbeat) > $timeoutSeconds) {

$this->redis->hDel('servers', $serverId);

$removedCount++;

$this->logger->info("[Prune] Removed stale server: {$serverId}");

}

} catch (Throwable $e) {

$this->logger->warning("[Prune] Could not parse server data for {$serverId}: " . $e->getMessage());

}

}