Queue consumer: All coroutines are asleep - deadlock

<?php
/**
 * RESTRUCTURED: no longer uses a single long-lived blocking
 * brpoplpush() call as the loop's own suspension point. That relied on
 * two things that turned out to be unreliable together: (1) Redis
 * runtime hooking making the call properly coroutine-yielding at all,
 * and (2) Coroutine::cancel()/isCanceled() actually being able to
 * interrupt it once blocked — OpenSwoole's own docs label cancel() as
 * "Experimental feature, not recommended for production use", and in
 * practice a stuck brpoplpush() sometimes didn't even honor its own
 * protocol-level timeout, suggesting a socket-level hang independent
 * of anything cancel() could reach at all.
 *
 * Now: non-blocking rpoplpush() (same atomic move-between-lists
 * semantics as brpoplpush(), just returns immediately with NULL if
 * empty) in a poll loop, with Coroutine::sleep() between empty polls.
 * sleep() is a core Swoole coroutine primitive, not a runtime-hooked
 * extension call — a reliable, well-established suspension point,
 * unlike the Redis call it replaces. The loop checks
 * $shutdownRequested/$restartRequested directly, frequently, between
 * short operations — that's now the PRIMARY shutdown path.
 * Coroutine::cancel()/isCanceled() are kept as a last-resort backstop
 * (still checked at the same points as before), not the mechanism
 * shutdown actually depends on working
 *
 * $pollIntervalMs trades latency (how long a freshly-queued item might
 * sit before being picked up) against idle CPU/Redis round-trips —
 * 2000ms keeps both reasonable
 */
function startQueueConsumer(int $maxConcurrency = 10, int $pollIntervalMs = 2000): void {

    go(function () use ($maxConcurrency, $pollIntervalMs) {

        global $shutdownRequested;
        global $restartRequested;
        global $queueConsumerCID;

        $queueConsumerCID->set(\Swoole\Coroutine::getCid());

        loadQueueStateFromFile();
        recoverStrandedQueueItems();

        $redis = new \Redis();
        $redis->connect('127.0.0.1', 6379);
        $redis->setOption(\Redis::OPT_READ_TIMEOUT, 5);

        $semaphore = new \Swoole\Coroutine\Channel($maxConcurrency);

        try {

            while (TRUE) {

                if ($shutdownRequested->get() === 1 || $restartRequested->get() === 1) {
                    logger('Queue consumer: shutdown/restart requested — stopping', LOG_INFO);
                    break;
                }

                \Swoole\Coroutine::sleep(0.2);

                // Backstop — kept exactly as before, still checked at
                // this same point, but no longer what shutdown
                // actually relies on to happen promptly
                if (\Swoole\Coroutine::isCanceled()) {
                    logger('Queue consumer: cancelled for reload — stopping (backstop check)', LOG_INFO);
                    break;
                }

                try {

                    // CHANGED: was brpoplpush() (blocking) — now
                    // non-blocking, returns immediately either way.
                    $raw = $redis->rpoplpush(QUEUE_KEY, QUEUE_PROCESSING_KEY);

                } catch (Throwable $e) {

                    logger('Queue consumer: RPOPLPUSH failed - ' . $e->getMessage() . ' — reconnecting.', LOG_ERR);

                    // Self-healing: discard whatever state this
                    // connection is in, don't try to reuse it.
                    try { $redis->close(); } catch (Throwable $ignore) {}

                    try {

                        $redis = new \Redis();
                        $redis->connect('127.0.0.1', 6379);
                        $redis->setOption(\Redis::OPT_READ_TIMEOUT, 5);

                    } catch (Throwable $reconnectError) {

                        logger('Queue consumer: reconnect failed - ' . $reconnectError->getMessage() . ' — retrying in 2s.', LOG_ERR);
                        \Swoole\Coroutine::sleep(2);
                    }

                    continue;
                }

                if (empty($raw)) {

                    // Nothing queued — THIS is the loop's primary
                    // suspension point now, not a blocked Redis call.
                    \Swoole\Coroutine::sleep($pollIntervalMs / 1000);
                    continue;
                }

                $payload = json_decode($raw, TRUE);

                if (!is_array($payload) || empty($payload['type'])) {

                    logger('Queue consumer: malformed payload — ' . $raw, LOG_ERR);

                    $redisCleanup = new \Redis();
                    $redisCleanup->connect('127.0.0.1', 6379);
                    $redisCleanup->lRem(QUEUE_PROCESSING_KEY, $raw, 1);
                    $redisCleanup->close();

                    continue;
                }

                $semaphore->push(TRUE);

                go(function () use ($payload, $raw, $semaphore) {

                    try {

                        dispatchQueuedJob($payload);

                    } finally {

                        $redisFinish = new \Redis();
                        $redisFinish->connect('127.0.0.1', 6379);
                        $redisFinish->lRem(QUEUE_PROCESSING_KEY, $raw, 1);
                        $redisFinish->close();

                        $semaphore->pop();
                    }
                });
            }

        } finally {

            $queueConsumerCID->set(0);
            try { $redis->close(); } catch (Throwable $ignore) {}
        }
    });
}

/**
 * Resolves a dequeued payload's type to its callback (Class::method
 * string OR bare function name — same "Class::method" convention
 * already used by the crontab task dispatcher, extended here to also
 * allow a plain function) and invokes it as ($id, $data).
 */
function dispatchQueuedJob(array $payload): void {

	$typeValue = $payload['type'] ?? '';
	$id = $payload['id'] ?? '';
	$data = $payload['data'] ?? [];

	try {
		
        $type = QueuedItemType::from($typeValue);

	} catch (ValueError $e) {

		logger("dispatchQueuedJob: unknown queue item type '{$typeValue}' - cannot dispatch.", LOG_ERR);
		return;
	}

	$callback = $type->value;

	try {

		if (str_contains($callback, '::')) {

			[$class, $method] = explode('::', $callback, 2);

			if (!class_exists($class) || !method_exists($class, $method)) {
				logger("dispatchQueuedJob: missing class/method {$callback} for type {$type->name}.", LOG_ERR);
				return;
			}

			$class::$method($id, $data);

		} else {

			if (!function_exists($callback)) {
				logger("dispatchQueuedJob: Missing function {$callback} for type {$type->name}", LOG_ERR);
				return;
			}

			$callback($id, $data);
		}

	} catch (Throwable $e) {

		logger("dispatchQueuedJob: job {$callback} (type {$type->name}, id {$id}) failed - " . $e->getMessage(), LOG_ERR);
	}
}

===================================================================
 [FATAL ERROR]: all coroutines (count: 1) are asleep - deadlock!
===================================================================

 [Coroutine-6]
--------------------------------------------------------------------
#0 /home/henrik/opcache/elycompile/site/shared.php(24310): Swoole\Coroutine::sleep() <!-- This line varies but it is somewhere inside the while() loop
#1 [internal function]: {closure}()

Runtime::enableCoroutine(SWOOLE_HOOK_ALL);

$ php --ri swoole
henrik@HSLAPTOP-ASUS:~/opcache/elycompile/site$ php --ri swoole

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 5.1.0
Built => Nov  5 2024 14:57:25
coroutine => enabled with boost asm context
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
openssl => OpenSSL 3.0.2 15 Mar 2022
dtls => enabled
http2 => enabled
json => enabled
pcre => enabled
zlib => 1.2.11
brotli => E16777225/D16777225
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_fiber_mock => Off => Off
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => On => On
swoole.unixsock_buffer_size => 8388608 => 8388608
should_send_telemetry returned false, skipping