fix(queue): refuse a concurrency an exclusive consumer cannot survive

[levivannoort]

The break

A NATS connection is one socket with one shared read pump. Broker\Nats's docblock has always said so, and tests/Queue/servers/Nats/worker.php hardcodes job('nats', 1) because of it. Nothing enforced it.

Above one coroutine, consumeBound() parks the receive loop inside a read on that socket while the coroutines processing earlier messages commit(), reject() or extend on the same socket. Swoole does not tolerate the overlap:

Swoole\Error: Socket#5 has already been bound to another coroutine#2,
reading of the same socket in coroutine#3 at the same time is not allowed

Reproduced against nats:2.11 on a Swoole build, N coroutines publishing over one shared connection:

coroutines	result
1	expected=15 acked=15 errors=0
2	fatal on the first message
4	fatal on the first message
8	fatal on the first message

It is not a slow degradation or a silent miscorrelation. The worker dies immediately, and it dies under traffic rather than at boot.

The existing guard in Server::start() only covers multi-queue workers sharing a consumer (count($this->jobs) > 1). A single-queue job at maxCoroutines above one walked straight past it.

The fix

Consumer\Exclusive marks a consumer that owns its transport. Broker\Nats carries it. Server::start() refuses before any loop runs:

Queue 'v1-functions' is registered with job('v1-functions', 8), but its consumer Utopia\Queue\Broker\Nats drives a single socket that only one coroutine may read at a time. Register it as job('v1-functions', 1) and add replicas for throughput.

The marker gates this, not the concurrency. Broker\Redis does not carry it and keeps its caps unchanged — Connection\Locking already serialises the coroutines sharing one connection. A Redis worker at job('v1-functions', 8) behaves exactly as before, and there is a test pinning that.

Tests

Three added to ServerJobsTest:

• testStartRefusesConcurrencyOnAnExclusiveConsumer — asserts the refusal and that $adapter->consumed === [], so the refusal is proven to land before any loop is handed a cap it cannot survive, rather than only pinning the message wording.
• testStartAcceptsOneCoroutineOnAnExclusiveConsumer — NATS still runs at 1.
• testStartKeepsConcurrencyOnAConsumerWithoutTheMarker — the Redis shape stays at 8.

Seen red: replacing the guard condition with if (false) fails only testStartRefusesConcurrencyOnAnExclusiveConsumer (the concurrency must be refused / Failed asserting that null is not null) and leaves the other ten green, which is what shows the guard is capability-gated rather than blanket.

ServerJobsTest 11/11. Unit suite 49 tests, 4 errors, all Swoole\Coroutine\run() missing on the host in SwooleContextTest and present before this change. PHPStan and Pint clean, bin/monorepo validate passes.

Note for callers

Anyone who sets a coroutine cap from configuration and can point a worker at NATS should cap it at 1 for that transport, or they trade a crash loop for this refusal. Appwrite Cloud already forces maxCoroutines to 1 on a nats:// DSN and warns; this makes the library safe for callers that do not.