fix(coroutine): honour an exception claimed from extended_dispose

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **One task that threw took down the `ThreadPool` worker thread that ran it, and the pool went on reporting that worker as alive.** In coroutine mode a task's exception is claimed by `pool_task_dispose()`, which rejects the task's Future and calls `ZEND_COROUTINE_SET_EXCEPTION_HANDLED`. That claim was discarded: the only place converting it into the durable `EXC_CAUGHT` flag ran in the callback-notify window, and `extended_dispose` — where the pool makes the claim — runs after that window closes. `coroutine_object_destroy()` then found no claim and rethrew the exception into the worker's `EG(exception)`, where the scheduler's exception check turned one failed task into a graceful shutdown of the whole thread. The printed `Fatal error` was the shutdown report, not the cause. The claim is now converted after `extended_dispose` too. Measured with `workers: 2`: two distinct workers before, one after the first throw and none after the second, `getWorkerCount()` reporting 2 throughout and the process hanging on the next submit; after the fix both workers serve every following task. Sync-mode pools were never affected, because there the claim is made inside the window.
 - **`ThreadChannel::send()` and `ThreadChannel::recv()` accepted a cancellation token and ignored it.** Both methods parsed the `?Async\Completable` argument and neither passed it on: `recv()` handed `NULL` to a receive that has supported cancellation all along, and `send()` had no parameter to take one. A parked call could then be broken only by cancelling the coroutine, so every bounded wait had to be hand-built from a cancel race — while `Async\Channel` honoured the same argument on the same signature. Measured before the fix: `recv(Async\timeout(300))` was still parked at 1000 ms. The token now ends either wait with `OperationCanceledException`, the exception `Channel::recv()` raises, carrying the token's own error as its previous, so one `catch (AsyncCancellation)` covers both classes. A token that fired before the call ends it before it waits. A wake is attributed to the token by asking the token: one freed slot wakes every parked sender and one sent value wakes every parked receiver, so the losers of that race park again instead of reporting a cancellation nobody requested — which would leave the method throwing an exception that was never raised (`ZEND_ASSERT(EG(exception))`, SIGABRT on a debug build, a silent `NULL` on a release one).
 - **Sending a value that cannot cross a thread boundary aborted the process instead of only throwing.** `ThreadChannel::send()` transfers its argument into persistent memory before it takes the lock, and the transfer refuses what it cannot copy — a resource, an object with dynamic properties — by releasing the partial graph, leaving the destination `IS_UNDEF` and throwing. The send did not check for that: it pushed the undefined slot into the buffer and reported success, so the caller got the right exception while the buffer held a value no receiver can interpret. `ThreadPool` reads the task as an array, so a debug build died on the assertion at `thread_pool.c:347` and a release build, where that assertion is compiled out, reads array fields from a value that is not one. The send now leaves the buffer untouched and returns false, which every caller already handles: `ThreadChannel::send()` rethrows, and `ThreadPool::submit()` and `map()` release the snapshot and the future first. Reproduced with `$pool->submit(fn () => 1, fopen('php://memory', 'r'))`: SIGABRT before, a caught `Error` and exit 0 after.
 

coroutine.c

@@ -690,6 +690,13 @@ void async_coroutine_finalize(async_coroutine_t *coroutine)
 			dispose(&coroutine->coroutine);
 		}
 
+		/* extended_dispose runs after the notify window, and a thread-pool task
+		 * claims its exception from there: unconverted, coroutine_object_destroy()
+		 * rethrows it into the worker and the scheduler shuts that thread down. */
+		if (exception != NULL && ZEND_COROUTINE_IS_EXCEPTION_HANDLED(&coroutine->coroutine)) {
+			ZEND_ASYNC_EVENT_SET_EXC_CAUGHT(&coroutine->coroutine.event);
+		}
+
 		zend_exception_restore_fast(exception_ptr, prev_exception_ptr);
 
 		// If the exception was handled by any handler, we do not propagate it further.

tests/thread_pool/081-coroutine_task_exception_worker_survives.phpt

@@ -0,0 +1,33 @@
+--TEST--
+ThreadPool: a coroutine-mode task that throws leaves its worker serving
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+// Under test: the OS thread that ran the failing task keeps taking work. One
+// worker makes it exact — if that thread exits, the next Future never settles.
+$pool = new Async\ThreadPool(workers: 1, coroutine: true);
+
+try {
+    Async\await($pool->submit(static fn(): never => throw new RuntimeException('boom')));
+} catch (RuntimeException $e) {
+    echo "rejected: ", $e->getMessage(), "\n";
+}
+
+try {
+    echo "next: ", Async\await($pool->submit(static fn() => 'served'), Async\timeout(2000)), "\n";
+} catch (Throwable $e) {
+    echo "next: ", $e::class, "\n";
+}
+
+$pool->close();
+echo "Done\n";
+?>
+--EXPECT--
+rejected: boom
+next: served
+Done