From 6110caa851b51d44d76025bfd7fcd6d2b6ff03fe Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Tue, 4 Aug 2026 15:45:56 +0400 Subject: [PATCH 1/9] feat(pipeline): make #[FallbackInterceptor] repeatable An Interceptable attribute may now wire several interceptors, each with its own pipeline position: Cache::resolveAliases() collects every FallbackInterceptor attribute (walking the parent chain as before), and InterceptorProvider instantiates each resolved class with the attribute. Needed by testo/fiber, where #[RunInFiber] wraps the test pipeline in a fiber outside the scoped-state guards and opens the coroutine scope inside them. Assisted-By: Claude Fable 5 --- .../Attribute/FallbackInterceptor.php | 5 +- core/Pipeline/InterceptorProvider.php | 9 +- core/Pipeline/Internal/Cache.php | 15 ++-- tests/Core/Pipeline/CacheTest.php | 85 ++++++++++++------- 4 files changed, 72 insertions(+), 42 deletions(-) diff --git a/core/Pipeline/Attribute/FallbackInterceptor.php b/core/Pipeline/Attribute/FallbackInterceptor.php index c9e6563b..1b7a69b5 100644 --- a/core/Pipeline/Attribute/FallbackInterceptor.php +++ b/core/Pipeline/Attribute/FallbackInterceptor.php @@ -15,11 +15,14 @@ * final class RetryPolicy {} * ``` * + * Repeatable: an attribute may wire several interceptors, each with its own pipeline position + * ({@see InterceptorOptions}); every one of them is instantiated with the attribute instance. + * * Makes sense only for interceptors that are executed during tests execution. * * @api */ -#[\Attribute(\Attribute::TARGET_CLASS)] +#[\Attribute(\Attribute::TARGET_CLASS | \Attribute::IS_REPEATABLE)] final class FallbackInterceptor { public function __construct( diff --git a/core/Pipeline/InterceptorProvider.php b/core/Pipeline/InterceptorProvider.php index 8b2d5b4b..8018b503 100644 --- a/core/Pipeline/InterceptorProvider.php +++ b/core/Pipeline/InterceptorProvider.php @@ -90,12 +90,15 @@ public function fromAttributes(string $class, Interceptable ...$attributes): arr $result = []; foreach ($attributes as $attribute) { - # Get alias interceptor - $iClass = Cache::resolveAlias($attribute::class) ?? throw new \RuntimeException( + # Get alias interceptors + $iClasses = Cache::resolveAliases($attribute::class); + $iClasses === [] and throw new \RuntimeException( \sprintf('No interceptor found for attribute %s.', $attribute::class), ); - \is_a($iClass, $class, true) and $result[] = $this->createInstance($iClass, [$attribute]); + foreach ($iClasses as $iClass) { + \is_a($iClass, $class, true) and $result[] = $this->createInstance($iClass, [$attribute]); + } } return $result; diff --git a/core/Pipeline/Internal/Cache.php b/core/Pipeline/Internal/Cache.php index 5b374506..526fe407 100644 --- a/core/Pipeline/Internal/Cache.php +++ b/core/Pipeline/Internal/Cache.php @@ -18,17 +18,17 @@ final class Cache { /** - * @var array, null|class-string> + * @var array, list>> */ private static array $map = []; /** - * Resolve alias interceptor for the given attribute class. + * Resolve alias interceptors for the given attribute class. * * @param class-string $class The attribute class. - * @return class-string|null The interceptor class or null if not found. + * @return list> The interceptor classes; empty if none found. */ - public static function resolveAlias(string $class): ?string + public static function resolveAliases(string $class): array { $c = $class; do { @@ -40,11 +40,14 @@ public static function resolveAlias(string $class): ?string } while ($c); /** - * Resolve fallback handler from the {@see FallbackInterceptor} attribute + * Resolve fallback handlers from the repeatable {@see FallbackInterceptor} attribute * @var list<\ReflectionAttribute> $attrs */ $attrs = Reflection::fetchClassAttributes($class, attributeClass: FallbackInterceptor::class); - return self::$map[$class] ??= $attrs === [] ? null : $attrs[0]->newInstance()->class; + return self::$map[$class] ??= \array_map( + static fn(\ReflectionAttribute $attr): string => $attr->newInstance()->class, + $attrs, + ); } } diff --git a/tests/Core/Pipeline/CacheTest.php b/tests/Core/Pipeline/CacheTest.php index 0b7fcfc3..4b92e4fb 100644 --- a/tests/Core/Pipeline/CacheTest.php +++ b/tests/Core/Pipeline/CacheTest.php @@ -17,32 +17,42 @@ final class CacheTest { /** - * When resolveAlias is called with a class that has a FallbackInterceptor attribute, - * it should cache and return the interceptor class. + * When resolveAliases is called with a class that has a FallbackInterceptor attribute, + * it should cache and return the interceptor classes. */ - public function resolveAliasWithFallbackInterceptorAttribute(): void + public function resolveAliasesWithFallbackInterceptorAttribute(): void { - $result = Cache::resolveAlias(AttributeWithFallback::class); + $result = Cache::resolveAliases(AttributeWithFallback::class); - Assert::same(MockInterceptor::class, $result); + Assert::same([MockInterceptor::class], $result); } /** - * When resolveAlias is called with a class that has no FallbackInterceptor attribute, - * it should return null. + * A repeated FallbackInterceptor attribute wires every listed interceptor, in declaration order. */ - public function resolveAliasWithoutFallbackInterceptorAttribute(): void + public function resolveAliasesCollectsRepeatedFallbacks(): void { - $result = Cache::resolveAlias(AttributeWithoutFallback::class); + $result = Cache::resolveAliases(AttributeWithSeveralFallbacks::class); - Assert::null($result); + Assert::same([MockInterceptor::class, SecondMockInterceptor::class], $result); } /** - * The first resolveAlias call memoises the resolved value in the private static map, + * When resolveAliases is called with a class that has no FallbackInterceptor attribute, + * it should return an empty list. + */ + public function resolveAliasesWithoutFallbackInterceptorAttribute(): void + { + $result = Cache::resolveAliases(AttributeWithoutFallback::class); + + Assert::same($result, []); + } + + /** + * The first resolveAliases call memoises the resolved value in the private static map, * so the key must be present (with the resolved value) afterwards. */ - public function resolveAliasMemoisesResultInMap(): void + public function resolveAliasesMemoisesResultInMap(): void { $map = self::mapProperty(); $orig = $map->getValue(); @@ -50,12 +60,12 @@ public function resolveAliasMemoisesResultInMap(): void try { $map->setValue(null, []); - $result = Cache::resolveAlias(AttributeWithFallbackForCache::class); - Assert::same(MockInterceptor::class, $result); + $result = Cache::resolveAliases(AttributeWithFallbackForCache::class); + Assert::same([MockInterceptor::class], $result); $stored = $map->getValue(); Assert::true(\array_key_exists(AttributeWithFallbackForCache::class, $stored)); - Assert::same(MockInterceptor::class, $stored[AttributeWithFallbackForCache::class]); + Assert::same([MockInterceptor::class], $stored[AttributeWithFallbackForCache::class]); } finally { $map->setValue(null, $orig); } @@ -65,7 +75,7 @@ public function resolveAliasMemoisesResultInMap(): void * Once a parent class is memoised in the map, resolving a child class walks up the * cached map (the do/while loop) and returns the parent's stored value. */ - public function resolveAliasWalksCachedParentInMap(): void + public function resolveAliasesWalksCachedParentInMap(): void { $map = self::mapProperty(); $orig = $map->getValue(); @@ -73,25 +83,25 @@ public function resolveAliasWalksCachedParentInMap(): void try { $map->setValue(null, []); - $parent = Cache::resolveAlias(ParentAttributeForMapWalk::class); - Assert::same(MockInterceptor::class, $parent); + $parent = Cache::resolveAliases(ParentAttributeForMapWalk::class); + Assert::same([MockInterceptor::class], $parent); $stored = $map->getValue(); Assert::true(\array_key_exists(ParentAttributeForMapWalk::class, $stored)); Assert::false(\array_key_exists(ChildAttributeForMapWalk::class, $stored)); - $child = Cache::resolveAlias(ChildAttributeForMapWalk::class); - Assert::same(MockInterceptor::class, $child); + $child = Cache::resolveAliases(ChildAttributeForMapWalk::class); + Assert::same([MockInterceptor::class], $child); } finally { $map->setValue(null, $orig); } } /** - * A class without a FallbackInterceptor caches null via `??=`; the lookup uses - * array_key_exists, so the stored null is a cache hit on subsequent calls. + * A class without a FallbackInterceptor caches an empty list; the lookup uses + * array_key_exists, so the stored empty list is a cache hit on subsequent calls. */ - public function resolveAliasCachesNullAsHit(): void + public function resolveAliasesCachesEmptyAsHit(): void { $map = self::mapProperty(); $orig = $map->getValue(); @@ -99,29 +109,29 @@ public function resolveAliasCachesNullAsHit(): void try { $map->setValue(null, []); - $first = Cache::resolveAlias(NoFallbackForNullCache::class); - Assert::null($first); + $first = Cache::resolveAliases(NoFallbackForNullCache::class); + Assert::same($first, []); $stored = $map->getValue(); Assert::true(\array_key_exists(NoFallbackForNullCache::class, $stored)); - Assert::null($stored[NoFallbackForNullCache::class]); + Assert::same($stored[NoFallbackForNullCache::class], []); - $second = Cache::resolveAlias(NoFallbackForNullCache::class); - Assert::null($second); + $second = Cache::resolveAliases(NoFallbackForNullCache::class); + Assert::same($second, []); } finally { $map->setValue(null, $orig); } } /** - * When resolveAlias is called with a class that inherits from a class with FallbackInterceptor, + * When resolveAliases is called with a class that inherits from a class with FallbackInterceptor, * it should walk up the parent class chain (reflection fallback) and find the interceptor. */ - public function resolveAliasWalksParentClassHierarchy(): void + public function resolveAliasesWalksParentClassHierarchy(): void { - $result = Cache::resolveAlias(ChildAttributeOfFallback::class); + $result = Cache::resolveAliases(ChildAttributeOfFallback::class); - Assert::same(MockInterceptor::class, $result); + Assert::same([MockInterceptor::class], $result); } private static function mapProperty(): \ReflectionProperty @@ -139,6 +149,13 @@ class AttributeWithFallback implements Interceptable { } +#[\Attribute(\Attribute::TARGET_CLASS)] +#[FallbackInterceptor(MockInterceptor::class)] +#[FallbackInterceptor(SecondMockInterceptor::class)] +final class AttributeWithSeveralFallbacks implements Interceptable +{ +} + #[\Attribute(\Attribute::TARGET_CLASS)] final class AttributeWithoutFallback implements Interceptable { @@ -174,3 +191,7 @@ final class ChildAttributeOfFallback extends AttributeWithFallback final class MockInterceptor implements Interceptor { } + +final class SecondMockInterceptor implements Interceptor +{ +} From f0128cfac36fe640c5c56cd4b0493bc6887326a1 Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Tue, 4 Aug 2026 15:46:44 +0400 Subject: [PATCH 2/9] =?UTF-8?q?feat(fiber):=20coroutine=20scope=20?= =?UTF-8?q?=E2=80=94=20Coroutine::spawn()/await()/concurrently()?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every #[RunInFiber] test now runs inside its own coroutine scope: the test body is task #0 of a per-test scheduler, and Coroutine::spawn() adds coroutines to the same round-robin schedule. Between rounds the scope relays control upward with a suspend of its own, so coroutines keep interleaving with the case's other tests under a class-level #[RunInFiber]. - Scheduler rewritten from a static one-shot into a dynamic instance: tasks may be spawned mid-drive, await parks a task until its target settles, and Scheduler::current() exposes the ambient scope to the Coroutine helpers. - The scope is structured: pending coroutines are driven after the body returns; a failed body cancels them (CancelledException is thrown into each pending fiber); an await cycle is broken with a DeadlockException raised at the first parked await(). - Coroutine failures always surface wrapped in CompositeException — even a single one — via await()/concurrently() or at scope close for unawaited ones (the test is marked Error). The body's own throw stays unwrapped. - The scope lives in the new CoroutineScopeInterceptor at ORDER_CLOSE_TO_TEST — inside the scoped-state guards, so coroutines are resumed with their test's assertion/messenger state swapped in — while RunInFiberInterceptor keeps the fiber wrap outside the guards; both are wired by the same attribute via the now-repeatable #[FallbackInterceptor]. Assisted-By: Claude Fable 5 --- bridge/rector/FEATURE_PARITY.md | 2 +- plugin/fiber/README.md | 17 +- plugin/fiber/src/Coroutine.php | 137 ++++++++ .../src/Exception/CancelledException.php | 17 + .../src/Exception/CompositeException.php | 17 +- .../fiber/src/Exception/DeadlockException.php | 18 + .../Internal/CoroutineScopeInterceptor.php | 97 ++++++ .../src/Internal/FiberTestBatchRunner.php | 25 +- .../src/Internal/RunInFiberInterceptor.php | 19 +- plugin/fiber/src/Internal/Scheduler.php | 308 +++++++++++++++--- plugin/fiber/src/Internal/Task.php | 46 +++ plugin/fiber/src/RunInFiber.php | 10 +- plugin/fiber/tests/Feature/StatusTest.php | 49 +++ .../tests/Self/CoroutineInterleaveTest.php | 72 ++++ .../fiber/tests/Self/CoroutineScopeTest.php | 62 ++++ plugin/fiber/tests/Stub/FiberScenarios.php | 48 +++ .../tests/Unit/CoroutineCoverageTest.php | 159 +++++++++ plugin/fiber/tests/Unit/CoroutineTest.php | 203 ++++++++++++ .../tests/Unit/RunInFiberAttributesTest.php | 13 + plugin/fiber/tests/Unit/SchedulerTest.php | 197 +++++++++-- skills/testo-fiber/SKILL.md | 43 ++- 21 files changed, 1448 insertions(+), 111 deletions(-) create mode 100644 plugin/fiber/src/Coroutine.php create mode 100644 plugin/fiber/src/Exception/CancelledException.php create mode 100644 plugin/fiber/src/Exception/DeadlockException.php create mode 100644 plugin/fiber/src/Internal/CoroutineScopeInterceptor.php create mode 100644 plugin/fiber/src/Internal/Task.php create mode 100644 plugin/fiber/tests/Self/CoroutineInterleaveTest.php create mode 100644 plugin/fiber/tests/Self/CoroutineScopeTest.php create mode 100644 plugin/fiber/tests/Unit/CoroutineCoverageTest.php create mode 100644 plugin/fiber/tests/Unit/CoroutineTest.php diff --git a/bridge/rector/FEATURE_PARITY.md b/bridge/rector/FEATURE_PARITY.md index 126a4eff..688224a3 100644 --- a/bridge/rector/FEATURE_PARITY.md +++ b/bridge/rector/FEATURE_PARITY.md @@ -31,7 +31,7 @@ Conversion coverage across the three directions supported by `testo/bridge-recto | **Mocks** (`createMock`/`getMockBuilder`/`prophesize`) | ➖ | ⛔ *Testo has no built-in mocking* | ➖ | | **Memory-leak expectations** | ⛔ *no PHPUnit equivalent* | ➖ | ➖ | | **Retry / Repeat** (`#[Retry]`/`#[Repeat]`) | ⛔ *no PHPUnit equivalent* | ➖ | ➖ | -| **Fiber** (`#[RunInFiber]`) | ⛔ *no PHPUnit/Pest equivalent — neither has a fiber/coroutine test attribute* | ➖ | ➖ | +| **Fiber** (`#[RunInFiber]`, `Coroutine::spawn/await/concurrently`) | ⛔ *no PHPUnit/Pest equivalent — neither has a fiber/coroutine test attribute or an in-test coroutine scope* | ➖ | ➖ | | **`uses()`** (Pest) | ➖ | ➖ | ⛔ *a converted function has no base class, traits or `$this` to attach to; closures that capture `$this`-shared state are left untouched* | | **`arch()` tests** (Pest) | ➖ | ➖ | ⛔ *Testo has no arch-assertion subsystem* | diff --git a/plugin/fiber/README.md b/plugin/fiber/README.md index 31b61fe3..bcf89923 100644 --- a/plugin/fiber/README.md +++ b/plugin/fiber/README.md @@ -28,10 +28,25 @@ Runs tests as plain PHP fibers driven by Testo's own cooperative scheduler, so a test (or the code it exercises) may suspend with `\Fiber::suspend()` and be resumed, and a case's tests may be interleaved to shake out order-dependent races. - `#[RunInFiber]` — run a test (method) or a whole case (class) inside fibers, scheduled by `Schedule::Solo` (default), `RoundRobin` or `Random`. +- `Coroutine::spawn()` / `->await()` / `Coroutine::concurrently()` — add coroutines to the running test's schedule and wait for them; they interleave with the test body (and, under a class-level `#[RunInFiber]`, with the case's other tests) at every suspension point. + +```php +#[RunInFiber] +public function pingPong(): void +{ + $server = Coroutine::spawn(fn(): string => $this->acceptAndEcho()); + $client = Coroutine::spawn(fn(): string => $this->connectAndSend('ping')); + + Assert::same($client->await(), 'pong'); + Assert::same($server->await(), 'ping'); +} +``` + +The scope is structured: the test is not finished until every coroutine it spawned is. Coroutine failures always surface wrapped in a `CompositeException` — even a single one; if the test body fails, pending coroutines are cancelled with a `CancelledException` thrown into them, and an await cycle is broken with a `DeadlockException` at the guilty `await()`. Switching is cooperative and happens only at suspension points — there is no event loop and no preemption. This is for fiber-based/cooperative code and race hunting, **not** for real async I/O: awaiting a timer, socket or `Future` needs the Revolt event loop — use the `testo/bridge-revolt` `#[RunInRevolt]` attribute for that. -The attribute lives under the `Testo\Fiber\` namespace. +Everything lives under the `Testo\Fiber\` namespace. ## Install diff --git a/plugin/fiber/src/Coroutine.php b/plugin/fiber/src/Coroutine.php new file mode 100644 index 00000000..da464448 --- /dev/null +++ b/plugin/fiber/src/Coroutine.php @@ -0,0 +1,137 @@ + $this->acceptAndEcho()); + * $client = Coroutine::spawn(fn(): string => $this->connectAndSend('ping')); + * + * Assert::same($client->await(), 'pong'); + * Assert::same($server->await(), 'ping'); + * } + * ``` + * + * The scope is structured: the test is not finished until every coroutine it spawned is. A coroutine + * still pending when the test body returns keeps being driven; if the body fails, pending coroutines + * are cancelled ({@see CancelledException} is thrown into them). Coroutine failures are always + * surfaced wrapped in a {@see CompositeException} — even a single one — whether rethrown by + * {@see await()} / {@see concurrently()} or reported by the scope for a coroutine nobody awaited. + * + * @api + */ +final class Coroutine +{ + private function __construct( + private readonly Task $task, + ) {} + + /** + * Schedule a closure (or an unstarted fiber) as a coroutine of the running test's scope. + * + * The coroutine gets its first step in the current scheduling round; from there it runs + * cooperatively — it holds the floor until it suspends, finishes, or awaits. + * + * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]`. + */ + public static function spawn(\Closure|\Fiber $body): self + { + $scheduler = Scheduler::current() ?? throw new \LogicException( + 'No active coroutine scope — run the test with #[RunInFiber] to use Coroutine::spawn().', + ); + + return new self($scheduler->spawn($body)); + } + + /** + * Run the given closures/fibers concurrently and wait for all of them. + * + * Sugar over {@see spawn()} + {@see await()}: schedules everything into the running scope, parks + * the caller until every coroutine finished, and returns the results keyed like the arguments + * (named arguments give string keys). Failures are collected until all coroutines settle, then + * bundled into one {@see CompositeException}. + * + * @return array Results keyed like the arguments. + * + * @throws CompositeException When any of the coroutines threw. + * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]`. + */ + public static function concurrently(\Closure|\Fiber ...$bodies): array + { + $handles = \array_map(self::spawn(...), $bodies); + + $results = $errors = []; + foreach ($handles as $key => $handle) { + try { + $results[$key] = $handle->await(); + } catch (CompositeException $e) { + $errors += $e->errors; + } + } + + $errors === [] or throw new CompositeException($errors); + + return $results; + } + + /** + * Whether the coroutine has settled — returned, thrown, or been cancelled. + */ + public function isFinished(): bool + { + return $this->task->finished; + } + + /** + * Park the calling coroutine until this one finishes, and return its result. + * + * Other coroutines — and, through the scope's relay, the case's other tests — keep running while + * the caller is parked. A throwable raised by the awaited coroutine is rethrown here wrapped in + * a {@see CompositeException}; rethrowing marks the failure as observed, so the scope will not + * report it again. + * + * @throws CompositeException When the awaited coroutine threw. + * @throws \LogicException When called outside a coroutine scope, or when a coroutine awaits itself. + */ + public function await(): mixed + { + while (!$this->task->finished) { + $caller = Scheduler::current()?->runningTask() ?? throw new \LogicException( + 'Coroutine::await() on a pending coroutine must be called from inside a coroutine scope.', + ); + $caller === $this->task and throw new \LogicException('A coroutine cannot await itself.'); + + $caller->awaiting = $this->task; + try { + \Fiber::suspend(); + } finally { + $caller->awaiting = null; + } + } + + if ($this->task->error !== null) { + $this->task->errorObserved = true; + throw new CompositeException([$this->task->id => $this->task->error]); + } + + return $this->task->result; + } +} diff --git a/plugin/fiber/src/Exception/CancelledException.php b/plugin/fiber/src/Exception/CancelledException.php new file mode 100644 index 00000000..6a148081 --- /dev/null +++ b/plugin/fiber/src/Exception/CancelledException.php @@ -0,0 +1,17 @@ + */ @@ -39,7 +40,7 @@ public function __construct(array $errors) parent::__construct( \sprintf( - "%d test fiber(s) failed while running the case batch:\n%s", + "%d fiber(s) failed:\n%s", \count($errors), \implode("\n", $lines), ), diff --git a/plugin/fiber/src/Exception/DeadlockException.php b/plugin/fiber/src/Exception/DeadlockException.php new file mode 100644 index 00000000..42964ee2 --- /dev/null +++ b/plugin/fiber/src/Exception/DeadlockException.php @@ -0,0 +1,18 @@ +spawn(static fn(): TestResult => $next($info)); + + $scheduler->drive($body); + + // The pipeline below captures test throwables as results, so a body error is unexpected + // infrastructure breakage — let it abort the pipeline. + $body->error === null or throw $body->error; + + /** @var TestResult $result */ + $result = $body->result; + + // Surface coroutine failures nobody awaited. An error rethrown by await() was observed — + // it already went through the body (and is part of its result); cancellations are ours. + $errors = []; + foreach ($scheduler->tasks() as $id => $task) { + if ($task === $body || $task->error === null || $task->errorObserved + || $task->error instanceof CancelledException + ) { + continue; + } + + $errors[$id] = $task->error; + } + + if ($errors !== []) { + // A failed body keeps its own failure as the root: chain it in front of the coroutine + // errors so nothing is dropped, and keep the harsher of the two statuses. + $result->failure === null or $errors = [$body->id => $result->failure] + $errors; + + $failed = $result->status === Status::Failed || $result->status === Status::Error; + $result = $result + ->with(status: $failed ? $result->status : Status::Error) + ->withFailure(new CompositeException($errors)); + } + + return $result; + } +} diff --git a/plugin/fiber/src/Internal/FiberTestBatchRunner.php b/plugin/fiber/src/Internal/FiberTestBatchRunner.php index 3b74672b..8f3f75e4 100644 --- a/plugin/fiber/src/Internal/FiberTestBatchRunner.php +++ b/plugin/fiber/src/Internal/FiberTestBatchRunner.php @@ -12,10 +12,10 @@ * Drives a case's test handlers on Testo's cooperative fiber {@see Scheduler}. * * An invokable runner — set on {@see \Testo\Core\Context\CaseInfo::$batchRunner} by - * {@see RunInFiberInterceptor::runTestCase()}. Wraps each handler in its own `\Fiber` and drives the - * whole set per the case {@see Schedule} (`Solo` to completion, or `RoundRobin` / `Random` interleaved). - * Each handler runs its test's pipeline synchronously inside the fiber, so Testo's fiber-aware guards - * cooperate and per-test state stays isolated across an interleave. + * {@see RunInFiberInterceptor::runTestCase()}. Spawns each handler as a task of a fresh scheduler and + * drives the whole set per the case {@see Schedule} (`Solo` to completion, or `RoundRobin` / `Random` + * interleaved). Each handler runs its test's pipeline synchronously inside the fiber, so Testo's + * fiber-aware guards cooperate and per-test state stays isolated across an interleave. * * @internal * @psalm-internal Testo\Fiber @@ -32,19 +32,26 @@ public function __construct( */ public function __invoke(array $handlers): array { - // One fiber per handler; the scheduler drives the whole set at once. - $fibers = \array_map(static fn(callable $handler): \Fiber => new \Fiber($handler), $handlers); + $scheduler = new Scheduler($this->schedule); + $tasks = \array_map( + static fn(callable $handler): Task => $scheduler->spawn($handler(...)), + $handlers, + ); - $errors = Scheduler::run($fibers, $this->schedule); + $scheduler->drive(); // Handlers never throw (a pipeline failure is captured as an Aborted result), so an error here is // unexpected; surface all of them together rather than dropping every failure but the first. + $errors = []; + foreach ($tasks as $i => $task) { + $task->error === null or $errors[$i] = $task->error; + } $errors === [] or throw new CompositeException($errors); return \array_map( /** @var TestResult */ - static fn(\Fiber $fiber): TestResult => $fiber->getReturn(), - $fibers, + static fn(Task $task): TestResult => $task->result, + $tasks, ); } } diff --git a/plugin/fiber/src/Internal/RunInFiberInterceptor.php b/plugin/fiber/src/Internal/RunInFiberInterceptor.php index 7ef86e81..0e1fb6a6 100644 --- a/plugin/fiber/src/Internal/RunInFiberInterceptor.php +++ b/plugin/fiber/src/Internal/RunInFiberInterceptor.php @@ -22,8 +22,8 @@ * - {@see runTestCase()} (class-level) sets a {@see FiberTestBatchRunner} on {@see CaseInfo::$batchRunner}; * `CaseRunner` reads it there and drives the whole case's batch on fibers per the class-level * {@see Schedule}. No container swapping, no re-emitted events. - * - {@see runTest()} (method-level) wraps a single test in its own fiber. When the case is already - * scheduling (a class-level `#[RunInFiber]`), it is a pass-through to avoid double-wrapping. + * - {@see runTest()} (method-level) wraps a single test in its own fiber. When a scheduler is already + * driving (a class-level `#[RunInFiber]`), it is a pass-through to avoid double-wrapping. * * Sits **outer** to the fiber-aware scoped-state guards (order just outside {@see * InterceptorOptions::ORDER_DATA_PROVIDER}): the method-level fiber wraps the whole per-test pipeline — @@ -32,6 +32,10 @@ * reads its own scoped state even while several interleave; a data-driven/retried test runs all its * datasets/attempts in its single fiber (data provider stays inner to the wrap). * + * The test's coroutine scope ({@see \Testo\Fiber\Coroutine}) is *not* opened here — that is + * {@see CoroutineScopeInterceptor}, wired by the same attribute at the innermost position, so spawned + * coroutines run inside the guards and read their test's scoped state. + * * @internal * @psalm-internal Testo\Fiber */ @@ -55,17 +59,18 @@ public function runTest(TestInfo $info, callable $next): TestResult { // Under a class-level #[RunInFiber] the batch runner already runs this test inside a scheduled // fiber — don't wrap it again. - if (Scheduler::active()) { + if (Scheduler::current() !== null) { return $next($info); } // Method-level #[RunInFiber] (no class scheduling): run this one test in its own fiber. - $fiber = new \Fiber(static fn(): TestResult => $next($info)); - $errors = Scheduler::run([$fiber], Schedule::Solo); + $scheduler = new Scheduler(Schedule::Solo); + $task = $scheduler->spawn(static fn(): TestResult => $next($info)); + $scheduler->drive(); - \array_key_exists(0, $errors) and throw $errors[0]; + $task->error === null or throw $task->error; /** @var TestResult */ - return $fiber->getReturn(); + return $task->result; } } diff --git a/plugin/fiber/src/Internal/Scheduler.php b/plugin/fiber/src/Internal/Scheduler.php index 7151bd7d..f5e2fede 100644 --- a/plugin/fiber/src/Internal/Scheduler.php +++ b/plugin/fiber/src/Internal/Scheduler.php @@ -4,92 +4,310 @@ namespace Testo\Fiber\Internal; +use Testo\Fiber\Exception\CancelledException; +use Testo\Fiber\Exception\DeadlockException; use Testo\Fiber\Schedule; /** - * Cooperative fiber scheduler for {@see \Testo\Fiber\RunInFiber}. + * Cooperative fiber scheduler for {@see \Testo\Fiber\RunInFiber} and {@see \Testo\Fiber\Coroutine}. * - * Drives a set of test fibers to completion on **plain fibers** (no event loop), switching between - * them only where the running fiber calls `\Fiber::suspend()`. This uses Testo's fiber-aware guard - * protocol (each guard re-suspends to its parent and swaps scoped state around the switch), so - * per-test assertion/messenger state stays isolated across an interleave. + * Drives a dynamic set of tasks to completion on **plain fibers** (no event loop), switching between + * them only where the running fiber calls `\Fiber::suspend()`. Tasks may be spawned while the + * scheduler is driving — under {@see Schedule::RoundRobin} they join the current round. + * + * When the scheduler itself runs inside a fiber (a test's coroutine scope under a case-level + * scheduler), it relays control upward after every round with a `\Fiber::suspend()` of its own — so + * its tasks keep interleaving with the outer schedule, and Testo's fiber-aware guards swap the + * scoped per-test state in and out at each relay. Tasks are only ever resumed from inside their own + * scheduler's drive frame, which is why coroutines always observe the state of the test that + * spawned them. * * @internal * @psalm-internal Testo\Fiber */ final class Scheduler { - private static int $depth = 0; + /** + * The scheduler owning the innermost task that is currently running. + */ + private static ?self $current = null; + + /** @var array */ + private array $tasks = []; + + private int $nextId = 0; + + private ?Task $running = null; + + public function __construct( + private readonly Schedule $schedule = Schedule::RoundRobin, + ) {} /** - * Whether a scheduler is currently driving (used by the interceptor to avoid re-wrapping a test - * that is already being scheduled). + * The scheduler whose task is currently running, if any. This is where the {@see \Testo\Fiber\Coroutine} + * helpers land: user code always runs inside a task, so the ambient scheduler is its scope. */ - public static function active(): bool + public static function current(): ?self { - return self::$depth > 0; + return self::$current; } /** - * Drive the given test fibers to completion under a {@see Schedule}. + * The task this scheduler is currently stepping. + */ + public function runningTask(): ?Task + { + return $this->running; + } + + /** + * @return array All scheduled tasks keyed by id, in spawn order. + */ + public function tasks(): array + { + return $this->tasks; + } + + /** + * Add a task to the schedule. May be called while the scheduler is driving. + * + * @param \Closure|\Fiber $body An unstarted fiber, or a closure to wrap into one. + */ + public function spawn(\Closure|\Fiber $body): Task + { + $fiber = $body instanceof \Fiber ? $body : new \Fiber($body); + $fiber->isStarted() and throw new \LogicException('Cannot schedule a fiber that has already been started.'); + + $id = $this->nextId++; + return $this->tasks[$id] = new Task($fiber, $this, $id); + } + + /** + * Drive the scheduled tasks to completion. + * + * A round steps every ready task once ({@see Schedule::RoundRobin}), or a single ready task + * ({@see Schedule::Solo} — always the first, so it runs to completion before the next starts; + * {@see Schedule::Random} — a random one). A task parked on an await is not ready until the + * awaited task finishes. Between rounds, when unfinished tasks remain and the scheduler runs + * inside a fiber, control is relayed to the parent scheduler. * - * - `Solo`: run each fiber to completion (resuming its own suspends) before the next. - * - `RoundRobin`: one step per ready fiber each round, in order. - * - `Random`: one step of a random ready fiber each round. + * With `$primary` set (a test's coroutine scope, where `$primary` is the test body), a primary + * failure cancels the remaining tasks instead of driving them further: a {@see CancelledException} + * is thrown into every pending fiber so its `finally` blocks run; a throwable escaping that unwind + * (other than the cancellation itself) is recorded as the task's error. * - * @param list<\Fiber> $fibers - * @return array Throwables thrown by fibers, keyed by fiber index (others done). + * An await cycle is broken by throwing a {@see DeadlockException} into the first parked task; + * the failure then cascades to its awaiters, so the deadlock surfaces as an ordinary task error + * with a stack trace pointing at the guilty `await()`. */ - public static function run(array $fibers, Schedule $schedule): array + public function drive(?Task $primary = null): void { - ++self::$depth; - $errors = []; + $prev = self::$current; + self::$current = $this; try { - if ($schedule === Schedule::Solo) { - foreach (\array_keys($fibers) as $i) { - // Drive this fiber to completion (resuming its own cooperative suspends) before - // moving on — no other fiber overlaps it. - while (!$fibers[$i]->isTerminated()) { - self::step($fibers, $i, $errors); + while (true) { + $ready = $parked = []; + foreach ($this->tasks as $id => $task) { + if ($task->finished) { + continue; } + + self::ready($task) ? $ready[] = $id : $parked[] = $id; } - return $errors; - } + if ($ready === [] && $parked === []) { + return; + } + + if ($ready === []) { + // Every unfinished task is parked in an await. If any of them is parked on + // another scheduler's task, the outer schedule may still unpark it — relay and + // retry. Otherwise no step can ever unpark them: an await cycle. + if (\Fiber::getCurrent() !== null && !$this->parkedTasksAreLocal($parked)) { + $this->relay($prev); + continue; + } + + // Break the cycle: the first parked task gets the deadlock at its await point + // and unwinds; its awaiters unpark and the failure cascades through the cycle. + $this->throwInto($this->tasks[$parked[0]], new DeadlockException($this->describeDeadlock($parked))); + continue; + } + + if ($this->schedule === Schedule::RoundRobin) { + // One step per ready task, in spawn order; tasks spawned during the round are + // appended to the list and get their first step in the same round. + for ($id = 0; $id < $this->nextId; $id++) { + $task = $this->tasks[$id]; + self::ready($task) and $this->step($task); + + if (self::failed($primary)) { + $this->cancelPending(); + return; + } + } + } else { + $pick = $this->schedule === Schedule::Solo + ? $ready[0] + : $ready[\random_int(0, \count($ready) - 1)]; + $this->step($this->tasks[$pick]); - $ready = \array_keys($fibers); - while ($ready !== []) { - // RoundRobin steps every ready fiber this round; Random steps one random ready fiber. - $round = $schedule === Schedule::Random ? [$ready[\random_int(0, \count($ready) - 1)]] : $ready; - foreach ($round as $i) { - self::step($fibers, $i, $errors); + if (self::failed($primary)) { + $this->cancelPending(); + return; + } } - $ready = \array_values(\array_filter($ready, static fn(int $i): bool => !$fibers[$i]->isTerminated())); + if (\Fiber::getCurrent() !== null && $this->hasUnfinished()) { + $this->relay($prev); + } } } finally { - --self::$depth; + self::$current = $prev; } + } + + private static function ready(Task $task): bool + { + return !$task->finished && ($task->awaiting === null || $task->awaiting->finished); + } + + private static function failed(?Task $primary): bool + { + return $primary !== null && $primary->finished && $primary->error !== null; + } - return $errors; + /** + * Give the running fiber a step: resume it (or start it), and record how it ended. + */ + private function step(Task $task): void + { + $previous = $this->running; + $this->running = $task; + try { + $task->fiber->isStarted() ? $task->fiber->resume() : $task->fiber->start(); + } catch (\Throwable $e) { + $task->error = $e; + } finally { + $this->running = $previous; + $this->settle($task); + } } /** - * @param list<\Fiber> $fibers - * @param array $errors + * Raise `$e` inside the task's fiber at its current suspension point. */ - private static function step(array $fibers, int $i, array &$errors): void + private function throwInto(Task $task, \Throwable $e): void { - $fiber = $fibers[$i]; - if ($fiber->isTerminated()) { + $previous = $this->running; + $this->running = $task; + try { + $task->fiber->throw($e); + } catch (\Throwable $err) { + $task->error = $err; + } finally { + $this->running = $previous; + $this->settle($task); + } + } + + private function settle(Task $task): void + { + if (!$task->fiber->isTerminated()) { return; } + $task->finished = true; + $task->error === null and $task->result = $task->fiber->getReturn(); + } + + /** + * Hand control to the parent scheduler until it steps us again. + */ + private function relay(?self $prev): void + { + self::$current = $prev; try { - $fiber->isStarted() ? $fiber->resume() : $fiber->start(); - } catch (\Throwable $e) { - // The fiber is terminated by the throw; record it against its index for the caller. - $errors[$i] = $e; + \Fiber::suspend(); + } finally { + self::$current = $this; + } + } + + private function hasUnfinished(): bool + { + foreach ($this->tasks as $task) { + if (!$task->finished) { + return true; + } } + + return false; + } + + /** + * @param non-empty-list $parked + */ + private function parkedTasksAreLocal(array $parked): bool + { + foreach ($parked as $id) { + if ($this->tasks[$id]->awaiting?->scheduler !== $this) { + return false; + } + } + + return true; + } + + /** + * Cancel every pending task: mark them all finished first (so an unwinding task that awaits a + * sibling sees it settled instead of parking forever), then throw a {@see CancelledException} + * into each started fiber so its `finally` blocks run. A fiber that swallows the cancellation + * and suspends again is resumed until it terminates. + */ + private function cancelPending(): void + { + $pending = []; + foreach ($this->tasks as $task) { + if (!$task->finished) { + $task->finished = true; + $pending[] = $task; + } + } + + foreach ($pending as $task) { + $fiber = $task->fiber; + if (!$fiber->isStarted()) { + continue; + } + + try { + $fiber->throw(new CancelledException('The coroutine scope is closing.')); + while (!$fiber->isTerminated()) { + $fiber->resume(); + } + } catch (CancelledException) { + // Unwound cleanly. + } catch (\Throwable $e) { + // A real failure while unwinding — the scope will surface it. + $task->error = $e; + } + } + } + + /** + * @param non-empty-list $parked + */ + private function describeDeadlock(array $parked): string + { + $lines = []; + foreach ($parked as $id) { + $lines[] = \sprintf('#%d awaits #%d', $id, $this->tasks[$id]->awaiting?->id ?? -1); + } + + return \sprintf( + 'Coroutine deadlock — every pending coroutine is parked on an await that can never complete: %s.', + \implode('; ', $lines), + ); } } diff --git a/plugin/fiber/src/Internal/Task.php b/plugin/fiber/src/Internal/Task.php new file mode 100644 index 00000000..498274b3 --- /dev/null +++ b/plugin/fiber/src/Internal/Task.php @@ -0,0 +1,46 @@ +status, Status::Passed); } + + public function unawaitedCoroutineFailureErrorsTheTest(): void + { + $result = TestRunner::runTest([FiberScenarios::class, 'unawaitedCoroutineFailure']); + + Assert::same($result->status, Status::Error); + # Coroutine failures always arrive as a composite, even a single one. + Assert::instanceOf($result->failure, CompositeException::class); + Assert::instanceOf($result->failure->getPrevious(), \RuntimeException::class); + } + + public function awaitedCoroutineFailureHandledInTestPasses(): void + { + $result = TestRunner::runTest([FiberScenarios::class, 'awaitedCoroutineFailureHandledInTest']); + + # await() marked the failure observed; the scope does not resurface it. + Assert::same($result->status, Status::Passed); + } + + public function bodyThrowKeepsWorkingWithExpectException(): void + { + $result = TestRunner::runTest([FiberScenarios::class, 'bodyThrowStaysUnwrapped']); + + # The body's own throw is not wrapped — #[ExpectException] matches it as usual. + Assert::same($result->status, Status::Passed); + } + + public function coroutineAssertionsCountTowardTheirTest(): void + { + $result = TestRunner::runTest([FiberScenarios::class, 'assertionsInsideCoroutinesCountForTheTest']); + + Assert::same($result->status, Status::Passed); + # 1 assert in the body + 2 inside the coroutine, attributed to the same test. + Assert::same($result->summary->metric('assertions'), 3); + } + + public function spawnWithoutScopeErrorsWithAHint(): void + { + $result = TestRunner::runTest([FiberScenarios::class, 'spawnWithoutFiberScope']); + + Assert::same($result->status, Status::Error); + Assert::instanceOf($result->failure, \LogicException::class); + Assert::string($result->failure->getMessage())->contains('RunInFiber'); + } } diff --git a/plugin/fiber/tests/Self/CoroutineInterleaveTest.php b/plugin/fiber/tests/Self/CoroutineInterleaveTest.php new file mode 100644 index 00000000..98408cdd --- /dev/null +++ b/plugin/fiber/tests/Self/CoroutineInterleaveTest.php @@ -0,0 +1,72 @@ + */ + private static array $log = []; + + public function first(): void + { + self::$log[] = 'first.body.1'; + $echo = Coroutine::spawn(static function (): string { + self::$log[] = 'first.co.1'; + \Fiber::suspend(); + self::$log[] = 'first.co.2'; + + return 'echo'; + }); + \Fiber::suspend(); + + self::$log[] = 'first.body.2'; + Assert::same($echo->await(), 'echo'); + + Assert::same(self::$log, [ + 'first.body.1', + 'first.co.1', + 'second.body.1', + 'first.body.2', + 'first.co.2', + 'second.body.2', + ]); + } + + public function second(): void + { + self::$log[] = 'second.body.1'; + \Fiber::suspend(); + self::$log[] = 'second.body.2'; + + Assert::same(self::$log, [ + 'first.body.1', + 'first.co.1', + 'second.body.1', + 'first.body.2', + 'first.co.2', + 'second.body.2', + ]); + } +} diff --git a/plugin/fiber/tests/Self/CoroutineScopeTest.php b/plugin/fiber/tests/Self/CoroutineScopeTest.php new file mode 100644 index 00000000..e4b0f9bd --- /dev/null +++ b/plugin/fiber/tests/Self/CoroutineScopeTest.php @@ -0,0 +1,62 @@ +isFinished()); + Assert::same($ping->await(), 'pong'); + Assert::true($ping->isFinished()); + } + + #[RunInFiber] + public function concurrentlyKeepsArgumentKeys(): void + { + $results = Coroutine::concurrently( + pull: static function (): string { + \Fiber::suspend(); + + return 'pulled'; + }, + push: static fn(): string => 'pushed', + ); + + Assert::same($results, ['pull' => 'pulled', 'push' => 'pushed']); + } + + #[RunInFiber] + public function acceptsAPreparedFiber(): void + { + $fiber = new \Fiber(static function (): int { + \Fiber::suspend(); + + return 7; + }); + + Assert::same(Coroutine::spawn($fiber)->await(), 7); + Assert::true($fiber->isTerminated()); + } +} diff --git a/plugin/fiber/tests/Stub/FiberScenarios.php b/plugin/fiber/tests/Stub/FiberScenarios.php index 49307c3b..472b9e43 100644 --- a/plugin/fiber/tests/Stub/FiberScenarios.php +++ b/plugin/fiber/tests/Stub/FiberScenarios.php @@ -5,6 +5,9 @@ namespace Tests\Fiber\Stub; use Testo\Assert; +use Testo\Assert\ExpectException; +use Testo\Fiber\Coroutine; +use Testo\Fiber\Exception\CompositeException; use Testo\Fiber\RunInFiber; use Testo\Test; @@ -31,4 +34,49 @@ public function untaggedRunsOnMainFiber(): void { Assert::null(\Fiber::getCurrent()); } + + #[RunInFiber] + public function unawaitedCoroutineFailure(): void + { + Coroutine::spawn(static fn() => throw new \RuntimeException('nobody awaited me')); + Assert::true(true); + } + + #[RunInFiber] + public function awaitedCoroutineFailureHandledInTest(): void + { + $bad = Coroutine::spawn(static fn() => throw new \RuntimeException('boom')); + try { + $bad->await(); + Assert::true(false); + } catch (CompositeException $e) { + Assert::instanceOf($e->getPrevious(), \RuntimeException::class); + } + } + + #[RunInFiber] + #[ExpectException(\DomainException::class)] + public function bodyThrowStaysUnwrapped(): void + { + Coroutine::spawn(static fn(): string => 'fine'); + + throw new \DomainException('straight from the body'); + } + + #[RunInFiber] + public function assertionsInsideCoroutinesCountForTheTest(): void + { + Coroutine::spawn(static function (): void { + Assert::true(true); + \Fiber::suspend(); + Assert::true(true); + })->await(); + + Assert::true(true); + } + + public function spawnWithoutFiberScope(): void + { + Coroutine::spawn(static fn(): string => 'no scope for me'); + } } diff --git a/plugin/fiber/tests/Unit/CoroutineCoverageTest.php b/plugin/fiber/tests/Unit/CoroutineCoverageTest.php new file mode 100644 index 00000000..382c8a56 --- /dev/null +++ b/plugin/fiber/tests/Unit/CoroutineCoverageTest.php @@ -0,0 +1,159 @@ +files), [self::FILE_BODY, self::FILE_COROUTINE]); + # Every slice survives the suspensions that split it up, on both sides of the spawn. + Assert::same(\array_keys($coverage->files[self::FILE_BODY]->lines), [1, 2]); + Assert::same(\array_keys($coverage->files[self::FILE_COROUTINE]->lines), [1, 2]); + } + + /** + * The invariant the collector's trampoline exists for still holds with a scope in the chain: when + * the scope relays a round to the outer schedule, no window is left open for a sibling test to + * record into. + */ + public function leavesNoWindowOpenWhenTheScopeRelaysOutward(): void + { + $driver = new WindowDriver(); + + $relays = 0; + self::drive($driver, static function () use ($driver, &$relays): void { + Assert::false($driver->open(), 'A coverage window is open while the test is parked.'); + ++$relays; + }); + + # Guard against a vacuous pass: the scope really did hand control outward. + Assert::true($relays > 0); + } + + /** + * The placement contract in one assertion: the scope takes the async-coroutine slot, which is + * inner to the coverage slot. + */ + public function isOrderedInsideTheCoverageSlot(): void + { + $options = (new \ReflectionClass(CoroutineScopeInterceptor::class)) + ->getAttributes(InterceptorOptions::class)[0] + ->newInstance(); + + Assert::same($options->order, InterceptorOptions::ORDER_ASYNC_COROUTINE); + Assert::true(InterceptorOptions::ORDER_ASYNC_COROUTINE > InterceptorOptions::ORDER_COVERAGE); + } + + /** + * Run one test — body plus a spawned coroutine, each touching two lines around a suspension — + * through `coverage(scope(body))`, driven from a fiber like the `#[RunInFiber]` wrap drives it. + * `$atRelay` is called wherever the scope hands control outward. + */ + private static function drive(WindowDriver $driver, ?callable $atRelay = null): CoverageResult + { + $interceptor = new CoverageTestInterceptor($driver); + $scope = new CoroutineScopeInterceptor(new RunInFiber()); + + $body = static function (TestInfo $info) use ($driver): TestResult { + $driver->touch(self::FILE_BODY, 1); + Coroutine::spawn(static function () use ($driver): void { + $driver->touch(self::FILE_COROUTINE, 1); + \Fiber::suspend(); + $driver->touch(self::FILE_COROUTINE, 2); + }); + + \Fiber::suspend(); + $driver->touch(self::FILE_BODY, 2); + + return new TestResult($info, Status::Passed); + }; + + $info = self::makeTestInfo(); + $fiber = new \Fiber(static fn(): TestResult => $interceptor->runTest( + $info, + static fn(TestInfo $i): TestResult => $scope->runTest($i, $body), + )); + + $fiber->start(); + while (!$fiber->isTerminated()) { + $atRelay === null or $atRelay(); + $fiber->resume(); + } + + /** @var TestResult $result */ + $result = $fiber->getReturn(); + $coverage = $result->getAttribute(CoverageResult::class); + + Assert::instanceOf($coverage, CoverageResult::class); + + return $coverage; + } + + private static function makeTestInfo(): TestInfo + { + return new TestInfo( + name: 'scopedTest', + caseInfo: new CaseInfo( + suiteIdentity: new SuiteIdentity('Fiber/Unit'), + definition: new CaseDefinition( + name: ScopedCase::class, + type: 'test', + file: Path::create(__FILE__), + reflection: new \ReflectionClass(ScopedCase::class), + ), + ), + testDefinition: new TestDefinition(new \ReflectionMethod(ScopedCase::class, 'scopedTest')), + ); + } +} + +/** + * Case shell the composed interceptors need a reflection of; the behaviour under test lives in the + * closures {@see CoroutineCoverageTest::drive()} passes down. + */ +final class ScopedCase +{ + public function scopedTest(): void {} +} diff --git a/plugin/fiber/tests/Unit/CoroutineTest.php b/plugin/fiber/tests/Unit/CoroutineTest.php new file mode 100644 index 00000000..1dbeffa1 --- /dev/null +++ b/plugin/fiber/tests/Unit/CoroutineTest.php @@ -0,0 +1,203 @@ + null); + } catch (\LogicException $e) { + $caught = $e; + } + + Assert::notNull($caught); + Assert::string($caught->getMessage())->contains('RunInFiber'); + } + + public function awaitReturnsTheCoroutineResult(): void + { + $result = $this->scope(static function (): mixed { + $sum = Coroutine::spawn(static function (): int { + \Fiber::suspend(); + return 40 + 2; + }); + + return $sum->await(); + }); + + Assert::same($result, 42); + } + + public function awaitRethrowsWrappedInAComposite(): void + { + $boom = new \RuntimeException('boom'); + + $caught = $this->scope(static function () use ($boom): mixed { + $bad = Coroutine::spawn(static fn() => throw $boom); + try { + $bad->await(); + } catch (CompositeException $e) { + return $e; + } + + return null; + }); + + # Even a single coroutine failure arrives as a composite, so handling code is uniform. + Assert::instanceOf($caught, CompositeException::class); + Assert::same(\array_values($caught->errors), [$boom]); + Assert::same($caught->getPrevious(), $boom); + } + + public function awaitOnAFinishedCoroutineReturnsImmediately(): void + { + $log = []; + $result = $this->scope(static function () use (&$log): mixed { + $quick = Coroutine::spawn(static fn(): string => 'done'); + \Fiber::suspend(); + $log[] = 'body'; + + return $quick->await(); + }); + + Assert::same($result, 'done'); + Assert::same($log, ['body']); + } + + public function selfAwaitThrows(): void + { + $caught = $this->scope(static function (): mixed { + $handle = null; + $inner = Coroutine::spawn(static function () use (&$handle): void { + \Fiber::suspend(); + $handle->await(); + }); + $handle = $inner; + try { + return $inner->await(); + } catch (CompositeException $e) { + return $e; + } + }); + + Assert::instanceOf($caught, CompositeException::class); + Assert::instanceOf($caught->getPrevious(), \LogicException::class); + } + + public function concurrentlyReturnsResultsKeyedLikeTheArguments(): void + { + $results = $this->scope(static function (): array { + return Coroutine::concurrently( + first: static function (): string { + \Fiber::suspend(); + return 'one'; + }, + second: static fn(): string => 'two', + ); + }); + + Assert::same($results, ['first' => 'one', 'second' => 'two']); + } + + public function concurrentlyBundlesEveryFailureAfterAllSettled(): void + { + $first = new \RuntimeException('first'); + $second = new \LogicException('second'); + $log = []; + + $caught = $this->scope(static function () use ($first, $second, &$log): mixed { + try { + Coroutine::concurrently( + static fn() => throw $first, + static function () use (&$log, $second): void { + \Fiber::suspend(); + $log[] = 'slow ran to its end'; + throw $second; + }, + ); + } catch (CompositeException $e) { + return $e; + } + + return null; + }); + + Assert::instanceOf($caught, CompositeException::class); + Assert::same(\array_values($caught->errors), [$first, $second]); + Assert::same($log, ['slow ran to its end']); + } + + public function awaitCycleIsBrokenAsADeadlock(): void + { + $caught = $this->scope(static function (): mixed { + $a = null; + $b = null; + $a = Coroutine::spawn(static function () use (&$b): void { + \Fiber::suspend(); + $b->await(); + }); + $b = Coroutine::spawn(static fn(): mixed => $a->await()); + + try { + return $a->await(); + } catch (DeadlockException $e) { + return $e; + } + }); + + # The first parked task (here: the body itself) gets the deadlock right at its await() call. + Assert::instanceOf($caught, DeadlockException::class); + Assert::string($caught->getMessage())->contains('await'); + } + + public function unfinishedCoroutinesAreDrivenAfterTheBodyReturns(): void + { + $log = []; + $this->scope(static function () use (&$log): void { + Coroutine::spawn(static function () use (&$log): void { + $log[] = 'child.1'; + \Fiber::suspend(); + $log[] = 'child.2'; + }); + $log[] = 'body done'; + }); + + Assert::same($log, ['body done', 'child.1', 'child.2']); + } + + /** + * Run `$body` as the primary task of a fresh coroutine scope and return its result. + */ + private function scope(\Closure $body): mixed + { + $scheduler = new Scheduler(); + $primary = $scheduler->spawn($body); + $scheduler->drive($primary); + + $primary->error === null or throw $primary->error; + + return $primary->result; + } +} diff --git a/plugin/fiber/tests/Unit/RunInFiberAttributesTest.php b/plugin/fiber/tests/Unit/RunInFiberAttributesTest.php index cbd245fa..0cbbbb6e 100644 --- a/plugin/fiber/tests/Unit/RunInFiberAttributesTest.php +++ b/plugin/fiber/tests/Unit/RunInFiberAttributesTest.php @@ -6,8 +6,11 @@ use Testo\Assert; use Testo\Codecov\Covers; +use Testo\Fiber\Internal\CoroutineScopeInterceptor; +use Testo\Fiber\Internal\RunInFiberInterceptor; use Testo\Fiber\RunInFiber; use Testo\Fiber\Schedule; +use Testo\Pipeline\Attribute\FallbackInterceptor; use Testo\Pipeline\Attribute\Interceptable; use Testo\Test; @@ -33,4 +36,14 @@ public function selfWiresAsInterceptable(): void { Assert::instanceOf(new RunInFiber(), Interceptable::class); } + + public function wiresTheFiberWrapAndTheCoroutineScope(): void + { + $classes = \array_map( + static fn(\ReflectionAttribute $attr): string => $attr->newInstance()->class, + (new \ReflectionClass(RunInFiber::class))->getAttributes(FallbackInterceptor::class), + ); + + Assert::same($classes, [RunInFiberInterceptor::class, CoroutineScopeInterceptor::class]); + } } diff --git a/plugin/fiber/tests/Unit/SchedulerTest.php b/plugin/fiber/tests/Unit/SchedulerTest.php index 95ae8d95..c96b5738 100644 --- a/plugin/fiber/tests/Unit/SchedulerTest.php +++ b/plugin/fiber/tests/Unit/SchedulerTest.php @@ -6,32 +6,28 @@ use Testo\Assert; use Testo\Codecov\Covers; +use Testo\Fiber\Exception\CancelledException; use Testo\Fiber\Internal\Scheduler; use Testo\Fiber\Schedule; use Testo\Test; /** - * Unit checks for the cooperative fiber scheduler driving `#[RunInFiber]`. Test fibers hand control + * Unit checks for the cooperative fiber scheduler driving `#[RunInFiber]` scopes. Tasks hand control * back to the scheduler by calling `\Fiber::suspend()`. */ #[Test] #[Covers(Scheduler::class)] final class SchedulerTest { - public function soloRunsEachFiberToCompletionInOrder(): void + public function soloRunsEachTaskToCompletionInOrder(): void { $log = []; - $make = function (string $id) use (&$log): \Fiber { - return new \Fiber(function () use ($id, &$log): void { - $log[] = "$id.1"; - \Fiber::suspend(); - $log[] = "$id.2"; - }); - }; + $scheduler = new Scheduler(Schedule::Solo); + $scheduler->spawn($this->logger('a', $log)); + $scheduler->spawn($this->logger('b', $log)); - $errors = Scheduler::run([$make('a'), $make('b')], Schedule::Solo); + $scheduler->drive(); - Assert::same($errors, []); # No interleaving: 'a' finishes before 'b' starts, the suspend just resumes the same fiber. Assert::same($log, ['a.1', 'a.2', 'b.1', 'b.2']); } @@ -39,51 +35,182 @@ public function soloRunsEachFiberToCompletionInOrder(): void public function roundRobinInterleavesAtSuspendPoints(): void { $log = []; - $make = function (string $id) use (&$log): \Fiber { - return new \Fiber(function () use ($id, &$log): void { - $log[] = "$id.1"; - \Fiber::suspend(); - $log[] = "$id.2"; - }); - }; + $scheduler = new Scheduler(Schedule::RoundRobin); + $scheduler->spawn($this->logger('a', $log)); + $scheduler->spawn($this->logger('b', $log)); - $errors = Scheduler::run([$make('a'), $make('b')], Schedule::RoundRobin); + $scheduler->drive(); - Assert::same($errors, []); Assert::same($log, ['a.1', 'b.1', 'a.2', 'b.2']); } - public function randomRunsEveryFiberToCompletion(): void + public function randomRunsEveryTaskToCompletion(): void { $done = []; - $make = function (string $id) use (&$done): \Fiber { - return new \Fiber(function () use ($id, &$done): void { + $make = static function (string $id) use (&$done): \Closure { + return static function () use ($id, &$done): void { \Fiber::suspend(); $done[] = $id; - }); + }; }; - $errors = Scheduler::run([$make('a'), $make('b'), $make('c')], Schedule::Random); + $scheduler = new Scheduler(Schedule::Random); + $scheduler->spawn($make('a')); + $scheduler->spawn($make('b')); + $scheduler->spawn($make('c')); + + $scheduler->drive(); \sort($done); - Assert::same($errors, []); Assert::same($done, ['a', 'b', 'c']); } - public function fiberThrowIsCapturedByIndex(): void + public function taskThrowIsRecordedOnTheTask(): void + { + $scheduler = new Scheduler(); + $ok = $scheduler->spawn(static fn(): string => 'fine'); + $bad = $scheduler->spawn(static fn() => throw new \RuntimeException('boom')); + + $scheduler->drive(); + + Assert::null($ok->error); + Assert::same($ok->result, 'fine'); + Assert::instanceOf($bad->error, \RuntimeException::class); + Assert::true($bad->finished); + } + + public function spawnDuringTheDriveJoinsTheCurrentRound(): void + { + $log = []; + $scheduler = new Scheduler(); + $scheduler->spawn(function () use (&$log, $scheduler): void { + $log[] = 'parent.1'; + $scheduler->spawn(static function () use (&$log): void { + $log[] = 'child.1'; + \Fiber::suspend(); + $log[] = 'child.2'; + }); + \Fiber::suspend(); + $log[] = 'parent.2'; + }); + + $scheduler->drive(); + + # The child got its first step in the round it was spawned, not a round later. + Assert::same($log, ['parent.1', 'child.1', 'parent.2', 'child.2']); + } + + public function currentPointsToTheDrivingSchedulerInsideATask(): void + { + Assert::null(Scheduler::current()); + + $seen = null; + $scheduler = new Scheduler(); + $scheduler->spawn(static function () use (&$seen): void { + $seen = Scheduler::current(); + }); + + $scheduler->drive(); + + Assert::same($seen, $scheduler); + Assert::null(Scheduler::current()); + } + + public function relaysToTheParentFiberBetweenRounds(): void + { + $scheduler = new Scheduler(); + $task = $scheduler->spawn(static function (): void { + \Fiber::suspend(); + }); + + $outer = new \Fiber(static fn() => $scheduler->drive()); + $outer->start(); + + # Round 1 stepped the task (it suspended); the scheduler relayed instead of spinning. + Assert::false($outer->isTerminated()); + Assert::false($task->finished); + + $outer->resume(); + + Assert::true($outer->isTerminated()); + Assert::true($task->finished); + } + + public function primaryFailureCancelsPendingTasks(): void + { + $log = []; + $scheduler = new Scheduler(); + $body = $scheduler->spawn(static function (): void { + \Fiber::suspend(); + throw new \RuntimeException('body died'); + }); + $child = $scheduler->spawn(static function () use (&$log): void { + try { + \Fiber::suspend(); + $log[] = 'unreachable'; + } finally { + $log[] = 'cleanup'; + } + }); + + $scheduler->drive($body); + + Assert::instanceOf($body->error, \RuntimeException::class); + Assert::true($child->finished); + # The child was unwound by the cancellation: its finally ran, no error recorded. + Assert::same($log, ['cleanup']); + Assert::null($child->error); + } + + public function swallowedCancellationIsDrivenToTermination(): void + { + $log = []; + $scheduler = new Scheduler(); + $body = $scheduler->spawn(static function (): void { + \Fiber::suspend(); + throw new \RuntimeException('body died'); + }); + $child = $scheduler->spawn(static function () use (&$log): void { + try { + \Fiber::suspend(); + } catch (CancelledException) { + $log[] = 'caught'; + } + $log[] = 'after'; + }); + + $scheduler->drive($body); + + Assert::true($child->finished); + Assert::same($log, ['caught', 'after']); + } + + public function rejectsAStartedFiber(): void { - $ok = new \Fiber(static fn() => null); - $bad = new \Fiber(static fn() => throw new \RuntimeException('boom')); + $fiber = new \Fiber(static fn() => \Fiber::suspend()); + $fiber->start(); + + $scheduler = new Scheduler(); - $errors = Scheduler::run([$ok, $bad], Schedule::RoundRobin); + $caught = null; + try { + $scheduler->spawn($fiber); + } catch (\LogicException $e) { + $caught = $e; + } - Assert::same(\array_keys($errors), [1]); - Assert::instanceOf($errors[1], \RuntimeException::class); + Assert::notNull($caught); } - public function activeIsFalseOutsideARun(): void + /** + * @param list $log + */ + private function logger(string $id, array &$log): \Closure { - # Scheduler::active() gates the interceptor's pass-through; it must be false when nothing runs. - Assert::false(Scheduler::active()); + return static function () use ($id, &$log): void { + $log[] = "$id.1"; + \Fiber::suspend(); + $log[] = "$id.2"; + }; } } diff --git a/skills/testo-fiber/SKILL.md b/skills/testo-fiber/SKILL.md index 8abd59ba..118ae4de 100644 --- a/skills/testo-fiber/SKILL.md +++ b/skills/testo-fiber/SKILL.md @@ -1,6 +1,6 @@ --- name: testo-fiber -description: Run Testo tests as cooperatively-scheduled plain PHP fibers with #[RunInFiber] — for fiber/coroutine code that suspends with \Fiber::suspend() and for interleaving a case's tests to shake out order-dependent races. Use when a test drives fibers, yields cooperatively, or needs deterministic interleaving. For real async I/O (amphp, Revolt timers/streams, Future::await()) use the testo/bridge-revolt #[RunInRevolt] attribute instead. +description: Run Testo tests as cooperatively-scheduled plain PHP fibers with #[RunInFiber] — for fiber/coroutine code that suspends with \Fiber::suspend() and for interleaving a case's tests to shake out order-dependent races. Coroutine::spawn()/await()/concurrently() add coroutines to the running test's schedule. Use when a test drives fibers, yields cooperatively, spawns concurrent coroutines, or needs deterministic interleaving. For real async I/O (amphp, Revolt timers/streams, Future::await()) use the testo/bridge-revolt #[RunInRevolt] attribute instead. --- # Fiber / coroutine tests in Testo @@ -9,12 +9,15 @@ Provided by the `testo/fiber` plugin (ships with Testo). It runs tests inside pl Fetch `https://php-testo.github.io/llms.txt` for the current attribute namespaces and parameters before writing code. -| Attribute | Level | Purpose | +| API | Level | Purpose | |---|---|---| | `#[RunInFiber]` | method | Run this test in its own fiber (so cooperative `\Fiber::suspend()` works). | | `#[RunInFiber(Schedule)]` | class | Schedule the case's tests: `Solo` (default), `RoundRobin` / `Random` cooperative interleaving. | +| `Coroutine::spawn(fn)` | in test | Add a coroutine to the running test's schedule; returns a `Coroutine` handle. | +| `$handle->await()` | in test | Park the caller until the coroutine finishes; return its result or rethrow its failure. | +| `Coroutine::concurrently(...)` | in test | Spawn several closures/fibers and wait for all; results keyed like the arguments. | -Everything lives in the `Testo\Fiber\` namespace (`Testo\Fiber\RunInFiber`, `Testo\Fiber\Schedule`). +Everything lives in the `Testo\Fiber\` namespace (`Testo\Fiber\RunInFiber`, `Testo\Fiber\Schedule`, `Testo\Fiber\Coroutine`). ## `#[RunInFiber]` — run a test in a fiber @@ -59,6 +62,40 @@ final class RaceTest - `RoundRobin` / `Random` interleave the case's tests on plain fibers, switching only where a fiber calls `\Fiber::suspend()`. Put a `\Fiber::suspend()` where a context switch should be allowed (in real use, the async driver the test exercises does this). Per-test assertion state stays isolated across the interleave. - Reports stay readable while tests interleave: each test carries a `TestIdentity`, so the terminal renders every test — its batch node, data sets, streamed `-vv` output and result line — as one contiguous block instead of splicing them together, and `--teamcity` stamps a per-test `flowId`. Blocks appear in the order tests finish, so a test that is not the one currently streaming shows up once it completes. +## `Coroutine` — spawn concurrent coroutines inside a test + +Every `#[RunInFiber]` test runs inside its own **coroutine scope**: the test body is the scope's first coroutine, and `Coroutine::spawn()` adds more to the same round-robin schedule. Coroutines interleave with the body (and each other) at every `\Fiber::suspend()`, and — under a class-level `#[RunInFiber]` — the whole scope keeps interleaving with the case's other tests. + +```php +use Testo\Assert; +use Testo\Fiber\Coroutine; +use Testo\Fiber\RunInFiber; +use Testo\Test; + +#[Test] +#[RunInFiber] +public function pingPong(): void +{ + $server = Coroutine::spawn(fn(): string => $this->acceptAndEcho()); // Closure or unstarted \Fiber + $client = Coroutine::spawn(fn(): string => $this->connectAndSend('ping')); + + Assert::same($client->await(), 'pong'); // parks the body; others keep running + Assert::true($server->isFinished()); + + // Sugar: spawn + await all; named arguments key the results. + $r = Coroutine::concurrently(pull: fn() => $q->pull(), push: fn() => $q->push(1)); + Assert::same($r['push'], 1); +} +``` + +Rules (verified against `plugin/fiber/src/Coroutine.php`): + +- `spawn()` needs an active scope — outside `#[RunInFiber]` it throws a `LogicException`. Assertions, messages **and coverage** inside a coroutine are attributed to the test that spawned it: the scope runs inside both the scoped-state guards and the test's coverage window. +- **The scope is structured**: the test is not finished until every coroutine it spawned is. Coroutines still pending when the body returns keep being driven; if the body *fails*, they are cancelled — a `Testo\Fiber\Exception\CancelledException` is thrown into each pending fiber (its `finally` blocks run; don't swallow it). +- **Coroutine failures always arrive wrapped in `Testo\Fiber\Exception\CompositeException`** — even a single one — whether rethrown by `await()` / `concurrently()` or reported at scope close for a coroutine nobody awaited (that marks the test `Error`). The body's own throw stays unwrapped, so `#[ExpectException]` on it works as usual; expect `CompositeException` when the throw comes from a coroutine. +- An await cycle is detected and broken with a `Testo\Fiber\Exception\DeadlockException` raised at the first parked `await()`. A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected. +- `concurrently()` waits for *all* its coroutines even after one fails, then bundles every failure into one composite. + ## Pitfalls - **This is NOT real async I/O.** There is no event loop. Awaiting a timer, socket, or `Future` (amphp/Revolt) does **not** work under `#[RunInFiber]` — a bare `\Fiber::suspend()` waiting on external I/O has no resumer. For real async work use the `testo/bridge-revolt` `#[RunInRevolt]` attribute (runs the test on the Revolt event loop). From e4e84082d8e5939188bb1658333b028da4c64395 Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Thu, 6 Aug 2026 16:25:26 +0400 Subject: [PATCH 3/9] fix(fiber): cancel pending coroutines when the test body fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pipeline below the coroutine scope captures test throwables into the TestResult, so a failed body settled with no error on its task and Scheduler::drive() never saw the failure — pending coroutines were driven to completion instead of being cancelled as documented. drive() now takes a failure predicate for the primary task, and CoroutineScopeInterceptor passes one that recognizes a failed result (Status::isFailure()), so the scope tears down as the docs promise. Assisted-By: Claude Fable 5 --- .../Internal/CoroutineScopeInterceptor.php | 9 +++++-- plugin/fiber/src/Internal/Scheduler.php | 20 +++++++++----- plugin/fiber/tests/Feature/StatusTest.php | 11 ++++++++ plugin/fiber/tests/Stub/FiberScenarios.php | 22 +++++++++++++++ plugin/fiber/tests/Unit/SchedulerTest.php | 27 +++++++++++++++++++ 5 files changed, 81 insertions(+), 8 deletions(-) diff --git a/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php b/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php index e40b4fab..2ce930c6 100644 --- a/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php +++ b/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php @@ -38,7 +38,9 @@ * * Coroutine failures nobody awaited fail the test: they are bundled into a {@see CompositeException} * (always, even a single one) and attached to the result as its failure with {@see Status::Error}. - * The body's own throw is captured as a result by the pipeline below and stays unwrapped. + * The body's own throw is captured as a result by the pipeline below and stays unwrapped. A body that + * settles with a failed result cancels its pending coroutines ({@see Scheduler::drive()}'s failure + * predicate) instead of driving them further; one that never got to start is simply dropped. * * @internal * @psalm-internal Testo\Fiber @@ -59,7 +61,10 @@ public function runTest(TestInfo $info, callable $next): TestResult $scheduler = new Scheduler(Schedule::RoundRobin); $body = $scheduler->spawn(static fn(): TestResult => $next($info)); - $scheduler->drive($body); + // The pipeline below captures test throwables into the result, so a failed body settles with + // no error on the task — the predicate is how the scheduler learns to cancel pending coroutines. + $scheduler->drive($body, static fn(Task $task): bool => + $task->result instanceof TestResult && $task->result->status->isFailure()); // The pipeline below captures test throwables as results, so a body error is unexpected // infrastructure breakage — let it abort the pipeline. diff --git a/plugin/fiber/src/Internal/Scheduler.php b/plugin/fiber/src/Internal/Scheduler.php index f5e2fede..0a1560ed 100644 --- a/plugin/fiber/src/Internal/Scheduler.php +++ b/plugin/fiber/src/Internal/Scheduler.php @@ -94,13 +94,17 @@ public function spawn(\Closure|\Fiber $body): Task * With `$primary` set (a test's coroutine scope, where `$primary` is the test body), a primary * failure cancels the remaining tasks instead of driving them further: a {@see CancelledException} * is thrown into every pending fiber so its `finally` blocks run; a throwable escaping that unwind - * (other than the cancellation itself) is recorded as the task's error. + * (other than the cancellation itself) is recorded as the task's error. A failure is an error that + * escaped the primary fiber, or `$primaryFailed` returning `true` for the settled task — the + * caller's chance to recognize failures its pipeline captured into the task's result. + * + * @param null|\Closure(Task): bool $primaryFailed * * An await cycle is broken by throwing a {@see DeadlockException} into the first parked task; * the failure then cascades to its awaiters, so the deadlock surfaces as an ordinary task error * with a stack trace pointing at the guilty `await()`. */ - public function drive(?Task $primary = null): void + public function drive(?Task $primary = null, ?\Closure $primaryFailed = null): void { $prev = self::$current; self::$current = $this; @@ -141,7 +145,7 @@ public function drive(?Task $primary = null): void $task = $this->tasks[$id]; self::ready($task) and $this->step($task); - if (self::failed($primary)) { + if (self::failed($primary, $primaryFailed)) { $this->cancelPending(); return; } @@ -152,7 +156,7 @@ public function drive(?Task $primary = null): void : $ready[\random_int(0, \count($ready) - 1)]; $this->step($this->tasks[$pick]); - if (self::failed($primary)) { + if (self::failed($primary, $primaryFailed)) { $this->cancelPending(); return; } @@ -172,9 +176,13 @@ private static function ready(Task $task): bool return !$task->finished && ($task->awaiting === null || $task->awaiting->finished); } - private static function failed(?Task $primary): bool + /** + * @param null|\Closure(Task): bool $predicate + */ + private static function failed(?Task $primary, ?\Closure $predicate): bool { - return $primary !== null && $primary->finished && $primary->error !== null; + return $primary !== null && $primary->finished + && ($primary->error !== null || $predicate !== null && $predicate($primary)); } /** diff --git a/plugin/fiber/tests/Feature/StatusTest.php b/plugin/fiber/tests/Feature/StatusTest.php index 174a78ac..5dda7913 100644 --- a/plugin/fiber/tests/Feature/StatusTest.php +++ b/plugin/fiber/tests/Feature/StatusTest.php @@ -85,6 +85,17 @@ public function coroutineAssertionsCountTowardTheirTest(): void Assert::same($result->summary->metric('assertions'), 3); } + public function failingBodyCancelsPendingCoroutines(): void + { + FiberScenarios::$cancellationLog = []; + + $result = TestRunner::runTest([FiberScenarios::class, 'failingBodyLeavesAPendingCoroutine']); + + Assert::same($result->status, Status::Failed); + # The pending coroutine was cancelled at its suspension point, not driven to completion. + Assert::same(FiberScenarios::$cancellationLog, ['cancelled']); + } + public function spawnWithoutScopeErrorsWithAHint(): void { $result = TestRunner::runTest([FiberScenarios::class, 'spawnWithoutFiberScope']); diff --git a/plugin/fiber/tests/Stub/FiberScenarios.php b/plugin/fiber/tests/Stub/FiberScenarios.php index 472b9e43..1d3df3b7 100644 --- a/plugin/fiber/tests/Stub/FiberScenarios.php +++ b/plugin/fiber/tests/Stub/FiberScenarios.php @@ -7,6 +7,7 @@ use Testo\Assert; use Testo\Assert\ExpectException; use Testo\Fiber\Coroutine; +use Testo\Fiber\Exception\CancelledException; use Testo\Fiber\Exception\CompositeException; use Testo\Fiber\RunInFiber; use Testo\Test; @@ -18,6 +19,9 @@ #[Test] final class FiberScenarios { + /** @var list What a pending coroutine observed when its test's body failed. */ + public static array $cancellationLog = []; + #[RunInFiber] public function runsInAFiber(): void { @@ -79,4 +83,22 @@ public function spawnWithoutFiberScope(): void { Coroutine::spawn(static fn(): string => 'no scope for me'); } + + #[RunInFiber] + public function failingBodyLeavesAPendingCoroutine(): void + { + Coroutine::spawn(static function (): void { + try { + \Fiber::suspend(); + self::$cancellationLog[] = 'survived'; + } catch (CancelledException) { + self::$cancellationLog[] = 'cancelled'; + } + }); + + // Let the coroutine reach its suspension point before the body fails. + \Fiber::suspend(); + + Assert::same(1, 2); + } } diff --git a/plugin/fiber/tests/Unit/SchedulerTest.php b/plugin/fiber/tests/Unit/SchedulerTest.php index c96b5738..22acd334 100644 --- a/plugin/fiber/tests/Unit/SchedulerTest.php +++ b/plugin/fiber/tests/Unit/SchedulerTest.php @@ -8,6 +8,7 @@ use Testo\Codecov\Covers; use Testo\Fiber\Exception\CancelledException; use Testo\Fiber\Internal\Scheduler; +use Testo\Fiber\Internal\Task; use Testo\Fiber\Schedule; use Testo\Test; @@ -162,6 +163,32 @@ public function primaryFailureCancelsPendingTasks(): void Assert::null($child->error); } + public function primaryFailedPredicateCancelsPendingTasks(): void + { + $log = []; + $scheduler = new Scheduler(); + $body = $scheduler->spawn(static function (): string { + \Fiber::suspend(); + + return 'captured failure'; + }); + $child = $scheduler->spawn(static function () use (&$log): void { + try { + \Fiber::suspend(); + $log[] = 'survived'; + } catch (CancelledException) { + $log[] = 'cancelled'; + } + }); + + # The primary settles without an error — the predicate is what recognizes the failure. + $scheduler->drive($body, static fn(Task $task): bool => $task->result === 'captured failure'); + + Assert::null($body->error); + Assert::true($child->finished); + Assert::same($log, ['cancelled']); + } + public function swallowedCancellationIsDrivenToTermination(): void { $log = []; From aef76d36a4bf55349bb6614f757f8a003314ec0c Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Thu, 6 Aug 2026 17:20:36 +0400 Subject: [PATCH 4/9] fix(fiber): detect await cycles that span coroutine scopes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An await cycle crossing schedulers — a scope's coroutine awaiting another scope's through a shared handle — was invisible to the local deadlock check: every scope saw its parked tasks as possibly unparkable by the outer schedule and relayed forever, livelocking the run. The check now walks the awaiting links themselves, which naturally cross scheduler boundaries: a chain that runs into a cycle can never be unparked by any schedule, so the first doomed task gets the DeadlockException; a chain that ends outside a cycle still relays. Each scheduler only ever throws into its own tasks, so coroutines are still resumed exclusively from their own scope's drive frame. The deadlock message marks foreign links as "another scope's" — task ids are per-scheduler and would collide unqualified. Assisted-By: Claude Fable 5 --- .../fiber/src/Exception/DeadlockException.php | 7 ++- plugin/fiber/src/Internal/Scheduler.php | 55 +++++++++++++----- plugin/fiber/tests/Unit/SchedulerTest.php | 57 +++++++++++++++++++ skills/testo-fiber/SKILL.md | 2 +- 4 files changed, 103 insertions(+), 18 deletions(-) diff --git a/plugin/fiber/src/Exception/DeadlockException.php b/plugin/fiber/src/Exception/DeadlockException.php index 42964ee2..f06909d8 100644 --- a/plugin/fiber/src/Exception/DeadlockException.php +++ b/plugin/fiber/src/Exception/DeadlockException.php @@ -5,10 +5,11 @@ namespace Testo\Fiber\Exception; /** - * Every pending coroutine of a scope is parked on an {@see \Testo\Fiber\Coroutine::await()} that can - * never complete — an await cycle. + * A coroutine is parked on an {@see \Testo\Fiber\Coroutine::await()} that can never complete — an + * await cycle, including one spanning several tests' scopes when handles are shared under a + * class-level `#[RunInFiber]`. * - * The scheduler breaks the cycle by raising this at the first parked coroutine's `await()` call, so + * The scheduler breaks the cycle by raising this at the first doomed coroutine's `await()` call, so * the stack trace points at the guilty wait; the failure then cascades to the coroutines awaiting it. * A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected — only * `await()` parks a coroutine in a way the scheduler can reason about. diff --git a/plugin/fiber/src/Internal/Scheduler.php b/plugin/fiber/src/Internal/Scheduler.php index 0a1560ed..071b0c56 100644 --- a/plugin/fiber/src/Internal/Scheduler.php +++ b/plugin/fiber/src/Internal/Scheduler.php @@ -100,9 +100,10 @@ public function spawn(\Closure|\Fiber $body): Task * * @param null|\Closure(Task): bool $primaryFailed * - * An await cycle is broken by throwing a {@see DeadlockException} into the first parked task; - * the failure then cascades to its awaiters, so the deadlock surfaces as an ordinary task error - * with a stack trace pointing at the guilty `await()`. + * An await cycle — even one spanning several schedulers' tasks — is broken by throwing a + * {@see DeadlockException} into the first task the cycle dooms; the failure then cascades to its + * awaiters, so the deadlock surfaces as an ordinary task error with a stack trace pointing at + * the guilty `await()`. */ public function drive(?Task $primary = null, ?\Closure $primaryFailed = null): void { @@ -124,17 +125,22 @@ public function drive(?Task $primary = null, ?\Closure $primaryFailed = null): v } if ($ready === []) { - // Every unfinished task is parked in an await. If any of them is parked on - // another scheduler's task, the outer schedule may still unpark it — relay and - // retry. Otherwise no step can ever unpark them: an await cycle. - if (\Fiber::getCurrent() !== null && !$this->parkedTasksAreLocal($parked)) { + // Every unfinished task is parked in an await. A task whose await chain runs + // into a cycle — even one spanning other schedulers — can never be unparked by + // any schedule; a chain that ends outside a cycle may still be unparked by the + // outer schedule, so with only those left, relay and retry. + $doomed = $this->deadlocked($parked); + if ($doomed === [] && \Fiber::getCurrent() !== null) { $this->relay($prev); continue; } - // Break the cycle: the first parked task gets the deadlock at its await point + // Break the cycle: the first doomed task gets the deadlock at its await point // and unwinds; its awaiters unpark and the failure cascades through the cycle. - $this->throwInto($this->tasks[$parked[0]], new DeadlockException($this->describeDeadlock($parked))); + // With no fiber to relay from, tasks parked on foreign tasks are stuck the same + // way — nobody else will ever drive those. + $stuck = $doomed === [] ? $parked : $doomed; + $this->throwInto($this->tasks[$stuck[0]], new DeadlockException($this->describeDeadlock($stuck))); continue; } @@ -254,17 +260,32 @@ private function hasUnfinished(): bool } /** + * Ids of parked tasks whose await chain runs into a cycle, so no schedule can ever unpark them. + * The chain follows {@see Task::$awaiting} links across schedulers — a scope's coroutine may + * await another scope's — and stops at a finished task or one that is suspended without + * awaiting (its own scheduler may still step it). + * * @param non-empty-list $parked + * @return list */ - private function parkedTasksAreLocal(array $parked): bool + private function deadlocked(array $parked): array { + $doomed = []; foreach ($parked as $id) { - if ($this->tasks[$id]->awaiting?->scheduler !== $this) { - return false; + $chain = []; + $task = $this->tasks[$id]; + while ($task !== null && !$task->finished) { + if (\in_array($task, $chain, true)) { + $doomed[] = $id; + break; + } + + $chain[] = $task; + $task = $task->awaiting; } } - return true; + return $doomed; } /** @@ -310,7 +331,13 @@ private function describeDeadlock(array $parked): string { $lines = []; foreach ($parked as $id) { - $lines[] = \sprintf('#%d awaits #%d', $id, $this->tasks[$id]->awaiting?->id ?? -1); + $target = $this->tasks[$id]->awaiting; + $lines[] = \sprintf( + '#%d awaits %s#%d', + $id, + $target === null || $target->scheduler === $this ? '' : "another scope's ", + $target?->id ?? -1, + ); } return \sprintf( diff --git a/plugin/fiber/tests/Unit/SchedulerTest.php b/plugin/fiber/tests/Unit/SchedulerTest.php index 22acd334..eeeb7d3d 100644 --- a/plugin/fiber/tests/Unit/SchedulerTest.php +++ b/plugin/fiber/tests/Unit/SchedulerTest.php @@ -6,7 +6,9 @@ use Testo\Assert; use Testo\Codecov\Covers; +use Testo\Fiber\Coroutine; use Testo\Fiber\Exception\CancelledException; +use Testo\Fiber\Exception\DeadlockException; use Testo\Fiber\Internal\Scheduler; use Testo\Fiber\Internal\Task; use Testo\Fiber\Schedule; @@ -212,6 +214,61 @@ public function swallowedCancellationIsDrivenToTermination(): void Assert::same($log, ['caught', 'after']); } + /** + * Two scopes driven by an outer schedule, each with a coroutine awaiting the other scope's + * coroutine through shared handles — an await cycle spanning schedulers. Neither scope may spin + * relaying forever: the cycle must be detected and broken like a local one. The outer loop is + * bounded so a livelock fails the test instead of hanging it. + */ + public function crossSchedulerAwaitCycleIsBrokenAsADeadlock(): void + { + $handleA = $handleB = null; + + $scopeA = new Scheduler(); + $bodyA = $scopeA->spawn(static function () use (&$handleA, &$handleB): mixed { + $handleA = Coroutine::spawn(static function () use (&$handleB): mixed { + while ($handleB === null) { + \Fiber::suspend(); + } + + return $handleB->await(); + }); + + return $handleA->await(); + }); + + $scopeB = new Scheduler(); + $bodyB = $scopeB->spawn(static function () use (&$handleA, &$handleB): mixed { + $handleB = Coroutine::spawn(static fn(): mixed => $handleA->await()); + + return $handleB->await(); + }); + + $fiberA = new \Fiber(static fn() => $scopeA->drive($bodyA)); + $fiberB = new \Fiber(static fn() => $scopeB->drive($bodyB)); + + for ($i = 0; $i < 100 && !($fiberA->isTerminated() && $fiberB->isTerminated()); $i++) { + $fiberA->isTerminated() or ($fiberA->isStarted() ? $fiberA->resume() : $fiberA->start()); + $fiberB->isTerminated() or ($fiberB->isStarted() ? $fiberB->resume() : $fiberB->start()); + } + + Assert::true( + $fiberA->isTerminated() && $fiberB->isTerminated(), + 'The cross-scheduler await cycle was never broken — the scopes relay forever.', + ); + + # Both bodies failed, and the deadlock is the root of the cascade in at least one of them. + Assert::notNull($bodyA->error); + Assert::notNull($bodyB->error); + $deadlocked = false; + foreach ([$bodyA->error, $bodyB->error] as $error) { + for (; $error !== null; $error = $error->getPrevious()) { + $error instanceof DeadlockException and $deadlocked = true; + } + } + Assert::true($deadlocked); + } + public function rejectsAStartedFiber(): void { $fiber = new \Fiber(static fn() => \Fiber::suspend()); diff --git a/skills/testo-fiber/SKILL.md b/skills/testo-fiber/SKILL.md index 118ae4de..e0f5033b 100644 --- a/skills/testo-fiber/SKILL.md +++ b/skills/testo-fiber/SKILL.md @@ -93,7 +93,7 @@ Rules (verified against `plugin/fiber/src/Coroutine.php`): - `spawn()` needs an active scope — outside `#[RunInFiber]` it throws a `LogicException`. Assertions, messages **and coverage** inside a coroutine are attributed to the test that spawned it: the scope runs inside both the scoped-state guards and the test's coverage window. - **The scope is structured**: the test is not finished until every coroutine it spawned is. Coroutines still pending when the body returns keep being driven; if the body *fails*, they are cancelled — a `Testo\Fiber\Exception\CancelledException` is thrown into each pending fiber (its `finally` blocks run; don't swallow it). - **Coroutine failures always arrive wrapped in `Testo\Fiber\Exception\CompositeException`** — even a single one — whether rethrown by `await()` / `concurrently()` or reported at scope close for a coroutine nobody awaited (that marks the test `Error`). The body's own throw stays unwrapped, so `#[ExpectException]` on it works as usual; expect `CompositeException` when the throw comes from a coroutine. -- An await cycle is detected and broken with a `Testo\Fiber\Exception\DeadlockException` raised at the first parked `await()`. A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected. +- An await cycle is detected and broken with a `Testo\Fiber\Exception\DeadlockException` raised at the first doomed `await()` — even a cycle spanning several tests' scopes (handles shared under a class-level `#[RunInFiber]`). A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected. - `concurrently()` waits for *all* its coroutines even after one fails, then bundles every failure into one composite. ## Pitfalls From fcd73a5d61a44438cad73ff8c60cc18b78ec3f89 Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Thu, 6 Aug 2026 17:25:32 +0400 Subject: [PATCH 5/9] fix(fiber): reject spawn while the coroutine scope is closing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A task spawned during the scope teardown — e.g. from a cancelled coroutine's finally block — silently joined a schedule nobody drives anymore: never stepped, never reported. The scheduler now marks itself closing when it starts cancelling pending tasks and rejects further spawns with a LogicException, which surfaces through the unwinding coroutine's error instead of vanishing. Assisted-By: Claude Fable 5 --- plugin/fiber/src/Coroutine.php | 3 ++- plugin/fiber/src/Internal/Scheduler.php | 13 ++++++++++++- plugin/fiber/tests/Unit/SchedulerTest.php | 23 +++++++++++++++++++++++ 3 files changed, 37 insertions(+), 2 deletions(-) diff --git a/plugin/fiber/src/Coroutine.php b/plugin/fiber/src/Coroutine.php index da464448..9b8a62df 100644 --- a/plugin/fiber/src/Coroutine.php +++ b/plugin/fiber/src/Coroutine.php @@ -50,7 +50,8 @@ private function __construct( * The coroutine gets its first step in the current scheduling round; from there it runs * cooperatively — it holds the floor until it suspends, finishes, or awaits. * - * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]`. + * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]` — + * or when the scope is already closing (spawning from a cancelled coroutine's `finally`). */ public static function spawn(\Closure|\Fiber $body): self { diff --git a/plugin/fiber/src/Internal/Scheduler.php b/plugin/fiber/src/Internal/Scheduler.php index 071b0c56..050b800d 100644 --- a/plugin/fiber/src/Internal/Scheduler.php +++ b/plugin/fiber/src/Internal/Scheduler.php @@ -39,6 +39,11 @@ final class Scheduler private ?Task $running = null; + /** + * The scope is tearing down ({@see cancelPending()}): nothing will be scheduled anymore. + */ + private bool $closing = false; + public function __construct( private readonly Schedule $schedule = Schedule::RoundRobin, ) {} @@ -69,12 +74,16 @@ public function tasks(): array } /** - * Add a task to the schedule. May be called while the scheduler is driving. + * Add a task to the schedule. May be called while the scheduler is driving — but not while the + * scope is closing: a task spawned during the teardown (a cancelled coroutine's `finally`) would + * silently join a schedule nobody drives anymore. * * @param \Closure|\Fiber $body An unstarted fiber, or a closure to wrap into one. */ public function spawn(\Closure|\Fiber $body): Task { + $this->closing and throw new \LogicException('Cannot spawn a coroutine while its scope is closing.'); + $fiber = $body instanceof \Fiber ? $body : new \Fiber($body); $fiber->isStarted() and throw new \LogicException('Cannot schedule a fiber that has already been started.'); @@ -296,6 +305,8 @@ private function deadlocked(array $parked): array */ private function cancelPending(): void { + $this->closing = true; + $pending = []; foreach ($this->tasks as $task) { if (!$task->finished) { diff --git a/plugin/fiber/tests/Unit/SchedulerTest.php b/plugin/fiber/tests/Unit/SchedulerTest.php index eeeb7d3d..08a77add 100644 --- a/plugin/fiber/tests/Unit/SchedulerTest.php +++ b/plugin/fiber/tests/Unit/SchedulerTest.php @@ -191,6 +191,29 @@ public function primaryFailedPredicateCancelsPendingTasks(): void Assert::same($log, ['cancelled']); } + public function spawnWhileTheScopeIsClosingThrows(): void + { + $scheduler = new Scheduler(); + $body = $scheduler->spawn(static function (): void { + \Fiber::suspend(); + throw new \RuntimeException('body died'); + }); + $child = $scheduler->spawn(static function () use ($scheduler): void { + try { + \Fiber::suspend(); + } finally { + $scheduler->spawn(static fn(): string => 'cleanup nobody will ever drive'); + } + }); + + $scheduler->drive($body); + + # The late spawn was rejected loudly, not silently added to a schedule nobody drives anymore. + Assert::instanceOf($child->error, \LogicException::class); + Assert::string($child->error->getMessage())->contains('closing'); + Assert::same(\count($scheduler->tasks()), 2); + } + public function swallowedCancellationIsDrivenToTermination(): void { $log = []; From cb69cf11b8cf676f6f386d661f6c8d52c21366cf Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Thu, 6 Aug 2026 19:21:35 +0400 Subject: [PATCH 6/9] fix(fiber): rethrow the cancellation when awaiting a cancelled coroutine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cancelled task settled with finished=true and a null error, so await() returned null — indistinguishable from a legitimate null result for a sibling's finally unwinding on the same cancellation. The task now remembers it was cancelled, and await() rethrows a CancelledException instead of forging a result. The exception is deliberately unwrapped: cancellation is the scope's control signal, not a failure raised by the coroutine, so the CompositeException contract does not apply — matching how the scope already keeps cancellations out of the surfaced errors. Assisted-By: Claude Fable 5 --- plugin/fiber/src/Coroutine.php | 7 +++- .../src/Exception/CancelledException.php | 3 ++ plugin/fiber/src/Internal/Scheduler.php | 1 + plugin/fiber/src/Internal/Task.php | 6 ++++ plugin/fiber/tests/Unit/CoroutineTest.php | 33 +++++++++++++++++++ skills/testo-fiber/SKILL.md | 2 +- 6 files changed, 50 insertions(+), 2 deletions(-) diff --git a/plugin/fiber/src/Coroutine.php b/plugin/fiber/src/Coroutine.php index 9b8a62df..64177da1 100644 --- a/plugin/fiber/src/Coroutine.php +++ b/plugin/fiber/src/Coroutine.php @@ -107,9 +107,12 @@ public function isFinished(): bool * Other coroutines — and, through the scope's relay, the case's other tests — keep running while * the caller is parked. A throwable raised by the awaited coroutine is rethrown here wrapped in * a {@see CompositeException}; rethrowing marks the failure as observed, so the scope will not - * report it again. + * report it again. Awaiting a cancelled coroutine rethrows the cancellation — unwrapped, it is + * the scope's control signal rather than a failure of the coroutine — so a `finally` unwinding + * on the same cancellation cannot mistake a torn-down sibling for one that returned `null`. * * @throws CompositeException When the awaited coroutine threw. + * @throws CancelledException When the awaited coroutine was cancelled with its scope. * @throws \LogicException When called outside a coroutine scope, or when a coroutine awaits itself. */ public function await(): mixed @@ -133,6 +136,8 @@ public function await(): mixed throw new CompositeException([$this->task->id => $this->task->error]); } + $this->task->cancelled and throw new CancelledException('The awaited coroutine was cancelled with its scope.'); + return $this->task->result; } } diff --git a/plugin/fiber/src/Exception/CancelledException.php b/plugin/fiber/src/Exception/CancelledException.php index 6a148081..7d906e74 100644 --- a/plugin/fiber/src/Exception/CancelledException.php +++ b/plugin/fiber/src/Exception/CancelledException.php @@ -12,6 +12,9 @@ * Don't swallow it: a coroutine that catches the cancellation and suspends again is resumed until it * terminates, but it has no schedule to cooperate with anymore. * + * Also rethrown by {@see \Testo\Fiber\Coroutine::await()} on a cancelled coroutine — it has no + * result to report. + * * @api */ final class CancelledException extends \RuntimeException {} diff --git a/plugin/fiber/src/Internal/Scheduler.php b/plugin/fiber/src/Internal/Scheduler.php index 050b800d..3c091774 100644 --- a/plugin/fiber/src/Internal/Scheduler.php +++ b/plugin/fiber/src/Internal/Scheduler.php @@ -311,6 +311,7 @@ private function cancelPending(): void foreach ($this->tasks as $task) { if (!$task->finished) { $task->finished = true; + $task->cancelled = true; $pending[] = $task; } } diff --git a/plugin/fiber/src/Internal/Task.php b/plugin/fiber/src/Internal/Task.php index 498274b3..06fcf5bc 100644 --- a/plugin/fiber/src/Internal/Task.php +++ b/plugin/fiber/src/Internal/Task.php @@ -32,6 +32,12 @@ final class Task */ public bool $errorObserved = false; + /** + * The scope cancelled the task while it was still pending — it has no result to report, even if + * its fiber swallowed the cancellation and terminated on its own terms. + */ + public bool $cancelled = false; + /** * The task this one is parked on (inside {@see \Testo\Fiber\Coroutine::await()}) — * not ready while the target is unfinished. diff --git a/plugin/fiber/tests/Unit/CoroutineTest.php b/plugin/fiber/tests/Unit/CoroutineTest.php index 1dbeffa1..edc5f996 100644 --- a/plugin/fiber/tests/Unit/CoroutineTest.php +++ b/plugin/fiber/tests/Unit/CoroutineTest.php @@ -7,6 +7,7 @@ use Testo\Assert; use Testo\Codecov\Covers; use Testo\Fiber\Coroutine; +use Testo\Fiber\Exception\CancelledException; use Testo\Fiber\Exception\CompositeException; use Testo\Fiber\Exception\DeadlockException; use Testo\Fiber\Internal\Scheduler; @@ -172,6 +173,38 @@ public function awaitCycleIsBrokenAsADeadlock(): void Assert::string($caught->getMessage())->contains('await'); } + /** + * A cancelled coroutine has no result to report: awaiting it from the teardown (a sibling's + * `catch`/`finally` unwinding on the same cancellation) rethrows the cancellation instead of + * forging a `null` result. + */ + public function awaitOnACancelledCoroutineThrowsTheCancellation(): void + { + $observed = null; + $scheduler = new Scheduler(); + $body = $scheduler->spawn(static function () use (&$observed): void { + $victim = Coroutine::spawn(static fn(): mixed => \Fiber::suspend()); + Coroutine::spawn(static function () use ($victim, &$observed): void { + try { + \Fiber::suspend(); + } catch (CancelledException) { + try { + $observed = $victim->await(); + } catch (CancelledException $e) { + $observed = $e; + } + } + }); + + \Fiber::suspend(); + throw new \RuntimeException('body died'); + }); + + $scheduler->drive($body); + + Assert::instanceOf($observed, CancelledException::class); + } + public function unfinishedCoroutinesAreDrivenAfterTheBodyReturns(): void { $log = []; diff --git a/skills/testo-fiber/SKILL.md b/skills/testo-fiber/SKILL.md index e0f5033b..61b436aa 100644 --- a/skills/testo-fiber/SKILL.md +++ b/skills/testo-fiber/SKILL.md @@ -91,7 +91,7 @@ public function pingPong(): void Rules (verified against `plugin/fiber/src/Coroutine.php`): - `spawn()` needs an active scope — outside `#[RunInFiber]` it throws a `LogicException`. Assertions, messages **and coverage** inside a coroutine are attributed to the test that spawned it: the scope runs inside both the scoped-state guards and the test's coverage window. -- **The scope is structured**: the test is not finished until every coroutine it spawned is. Coroutines still pending when the body returns keep being driven; if the body *fails*, they are cancelled — a `Testo\Fiber\Exception\CancelledException` is thrown into each pending fiber (its `finally` blocks run; don't swallow it). +- **The scope is structured**: the test is not finished until every coroutine it spawned is. Coroutines still pending when the body returns keep being driven; if the body *fails*, they are cancelled — a `Testo\Fiber\Exception\CancelledException` is thrown into each pending fiber (its `finally` blocks run; don't swallow it). Awaiting a cancelled coroutine rethrows the `CancelledException` (unwrapped — it is a control signal, not a coroutine failure). - **Coroutine failures always arrive wrapped in `Testo\Fiber\Exception\CompositeException`** — even a single one — whether rethrown by `await()` / `concurrently()` or reported at scope close for a coroutine nobody awaited (that marks the test `Error`). The body's own throw stays unwrapped, so `#[ExpectException]` on it works as usual; expect `CompositeException` when the throw comes from a coroutine. - An await cycle is detected and broken with a `Testo\Fiber\Exception\DeadlockException` raised at the first doomed `await()` — even a cycle spanning several tests' scopes (handles shared under a class-level `#[RunInFiber]`). A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected. - `concurrently()` waits for *all* its coroutines even after one fails, then bundles every failure into one composite. From e8aa405af54623993ebc3477d589e986d4ebfc3a Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Thu, 6 Aug 2026 19:56:00 +0400 Subject: [PATCH 7/9] fix(fiber): key concurrently() failures like the arguments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit concurrently() returned results keyed by the argument keys but bundled failures keyed by the scheduler's internal task ids — with named arguments the error's origin was untraceable, and with positional ones the ids looked like argument indices while being off by one (the test body is task #0). Each await() composite wraps exactly one task's error; concurrently() now unwraps it and re-keys it by the argument, making $errors symmetric to the results. CompositeException accepts string keys (named arguments) and prints them as-is in the message, keeping the #N form for int keys. Assisted-By: Claude Fable 5 --- plugin/fiber/src/Coroutine.php | 10 +++++-- .../src/Exception/CompositeException.php | 14 ++++++--- plugin/fiber/tests/Unit/CoroutineTest.php | 29 ++++++++++++++++++- skills/testo-fiber/SKILL.md | 2 +- 4 files changed, 46 insertions(+), 9 deletions(-) diff --git a/plugin/fiber/src/Coroutine.php b/plugin/fiber/src/Coroutine.php index 64177da1..79a80d9c 100644 --- a/plugin/fiber/src/Coroutine.php +++ b/plugin/fiber/src/Coroutine.php @@ -68,11 +68,13 @@ public static function spawn(\Closure|\Fiber $body): self * Sugar over {@see spawn()} + {@see await()}: schedules everything into the running scope, parks * the caller until every coroutine finished, and returns the results keyed like the arguments * (named arguments give string keys). Failures are collected until all coroutines settle, then - * bundled into one {@see CompositeException}. + * bundled into one {@see CompositeException} — its errors keyed like the arguments too, + * symmetric to the results. A coroutine that itself died with a `CompositeException` appears + * nested: that whole exception sits under the argument's key, its own structure intact. * * @return array Results keyed like the arguments. * - * @throws CompositeException When any of the coroutines threw. + * @throws CompositeException When any of the coroutines threw — errors keyed like the arguments. * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]`. */ public static function concurrently(\Closure|\Fiber ...$bodies): array @@ -84,7 +86,9 @@ public static function concurrently(\Closure|\Fiber ...$bodies): array try { $results[$key] = $handle->await(); } catch (CompositeException $e) { - $errors += $e->errors; + // await() wraps exactly one task's error — unwrap and re-key it by the argument, + // so callers never see the scheduler's internal task ids. + $errors[$key] = $e->errors[\array_key_first($e->errors)]; } } diff --git a/plugin/fiber/src/Exception/CompositeException.php b/plugin/fiber/src/Exception/CompositeException.php index d17ef954..8844e11b 100644 --- a/plugin/fiber/src/Exception/CompositeException.php +++ b/plugin/fiber/src/Exception/CompositeException.php @@ -19,21 +19,27 @@ final class CompositeException extends \RuntimeException { /** - * The collected throwables, keyed by the fiber (task) index that raised each one. + * The collected throwables, keyed by whatever names each fiber to the producer: the task id for + * scope/batch failures, or the argument key for {@see \Testo\Fiber\Coroutine::concurrently()}. * - * @var non-empty-array + * @var non-empty-array */ public readonly array $errors; /** - * @param non-empty-array $errors + * @param non-empty-array $errors */ public function __construct(array $errors) { $this->errors = $errors; $lines = \array_map( - static fn(int $i, \Throwable $e): string => \sprintf(' #%d %s: %s', $i, $e::class, $e->getMessage()), + static fn(int|string $key, \Throwable $e): string => \sprintf( + ' %s %s: %s', + \is_int($key) ? "#$key" : $key, + $e::class, + $e->getMessage(), + ), \array_keys($errors), \array_values($errors), ); diff --git a/plugin/fiber/tests/Unit/CoroutineTest.php b/plugin/fiber/tests/Unit/CoroutineTest.php index edc5f996..c42113e4 100644 --- a/plugin/fiber/tests/Unit/CoroutineTest.php +++ b/plugin/fiber/tests/Unit/CoroutineTest.php @@ -146,10 +146,37 @@ static function () use (&$log, $second): void { }); Assert::instanceOf($caught, CompositeException::class); - Assert::same(\array_values($caught->errors), [$first, $second]); + # Errors are keyed like the arguments, symmetric to the results. + Assert::same($caught->errors, [0 => $first, 1 => $second]); Assert::same($log, ['slow ran to its end']); } + public function concurrentlyKeysFailuresLikeTheArguments(): void + { + $pullError = new \RuntimeException('pull broke'); + $pushError = new \LogicException('push broke'); + + $caught = $this->scope(static function () use ($pullError, $pushError): mixed { + try { + Coroutine::concurrently( + pull: static fn() => throw $pullError, + ok: static fn(): string => 'fine', + push: static fn() => throw $pushError, + ); + } catch (CompositeException $e) { + return $e; + } + + return null; + }); + + Assert::instanceOf($caught, CompositeException::class); + Assert::same($caught->errors, ['pull' => $pullError, 'push' => $pushError]); + Assert::same($caught->getPrevious(), $pullError); + # String keys name the fiber in the message as-is; int keys keep the #N form. + Assert::string($caught->getMessage())->contains('pull'); + } + public function awaitCycleIsBrokenAsADeadlock(): void { $caught = $this->scope(static function (): mixed { diff --git a/skills/testo-fiber/SKILL.md b/skills/testo-fiber/SKILL.md index 61b436aa..18a88630 100644 --- a/skills/testo-fiber/SKILL.md +++ b/skills/testo-fiber/SKILL.md @@ -94,7 +94,7 @@ Rules (verified against `plugin/fiber/src/Coroutine.php`): - **The scope is structured**: the test is not finished until every coroutine it spawned is. Coroutines still pending when the body returns keep being driven; if the body *fails*, they are cancelled — a `Testo\Fiber\Exception\CancelledException` is thrown into each pending fiber (its `finally` blocks run; don't swallow it). Awaiting a cancelled coroutine rethrows the `CancelledException` (unwrapped — it is a control signal, not a coroutine failure). - **Coroutine failures always arrive wrapped in `Testo\Fiber\Exception\CompositeException`** — even a single one — whether rethrown by `await()` / `concurrently()` or reported at scope close for a coroutine nobody awaited (that marks the test `Error`). The body's own throw stays unwrapped, so `#[ExpectException]` on it works as usual; expect `CompositeException` when the throw comes from a coroutine. - An await cycle is detected and broken with a `Testo\Fiber\Exception\DeadlockException` raised at the first doomed `await()` — even a cycle spanning several tests' scopes (handles shared under a class-level `#[RunInFiber]`). A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected. -- `concurrently()` waits for *all* its coroutines even after one fails, then bundles every failure into one composite. +- `concurrently()` waits for *all* its coroutines even after one fails, then bundles every failure into one composite whose `$errors` are keyed like the arguments — symmetric to the results, so a named argument's failure is found under its name. ## Pitfalls From 5b38214407078434393b75f326a2784230df054f Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Thu, 6 Aug 2026 23:31:06 +0400 Subject: [PATCH 8/9] =?UTF-8?q?refactor:=20address=20review=20nits=20?= =?UTF-8?q?=E2=80=94=20assert=20argument=20order,=20Status::isFailure()?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CacheTest mixed expected-first and actual-first Assert::same() calls; the facade's signature is same($actual, $expected), so the inverted calls produced swapped "expected X, got Y" messages on failure. All calls now pass the actual value first. CoroutineScopeInterceptor spelled out Failed/Error where an isFailure() call away the same interceptor already uses the enum's own check — one place to update if a failure-like status is ever added. Assisted-By: Claude Fable 5 --- .../src/Internal/CoroutineScopeInterceptor.php | 3 +-- tests/Core/Pipeline/CacheTest.php | 14 +++++++------- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php b/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php index 2ce930c6..cae7b3fa 100644 --- a/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php +++ b/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php @@ -91,9 +91,8 @@ public function runTest(TestInfo $info, callable $next): TestResult // errors so nothing is dropped, and keep the harsher of the two statuses. $result->failure === null or $errors = [$body->id => $result->failure] + $errors; - $failed = $result->status === Status::Failed || $result->status === Status::Error; $result = $result - ->with(status: $failed ? $result->status : Status::Error) + ->with(status: $result->status->isFailure() ? $result->status : Status::Error) ->withFailure(new CompositeException($errors)); } diff --git a/tests/Core/Pipeline/CacheTest.php b/tests/Core/Pipeline/CacheTest.php index 4b92e4fb..8971f885 100644 --- a/tests/Core/Pipeline/CacheTest.php +++ b/tests/Core/Pipeline/CacheTest.php @@ -24,7 +24,7 @@ public function resolveAliasesWithFallbackInterceptorAttribute(): void { $result = Cache::resolveAliases(AttributeWithFallback::class); - Assert::same([MockInterceptor::class], $result); + Assert::same($result, [MockInterceptor::class]); } /** @@ -34,7 +34,7 @@ public function resolveAliasesCollectsRepeatedFallbacks(): void { $result = Cache::resolveAliases(AttributeWithSeveralFallbacks::class); - Assert::same([MockInterceptor::class, SecondMockInterceptor::class], $result); + Assert::same($result, [MockInterceptor::class, SecondMockInterceptor::class]); } /** @@ -61,11 +61,11 @@ public function resolveAliasesMemoisesResultInMap(): void $map->setValue(null, []); $result = Cache::resolveAliases(AttributeWithFallbackForCache::class); - Assert::same([MockInterceptor::class], $result); + Assert::same($result, [MockInterceptor::class]); $stored = $map->getValue(); Assert::true(\array_key_exists(AttributeWithFallbackForCache::class, $stored)); - Assert::same([MockInterceptor::class], $stored[AttributeWithFallbackForCache::class]); + Assert::same($stored[AttributeWithFallbackForCache::class], [MockInterceptor::class]); } finally { $map->setValue(null, $orig); } @@ -84,14 +84,14 @@ public function resolveAliasesWalksCachedParentInMap(): void $map->setValue(null, []); $parent = Cache::resolveAliases(ParentAttributeForMapWalk::class); - Assert::same([MockInterceptor::class], $parent); + Assert::same($parent, [MockInterceptor::class]); $stored = $map->getValue(); Assert::true(\array_key_exists(ParentAttributeForMapWalk::class, $stored)); Assert::false(\array_key_exists(ChildAttributeForMapWalk::class, $stored)); $child = Cache::resolveAliases(ChildAttributeForMapWalk::class); - Assert::same([MockInterceptor::class], $child); + Assert::same($child, [MockInterceptor::class]); } finally { $map->setValue(null, $orig); } @@ -131,7 +131,7 @@ public function resolveAliasesWalksParentClassHierarchy(): void { $result = Cache::resolveAliases(ChildAttributeOfFallback::class); - Assert::same([MockInterceptor::class], $result); + Assert::same($result, [MockInterceptor::class]); } private static function mapProperty(): \ReflectionProperty From b0bb22acbef50e1ccca3fd00ed2250165deafd93 Mon Sep 17 00:00:00 2001 From: roxblnfk Date: Sat, 8 Aug 2026 15:54:23 +0400 Subject: [PATCH 9/9] refactor(fiber): dedupe docblocks, drop dead null-guards, merge overlapping tests Assisted-By: Claude Fable 5 --- plugin/fiber/src/Coroutine.php | 13 ++---- .../Internal/CoroutineScopeInterceptor.php | 23 ++++------ plugin/fiber/src/Internal/Scheduler.php | 33 ++++++------- plugin/fiber/tests/Feature/StatusTest.php | 1 - plugin/fiber/tests/Unit/CoroutineTest.php | 46 +++++-------------- tests/Sandbox/Self/AsyncTest.php | 8 ---- 6 files changed, 38 insertions(+), 86 deletions(-) diff --git a/plugin/fiber/src/Coroutine.php b/plugin/fiber/src/Coroutine.php index 79a80d9c..73553826 100644 --- a/plugin/fiber/src/Coroutine.php +++ b/plugin/fiber/src/Coroutine.php @@ -38,10 +38,10 @@ * * @api */ -final class Coroutine +final readonly class Coroutine { private function __construct( - private readonly Task $task, + private Task $task, ) {} /** @@ -108,12 +108,9 @@ public function isFinished(): bool /** * Park the calling coroutine until this one finishes, and return its result. * - * Other coroutines — and, through the scope's relay, the case's other tests — keep running while - * the caller is parked. A throwable raised by the awaited coroutine is rethrown here wrapped in - * a {@see CompositeException}; rethrowing marks the failure as observed, so the scope will not - * report it again. Awaiting a cancelled coroutine rethrows the cancellation — unwrapped, it is - * the scope's control signal rather than a failure of the coroutine — so a `finally` unwinding - * on the same cancellation cannot mistake a torn-down sibling for one that returned `null`. + * Other coroutines keep running while the caller is parked. Rethrowing a failure here marks it + * as observed, so the scope will not report it again. A cancellation is rethrown unwrapped — it + * is the scope's control signal, not a failure of the coroutine. * * @throws CompositeException When the awaited coroutine threw. * @throws CancelledException When the awaited coroutine was cancelled with its scope. diff --git a/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php b/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php index cae7b3fa..4f95edc5 100644 --- a/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php +++ b/plugin/fiber/src/Internal/CoroutineScopeInterceptor.php @@ -22,19 +22,12 @@ * relays control upward between rounds, so its coroutines keep interleaving with whatever schedule * drives the test — the case batch of a class-level `#[RunInFiber]`, or the single method-level fiber. * - * Sits at {@see InterceptorOptions::ORDER_ASYNC_COROUTINE} — the innermost position, *inside* both the - * fiber-aware scoped-state guards (assertion collector, messenger scope) and the coverage window, - * unlike {@see RunInFiberInterceptor} which wraps them. That placement is load-bearing in both - * directions: coroutines are only ever resumed from this scope's drive frame, so each one runs with - * its test's state swapped in and inside its coverage window — assertions, messages and executed - * lines are all attributed to the test that spawned it. Were the scope outer to - * {@see InterceptorOptions::ORDER_COVERAGE}, the collector's trampoline would close the window every - * time the body relayed a suspension to us, and everything the coroutines then ran would be measured - * for nobody. - * - * Unlike an event loop at the same order (`testo/bridge-revolt`), this scope drives plain fibers Testo - * owns and relays suspensions outward, so the trampoline above keeps working: what it must not have - * above it is a driver that resumes fibers past its wrapper. + * Sits at {@see InterceptorOptions::ORDER_ASYNC_COROUTINE} (see the placement contract there) — the + * innermost position, *inside* both the fiber-aware scoped-state guards (assertion collector, + * messenger scope) and the coverage window, unlike {@see RunInFiberInterceptor} which wraps them. + * Coroutines are only ever resumed from this scope's drive frame, so each one runs with its test's + * state swapped in and inside its coverage window — assertions, messages and executed lines are all + * attributed to the test that spawned it. * * Coroutine failures nobody awaited fail the test: they are bundled into a {@see CompositeException} * (always, even a single one) and attached to the result as its failure with {@see Status::Error}. @@ -66,8 +59,8 @@ public function runTest(TestInfo $info, callable $next): TestResult $scheduler->drive($body, static fn(Task $task): bool => $task->result instanceof TestResult && $task->result->status->isFailure()); - // The pipeline below captures test throwables as results, so a body error is unexpected - // infrastructure breakage — let it abort the pipeline. + // An error that did escape the body fiber is unexpected infrastructure breakage — let it + // abort the pipeline. $body->error === null or throw $body->error; /** @var TestResult $result */ diff --git a/plugin/fiber/src/Internal/Scheduler.php b/plugin/fiber/src/Internal/Scheduler.php index 3c091774..c8c5da36 100644 --- a/plugin/fiber/src/Internal/Scheduler.php +++ b/plugin/fiber/src/Internal/Scheduler.php @@ -101,18 +101,14 @@ public function spawn(\Closure|\Fiber $body): Task * inside a fiber, control is relayed to the parent scheduler. * * With `$primary` set (a test's coroutine scope, where `$primary` is the test body), a primary - * failure cancels the remaining tasks instead of driving them further: a {@see CancelledException} - * is thrown into every pending fiber so its `finally` blocks run; a throwable escaping that unwind - * (other than the cancellation itself) is recorded as the task's error. A failure is an error that - * escaped the primary fiber, or `$primaryFailed` returning `true` for the settled task — the - * caller's chance to recognize failures its pipeline captured into the task's result. + * failure cancels the remaining tasks instead of driving them further ({@see cancelPending()}). + * A failure is an error that escaped the primary fiber, or `$primaryFailed` returning `true` for + * the settled task — the caller's chance to recognize failures its pipeline captured into the + * task's result. An await cycle ({@see deadlocked()}) is broken by throwing a + * {@see DeadlockException} into the first task the cycle dooms; the failure then cascades to its + * awaiters, so the stack trace points at the guilty `await()`. * * @param null|\Closure(Task): bool $primaryFailed - * - * An await cycle — even one spanning several schedulers' tasks — is broken by throwing a - * {@see DeadlockException} into the first task the cycle dooms; the failure then cascades to its - * awaiters, so the deadlock surfaces as an ordinary task error with a stack trace pointing at - * the guilty `await()`. */ public function drive(?Task $primary = null, ?\Closure $primaryFailed = null): void { @@ -134,20 +130,16 @@ public function drive(?Task $primary = null, ?\Closure $primaryFailed = null): v } if ($ready === []) { - // Every unfinished task is parked in an await. A task whose await chain runs - // into a cycle — even one spanning other schedulers — can never be unparked by - // any schedule; a chain that ends outside a cycle may still be unparked by the - // outer schedule, so with only those left, relay and retry. + // Every unfinished task is parked in an await. A chain that ends outside a + // cycle may still be unparked by the outer schedule — relay and retry. $doomed = $this->deadlocked($parked); if ($doomed === [] && \Fiber::getCurrent() !== null) { $this->relay($prev); continue; } - // Break the cycle: the first doomed task gets the deadlock at its await point - // and unwinds; its awaiters unpark and the failure cascades through the cycle. - // With no fiber to relay from, tasks parked on foreign tasks are stuck the same - // way — nobody else will ever drive those. + // With no fiber to relay from, tasks parked on foreign tasks are just as stuck + // as a cycle — nobody else will ever drive those. $stuck = $doomed === [] ? $parked : $doomed; $this->throwInto($this->tasks[$stuck[0]], new DeadlockException($this->describeDeadlock($stuck))); continue; @@ -344,11 +336,12 @@ private function describeDeadlock(array $parked): string $lines = []; foreach ($parked as $id) { $target = $this->tasks[$id]->awaiting; + \assert($target !== null); $lines[] = \sprintf( '#%d awaits %s#%d', $id, - $target === null || $target->scheduler === $this ? '' : "another scope's ", - $target?->id ?? -1, + $target->scheduler === $this ? '' : "another scope's ", + $target->id, ); } diff --git a/plugin/fiber/tests/Feature/StatusTest.php b/plugin/fiber/tests/Feature/StatusTest.php index 5dda7913..0738f491 100644 --- a/plugin/fiber/tests/Feature/StatusTest.php +++ b/plugin/fiber/tests/Feature/StatusTest.php @@ -55,7 +55,6 @@ public function unawaitedCoroutineFailureErrorsTheTest(): void $result = TestRunner::runTest([FiberScenarios::class, 'unawaitedCoroutineFailure']); Assert::same($result->status, Status::Error); - # Coroutine failures always arrive as a composite, even a single one. Assert::instanceOf($result->failure, CompositeException::class); Assert::instanceOf($result->failure->getPrevious(), \RuntimeException::class); } diff --git a/plugin/fiber/tests/Unit/CoroutineTest.php b/plugin/fiber/tests/Unit/CoroutineTest.php index c42113e4..781eb5c0 100644 --- a/plugin/fiber/tests/Unit/CoroutineTest.php +++ b/plugin/fiber/tests/Unit/CoroutineTest.php @@ -66,7 +66,6 @@ public function awaitRethrowsWrappedInAComposite(): void return null; }); - # Even a single coroutine failure arrives as a composite, so handling code is uniform. Assert::instanceOf($caught, CompositeException::class); Assert::same(\array_values($caught->errors), [$boom]); Assert::same($caught->getPrevious(), $boom); @@ -122,20 +121,21 @@ public function concurrentlyReturnsResultsKeyedLikeTheArguments(): void Assert::same($results, ['first' => 'one', 'second' => 'two']); } - public function concurrentlyBundlesEveryFailureAfterAllSettled(): void + public function concurrentlyBundlesEveryFailureKeyedLikeTheArguments(): void { - $first = new \RuntimeException('first'); - $second = new \LogicException('second'); + $first = new \RuntimeException('first broke'); + $pushError = new \LogicException('push broke'); $log = []; - $caught = $this->scope(static function () use ($first, $second, &$log): mixed { + $caught = $this->scope(static function () use ($first, $pushError, &$log): mixed { try { Coroutine::concurrently( static fn() => throw $first, - static function () use (&$log, $second): void { + ok: static fn(): string => 'fine', + push: static function () use (&$log, $pushError): void { \Fiber::suspend(); $log[] = 'slow ran to its end'; - throw $second; + throw $pushError; }, ); } catch (CompositeException $e) { @@ -146,35 +146,13 @@ static function () use (&$log, $second): void { }); Assert::instanceOf($caught, CompositeException::class); - # Errors are keyed like the arguments, symmetric to the results. - Assert::same($caught->errors, [0 => $first, 1 => $second]); + # Every coroutine settled before the bundle was thrown; errors are keyed like the arguments. + Assert::same($caught->errors, [0 => $first, 'push' => $pushError]); + Assert::same($caught->getPrevious(), $first); Assert::same($log, ['slow ran to its end']); - } - - public function concurrentlyKeysFailuresLikeTheArguments(): void - { - $pullError = new \RuntimeException('pull broke'); - $pushError = new \LogicException('push broke'); - - $caught = $this->scope(static function () use ($pullError, $pushError): mixed { - try { - Coroutine::concurrently( - pull: static fn() => throw $pullError, - ok: static fn(): string => 'fine', - push: static fn() => throw $pushError, - ); - } catch (CompositeException $e) { - return $e; - } - - return null; - }); - - Assert::instanceOf($caught, CompositeException::class); - Assert::same($caught->errors, ['pull' => $pullError, 'push' => $pushError]); - Assert::same($caught->getPrevious(), $pullError); # String keys name the fiber in the message as-is; int keys keep the #N form. - Assert::string($caught->getMessage())->contains('pull'); + Assert::string($caught->getMessage())->contains('push'); + Assert::string($caught->getMessage())->contains('#0'); } public function awaitCycleIsBrokenAsADeadlock(): void diff --git a/tests/Sandbox/Self/AsyncTest.php b/tests/Sandbox/Self/AsyncTest.php index 41dde321..b1440b20 100644 --- a/tests/Sandbox/Self/AsyncTest.php +++ b/tests/Sandbox/Self/AsyncTest.php @@ -75,14 +75,6 @@ public function slowDataSets(string $label): void #[DataSet(['fast-set-d'])] #[DataSet(['fast-set-e'])] #[DataSet(['fast-set-f'])] - #[DataSet(['fast-set-g'])] - #[DataSet(['fast-set-h'])] - #[DataSet(['fast-set-i'])] - #[DataSet(['fast-set-j'])] - #[DataSet(['fast-set-k'])] - #[DataSet(['fast-set-l'])] - #[DataSet(['fast-set-m'])] - #[DataSet(['fast-set-n'])] public function fastDataSets(string $label): void { self::workThenYield('quick', $label);