From 6b95909d0cab88d30eee84b5089b271c73e6bddc Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 09:33:21 +0200 Subject: [PATCH 1/7] feat: add amphp/amp v3 fiber-based promise adapter Add a fiber-based AmpFutureAdapter and update DataLoader to dispatch, await, and resolve amphp/amp v3 futures. Preserve keyed and positional batch result handling, propagate throwables, and keep synchronous adapters unchanged. --- composer.json | 2 + .../src/Adapter/AmpFutureAdapter.php | 121 ++++++++++++++++++ src/DataLoader.php | 38 +++++- tests/AmpFutureDataLoaderTest.php | 42 ++++++ 4 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 lib/promise-adapter/src/Adapter/AmpFutureAdapter.php create mode 100644 tests/AmpFutureDataLoaderTest.php diff --git a/composer.json b/composer.json index d34811d..b52eae0 100644 --- a/composer.json +++ b/composer.json @@ -26,12 +26,14 @@ "php": "^8.2" }, "require-dev": { + "amphp/amp": "^3.1", "guzzlehttp/promises": "^1.5.0 || ^2.0.0", "phpunit/phpunit": "^10.3", "react/promise": "^2.8 || ^3.0", "webonyx/graphql-php": "^15.0" }, "suggest": { + "amphp/amp": "To use with amphp/amp v3 futures (fiber-based)", "guzzlehttp/promises": "To use with Guzzle promise", "react/promise": "To use with ReactPhp promise", "webonyx/graphql-php": "To use with Webonyx GraphQL native promise" diff --git a/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php b/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php new file mode 100644 index 0000000..ad24b0a --- /dev/null +++ b/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php @@ -0,0 +1,121 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Overblog\PromiseAdapter\Adapter; + +use Amp\DeferredFuture; +use Amp\Future; +use Overblog\PromiseAdapter\PromiseAdapterInterface; +use function Amp\async; +use function Amp\Future\await; + +/** + * Promise adapter backed by amphp/amp v3 fiber-based futures. + * + * Unlike the Guzzle/React adapters, amp v3 futures do not expose a `then()` + * method and only settle once the event loop advances (inside a fiber). + * The DataLoader core is patched to cooperate with this model via `Amp\async` + * and `Future::await()` instead of `->then()`. + * + * @implements PromiseAdapterInterface + */ +class AmpFutureAdapter implements PromiseAdapterInterface +{ + /** + * @return Future + */ + public function create(&$resolve = null, &$reject = null, ?callable $canceller = null): Future + { + $deferred = new DeferredFuture(); + + $resolve = static function ($value) use ($deferred): void { + if ($deferred->isComplete()) { + return; + } + + $deferred->complete($value); + }; + $reject = static function (\Throwable $reason) use ($deferred): void { + if ($deferred->isComplete()) { + return; + } + + $deferred->error($reason); + }; + + return $deferred->getFuture(); + } + + /** + * @return Future + */ + public function createFulfilled($promiseOrValue = null): Future + { + if ($promiseOrValue instanceof Future) { + return $promiseOrValue; + } + + return Future::complete($promiseOrValue); + } + + /** + * @return Future + */ + public function createRejected($reason): Future + { + return Future::error($reason); + } + + /** + * @return Future + */ + public function createAll($promisesOrValues): Future + { + $futures = []; + foreach ($promisesOrValues as $key => $value) { + $futures[$key] = $value instanceof Future ? $value : Future::complete($value); + } + + return async(static fn () => await($futures)); + } + + public function isPromise($value, $strict = false): bool + { + return $value instanceof Future; + } + + public function await($promise = null, $unwrap = true): mixed + { + if (null === $promise) { + return null; + } + + if (!$promise instanceof Future) { + throw new \InvalidArgumentException(sprintf('The "%s" method must be called with an amp Future.', __METHOD__)); + } + + try { + return $promise->await(); + } catch (\Throwable $reason) { + if (!$unwrap) { + return $reason; + } + + throw $reason; + } + } + + public function cancel($promise): void + { + // amp v3 futures are cancelled through a Cancellation token passed at + // creation time; there is no post-hoc cancel handle here. + } +} diff --git a/src/DataLoader.php b/src/DataLoader.php index f2236e2..e681278 100644 --- a/src/DataLoader.php +++ b/src/DataLoader.php @@ -122,6 +122,17 @@ static function () { if (!$shouldBatch) { // Otherwise dispatch the (queue of one) immediately. $this->dispatchQueue(); + } elseif ($this->getPromiseAdapter() instanceof \Overblog\PromiseAdapter\Adapter\AmpFutureAdapter) { + // On the fiber-based adapter there is no synchronous drain loop + // (SyncPromiseAdapter::onWait). Instead, schedule the batch to + // dispatch on the next event-loop tick, once the current call + // stack has finished enqueuing all loads for this frame. This + // mirrors the JS reference implementation's process.nextTick. + \Revolt\EventLoop::queue(function (): void { + if ($this->needProcess()) { + $this->dispatchQueue(); + } + }); } } @@ -265,6 +276,16 @@ public static function await($promise = null, $unwrap = true) return null; } + // amp v3 futures are fiber-based and expose no `then()`; defer to the + // adapter's `await()` which suspends the current fiber until settled. + if ($promise instanceof \Amp\Future) { + if ([] === self::$instances) { + throw new \RuntimeException('Found no active DataLoader instance.'); + } + + return self::$instances[0]->getPromiseAdapter()->await($promise, $unwrap); + } + if (is_callable([$promise, 'then'])) { $isPromiseCompleted = false; $resolvedValue = null; @@ -322,7 +343,7 @@ function ($reason) use (&$isPromiseCompleted, &$rejectedReason) { private static function awaitInstances() { - if (empty(self::$activeInstances)) { + if ([] === self::$activeInstances) { return; } @@ -409,6 +430,21 @@ private function dispatchQueueBatch(array $queue) return; } + // amp v3 futures are fiber-based and expose no `then()`; drive the + // fan-out inside a fiber via `Amp\async` instead. + if ($batchPromise instanceof \Amp\Future) { + \Amp\async(function () use ($batchPromise, $queue) { + try { + $values = $batchPromise->await(); + $this->resolveDispatchedBatch($values, $queue); + } catch (\Throwable $error) { + $this->failedDispatch($queue, $error); + } + }); + + return; + } + // Assert the expected response from batchLoadFn if (!$batchPromise || !is_callable([$batchPromise, 'then'])) { $this->failedDispatch($queue, new \RuntimeException( diff --git a/tests/AmpFutureDataLoaderTest.php b/tests/AmpFutureDataLoaderTest.php new file mode 100644 index 0000000..7266fea --- /dev/null +++ b/tests/AmpFutureDataLoaderTest.php @@ -0,0 +1,42 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Overblog\DataLoader\Test; + +use Amp\Future; +use Error; +use Overblog\DataLoader\DataLoader; +use Overblog\PromiseAdapter\Adapter\AmpFutureAdapter; + +use function Amp\async; + +class AmpFutureDataLoaderTest extends TestCase +{ + public function testRejectsEveryQueuedLoadWhenBatchFutureFailsWithAnError() + { + $loader = new DataLoader( + static fn (array $keys): Future => Future::error(new Error('batch load failed')), + new AmpFutureAdapter(), + ); + + $first = $loader->load('first'); + $second = $loader->load('second'); + + foreach ([$first, $second] as $future) { + try { + async(static fn () => $future->await())->await(); + self::fail('Expected the queued load to fail.'); + } catch (Error $error) { + self::assertSame('batch load failed', $error->getMessage()); + } + } + } +} From 5974cedfc18bff68b8d506d9189e60fe1762bed6 Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 09:45:43 +0200 Subject: [PATCH 2/7] fix: await registered fiber futures --- src/DataLoader.php | 59 ++++++++++++++----------------- tests/AmpFutureDataLoaderTest.php | 16 +++++++++ 2 files changed, 43 insertions(+), 32 deletions(-) diff --git a/src/DataLoader.php b/src/DataLoader.php index e681278..642651c 100644 --- a/src/DataLoader.php +++ b/src/DataLoader.php @@ -276,46 +276,41 @@ public static function await($promise = null, $unwrap = true) return null; } - // amp v3 futures are fiber-based and expose no `then()`; defer to the - // adapter's `await()` which suspends the current fiber until settled. - if ($promise instanceof \Amp\Future) { - if ([] === self::$instances) { - throw new \RuntimeException('Found no active DataLoader instance.'); + if (!is_callable([$promise, 'then'])) { + if (is_object($promise) && null !== self::$promiseAdapters && isset(self::$promiseAdapters[$promise])) { + return self::$promiseAdapters[$promise]->await($promise, $unwrap); } - return self::$instances[0]->getPromiseAdapter()->await($promise, $unwrap); + throw new \InvalidArgumentException(sprintf('The "%s" method must be called with a Promise ("then" method).', __METHOD__)); } - if (is_callable([$promise, 'then'])) { - $isPromiseCompleted = false; - $resolvedValue = null; - $rejectedReason = null; - - $promise->then( - function ($value) use (&$isPromiseCompleted, &$resolvedValue) { - $isPromiseCompleted = true; - $resolvedValue = $value; - }, - function ($reason) use (&$isPromiseCompleted, &$rejectedReason) { - $isPromiseCompleted = true; - $rejectedReason = $reason; - } - ); + $isPromiseCompleted = false; + $resolvedValue = null; + $rejectedReason = null; - //Promise is completed? - if ($isPromiseCompleted) { - // rejected ? - if ($rejectedReason instanceof \Throwable) { - if (!$unwrap) { - return $rejectedReason; - } - throw $rejectedReason; + $promise->then( + function ($value) use (&$isPromiseCompleted, &$resolvedValue) { + $isPromiseCompleted = true; + $resolvedValue = $value; + }, + function ($reason) use (&$isPromiseCompleted, &$rejectedReason) { + $isPromiseCompleted = true; + $rejectedReason = $reason; + } + ); + + //Promise is completed? + if ($isPromiseCompleted) { + // rejected ? + if ($rejectedReason instanceof \Throwable) { + if (!$unwrap) { + return $rejectedReason; } - return $resolvedValue; + throw $rejectedReason; } - } else { - throw new \InvalidArgumentException(sprintf('The "%s" method must be called with a Promise ("then" method).', __METHOD__)); + + return $resolvedValue; } if (null !== self::$promiseAdapters && isset(self::$promiseAdapters[$promise])) { diff --git a/tests/AmpFutureDataLoaderTest.php b/tests/AmpFutureDataLoaderTest.php index 7266fea..cc15277 100644 --- a/tests/AmpFutureDataLoaderTest.php +++ b/tests/AmpFutureDataLoaderTest.php @@ -15,11 +15,27 @@ use Error; use Overblog\DataLoader\DataLoader; use Overblog\PromiseAdapter\Adapter\AmpFutureAdapter; +use Overblog\PromiseAdapter\PromiseAdapterInterface; use function Amp\async; class AmpFutureDataLoaderTest extends TestCase { + protected function createPromiseAdapter(): PromiseAdapterInterface + { + return new AmpFutureAdapter(); + } + + public function testAwaitUsesTheRegisteredAmpFutureAdapter() + { + $loader = new DataLoader( + static fn (array $keys): Future => Future::complete(['value', 'value']), + new AmpFutureAdapter(), + ); + + self::assertSame(['value', 'value'], DataLoader::await($loader->loadMany(['first', 'second']))); + } + public function testRejectsEveryQueuedLoadWhenBatchFutureFailsWithAnError() { $loader = new DataLoader( From 73f5cf7e91e2a03677903079fcebb089696930d3 Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 09:47:56 +0200 Subject: [PATCH 3/7] fix: resolve dispatched fiber batches --- src/DataLoader.php | 65 +++++++++++++++++++++++++++------------------- 1 file changed, 38 insertions(+), 27 deletions(-) diff --git a/src/DataLoader.php b/src/DataLoader.php index 642651c..c028899 100644 --- a/src/DataLoader.php +++ b/src/DataLoader.php @@ -453,39 +453,50 @@ private function dispatchQueueBatch(array $queue) // Await the resolution of the call to batchLoadFn. $batchPromise->then( - function ($values) use ($keys, $queue) { - // Assert the expected resolution from batchLoadFn. - if (!is_array($values) && !$values instanceof \Traversable) { - throw new \RuntimeException( - 'DataLoader must be constructed with a function which accepts ' . - 'Array and returns Promise>, but the function did ' . - sprintf('not return a Promise of an Array: %s.', gettype($values)) - ); - } - if (count($values) !== count($keys)) { - throw new \RuntimeException( - 'DataLoader must be constructed with a function which accepts ' . - 'Array and returns Promise>, but the function did ' . - 'not return a Promise of an Array of the same length as the Array of keys.' - ); - } - - // Step through the values, resolving or rejecting each Promise in the - // loaded queue. - foreach ($queue as $index => $data) { - $value = $values[$index]; - if ($value instanceof \Throwable) { - $data['reject']($value); - } else { - $data['resolve']($value); - } - }; + function ($values) use ($queue) { + $this->resolveDispatchedBatch($values, $queue); } )->then(null, function ($error) use ($queue) { $this->failedDispatch($queue, $error); }); } + /** + * Fan a resolved batch result out to the individual queued promises. + * + * @param mixed $values + * @param array $queue + */ + private function resolveDispatchedBatch($values, $queue) + { + // Assert the expected response from batchLoadFn. + if (!is_array($values) && !$values instanceof \Traversable) { + throw new \RuntimeException( + 'DataLoader must be constructed with a function which accepts ' . + 'Array and returns Promise>, but the function did ' . + sprintf('not return a Promise of an Array: %s.', gettype($values)) + ); + } + if (count($values) !== count($queue)) { + throw new \RuntimeException( + 'DataLoader must be constructed with a function which accepts ' . + 'Array and returns Promise>, but the function did ' . + 'not return a Promise of an Array of the same length as the Array of keys.' + ); + } + + // Step through the values, resolving or rejecting each Promise in the + // loaded queue. + foreach ($queue as $index => $data) { + $value = $values[$index]; + if ($value instanceof \Throwable) { + $data['reject']($value); + } else { + $data['resolve']($value); + } + } + } + /** * Do not cache individual loads if the entire batch dispatch fails, * but still reject each request so they do not hang. From f7a0eb3e378ec06618f0aa9f83f931954239b544 Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 10:02:44 +0200 Subject: [PATCH 4/7] test: reuse amp adapter fixture --- tests/AmpFutureDataLoaderTest.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/AmpFutureDataLoaderTest.php b/tests/AmpFutureDataLoaderTest.php index cc15277..7548066 100644 --- a/tests/AmpFutureDataLoaderTest.php +++ b/tests/AmpFutureDataLoaderTest.php @@ -30,7 +30,7 @@ public function testAwaitUsesTheRegisteredAmpFutureAdapter() { $loader = new DataLoader( static fn (array $keys): Future => Future::complete(['value', 'value']), - new AmpFutureAdapter(), + $this->createPromiseAdapter(), ); self::assertSame(['value', 'value'], DataLoader::await($loader->loadMany(['first', 'second']))); From 9f3c993334f96ff60e556b75b40519903b86b668 Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 11:40:59 +0200 Subject: [PATCH 5/7] fix: type amp adapter futures --- lib/promise-adapter/src/Adapter/AmpFutureAdapter.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php b/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php index ad24b0a..9c4dd89 100644 --- a/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php +++ b/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php @@ -25,7 +25,7 @@ * The DataLoader core is patched to cooperate with this model via `Amp\async` * and `Future::await()` instead of `->then()`. * - * @implements PromiseAdapterInterface + * @implements PromiseAdapterInterface> */ class AmpFutureAdapter implements PromiseAdapterInterface { From 4873ce35252300cc852c02e2d9d44d8ecbb5a3e2 Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 16:43:03 +0200 Subject: [PATCH 6/7] fix(amp): make future lifecycle reliable Context: Amp futures require event-loop scheduling, explicit observation, and synchronous cancellation handling. Decision: add an optional async adapter interface and keep Amp scheduling, draining, observation, and cancellation inside the adapter. Consequences: DataLoader remains promise-library agnostic, and nested or concurrent waits settle without deadlocks. --- ...001-support-event-loop-promise-adapters.md | 34 ++++ lib/promise-adapter/docs/usage.md | 11 ++ .../src/Adapter/AmpFutureAdapter.php | 154 +++++++++++++++++- .../src/AsyncPromiseAdapterInterface.php | 38 +++++ .../tests/AmpFutureAdapterTest.php | 125 ++++++++++++++ src/DataLoader.php | 36 ++-- tests/AmpFutureDataLoaderTest.php | 146 +++++++++++++++++ 7 files changed, 517 insertions(+), 27 deletions(-) create mode 100644 docs/adr/0001-support-event-loop-promise-adapters.md create mode 100644 lib/promise-adapter/src/AsyncPromiseAdapterInterface.php create mode 100644 lib/promise-adapter/tests/AmpFutureAdapterTest.php diff --git a/docs/adr/0001-support-event-loop-promise-adapters.md b/docs/adr/0001-support-event-loop-promise-adapters.md new file mode 100644 index 0000000..aaaa70f --- /dev/null +++ b/docs/adr/0001-support-event-loop-promise-adapters.md @@ -0,0 +1,34 @@ +# Support Event-Loop Promise Adapters + +## Status + +Accepted + +## Context + +DataLoader dispatches promise batches through adapters. Existing adapters expose +promises with a `then()` method or provide a synchronous drain operation. Amp v3 +Futures expose neither behavior. They require event-loop scheduling and explicit +observation. + +Putting Amp and Revolt checks in DataLoader would couple its batching logic to one +promise implementation. Other event-loop promise adapters would require more +concrete checks in the same module. + +## Decision + +Add `AsyncPromiseAdapterInterface` as an optional extension of +`PromiseAdapterInterface`. It lets an adapter enqueue a batch dispatch and observe +a promise without exposing implementation-specific promise or event-loop types to +DataLoader. + +Existing adapters continue to implement `PromiseAdapterInterface` without +changes. `AmpFutureAdapter` implements the optional interface and owns its +scheduling, observation, pending-work draining, and cancellation behavior. + +## Consequences + +DataLoader does not depend directly on Amp or Revolt classes. New event-loop +adapters can use the same seam without changes to DataLoader. Async adapters must +track scheduled and observed work so a no-argument `await()` drains all work and +does not leave unhandled failures. diff --git a/lib/promise-adapter/docs/usage.md b/lib/promise-adapter/docs/usage.md index 82bc6ec..808ab56 100644 --- a/lib/promise-adapter/docs/usage.md +++ b/lib/promise-adapter/docs/usage.md @@ -14,10 +14,21 @@ Optional to use ReactPhp: composer require "react/promise" ``` +Optional to use Amp v3: + +```sh +composer require "amphp/amp" +``` + ## Supported Adapter *Guzzle*: `Overblog\PromiseAdapter\Adapter\GuzzleHttpPromiseAdapter` *ReactPhp*: `Overblog\PromiseAdapter\Adapter\ReactPromiseAdapter` +*Amp v3*: `Overblog\PromiseAdapter\Adapter\AmpFutureAdapter` + To use a custom Promise lib you can implement `Overblog\PromiseAdapter\PromiseAdapterInterface` + +Adapters for promises that require event-loop scheduling and do not expose a +`then()` method can implement `Overblog\PromiseAdapter\AsyncPromiseAdapterInterface`. diff --git a/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php b/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php index 9c4dd89..3a6d4a7 100644 --- a/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php +++ b/lib/promise-adapter/src/Adapter/AmpFutureAdapter.php @@ -11,11 +11,13 @@ namespace Overblog\PromiseAdapter\Adapter; +use Amp\CancelledException; use Amp\DeferredFuture; use Amp\Future; -use Overblog\PromiseAdapter\PromiseAdapterInterface; +use Overblog\PromiseAdapter\AsyncPromiseAdapterInterface; use function Amp\async; use function Amp\Future\await; +use function Amp\Future\awaitAll; /** * Promise adapter backed by amphp/amp v3 fiber-based futures. @@ -25,33 +27,52 @@ * The DataLoader core is patched to cooperate with this model via `Amp\async` * and `Future::await()` instead of `->then()`. * - * @implements PromiseAdapterInterface> + * @implements AsyncPromiseAdapterInterface> */ -class AmpFutureAdapter implements PromiseAdapterInterface +class AmpFutureAdapter implements AsyncPromiseAdapterInterface { + /** @var \WeakMap, array{deferred: DeferredFuture|null, canceller: callable|null}>|null */ + private ?\WeakMap $cancellations = null; + + /** @var array> */ + private array $pending = []; + + /** @var array */ + private array $running = []; + + private int $nextPendingId = 0; + /** * @return Future */ public function create(&$resolve = null, &$reject = null, ?callable $canceller = null): Future { $deferred = new DeferredFuture(); + $future = $deferred->getFuture(); + $this->cancellations ??= new \WeakMap(); + $this->cancellations[$future] = [ + 'deferred' => $deferred, + 'canceller' => $canceller, + ]; - $resolve = static function ($value) use ($deferred): void { + $resolve = function ($value) use ($deferred, $future): void { if ($deferred->isComplete()) { return; } $deferred->complete($value); + $this->markSettled($future); }; - $reject = static function (\Throwable $reason) use ($deferred): void { + $reject = function (\Throwable $reason) use ($deferred, $future): void { if ($deferred->isComplete()) { return; } $deferred->error($reason); + $this->markSettled($future); }; - return $deferred->getFuture(); + return $future; } /** @@ -71,6 +92,10 @@ public function createFulfilled($promiseOrValue = null): Future */ public function createRejected($reason): Future { + if ($reason instanceof Future) { + return $reason; + } + return Future::error($reason); } @@ -92,9 +117,35 @@ public function isPromise($value, $strict = false): bool return $value instanceof Future; } - public function await($promise = null, $unwrap = true): mixed + public function await($promise = null, $unwrap = false): mixed { if (null === $promise) { + $firstError = null; + + while ([] !== $this->pending) { + $pending = $this->pending; + $currentFiber = \Fiber::getCurrent(); + if (null !== $currentFiber) { + foreach ($this->running as $id => $fiber) { + if ($fiber === $currentFiber) { + unset($pending[$id]); + } + } + } + if ([] === $pending) { + break; + } + [$errors] = awaitAll($pending); + + if (null === $firstError && [] !== $errors) { + $firstError = reset($errors); + } + } + + if (null !== $firstError) { + throw $firstError; + } + return null; } @@ -115,7 +166,92 @@ public function await($promise = null, $unwrap = true): mixed public function cancel($promise): void { - // amp v3 futures are cancelled through a Cancellation token passed at - // creation time; there is no post-hoc cancel handle here. + if (!$promise instanceof Future || null === $this->cancellations || !$this->cancellations->offsetExists($promise)) { + throw new \InvalidArgumentException(sprintf('The "%s" method must be called with a compatible Future.', __METHOD__)); + } + + $cancellation = $this->cancellations[$promise]; + $deferred = $cancellation['deferred']; + if (null === $deferred) { + return; + } + + $this->markSettled($promise); + $promise->ignore(); + + try { + if (null !== $cancellation['canceller']) { + ($cancellation['canceller'])(); + } + } catch (\Throwable $reason) { + if (!$deferred->isComplete()) { + $deferred->error($reason); + } + + return; + } + + if (!$deferred->isComplete()) { + $deferred->error(new CancelledException()); + } + } + + public function enqueue(callable $callback): void + { + $id = ++$this->nextPendingId; + $this->track(async(function () use ($callback, $id): void { + $this->running[$id] = \Fiber::getCurrent(); + + try { + $callback(); + } finally { + unset($this->running[$id]); + } + }), $id); + } + + public function observe($promise, callable $onFulfilled, callable $onRejected): void + { + if (!$promise instanceof Future) { + throw new \InvalidArgumentException(sprintf('The "%s" method must be called with a compatible Future.', __METHOD__)); + } + + $id = ++$this->nextPendingId; + $observer = async(function () use ($promise, $onFulfilled, $onRejected, $id): void { + $this->running[$id] = \Fiber::getCurrent(); + + try { + try { + $onFulfilled($promise->await()); + } catch (\Throwable $error) { + $onRejected($error); + } + } finally { + unset($this->running[$id]); + } + }); + + $this->track($observer, $id); + } + + private function markSettled(Future $future): void + { + if (null === $this->cancellations || !$this->cancellations->offsetExists($future)) { + return; + } + + $this->cancellations[$future] = [ + 'deferred' => null, + 'canceller' => null, + ]; + } + + private function track(Future $future, int $id): void + { + $tracked = $future->finally(function () use ($id): void { + unset($this->pending[$id]); + }); + $tracked->ignore(); + $this->pending[$id] = $tracked; } } diff --git a/lib/promise-adapter/src/AsyncPromiseAdapterInterface.php b/lib/promise-adapter/src/AsyncPromiseAdapterInterface.php new file mode 100644 index 0000000..8edcd09 --- /dev/null +++ b/lib/promise-adapter/src/AsyncPromiseAdapterInterface.php @@ -0,0 +1,38 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Overblog\PromiseAdapter; + +/** + * Supports promises that require event-loop scheduling and do not expose then(). + * + * @template TPromise + * + * @extends PromiseAdapterInterface + */ +interface AsyncPromiseAdapterInterface extends PromiseAdapterInterface +{ + /** + * Queue work for the next event-loop turn. + * + * A no-argument await() call must drain this work, including work that it + * enqueues or observes. cancel() must settle adapter-created promises + * without requiring an event-loop turn. + */ + public function enqueue(callable $callback): void; + + /** + * @param TPromise $promise + * + * A no-argument await() call must wait for the observer callbacks to finish. + */ + public function observe($promise, callable $onFulfilled, callable $onRejected): void; +} diff --git a/lib/promise-adapter/tests/AmpFutureAdapterTest.php b/lib/promise-adapter/tests/AmpFutureAdapterTest.php new file mode 100644 index 0000000..92b8458 --- /dev/null +++ b/lib/promise-adapter/tests/AmpFutureAdapterTest.php @@ -0,0 +1,125 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Overblog\PromiseAdapter\Tests; + +use Amp\CancelledException; +use Amp\Future; +use Overblog\PromiseAdapter\Adapter\AmpFutureAdapter; + +use function Amp\async; +use function Amp\delay; + +class AmpFutureAdapterTest extends \PHPUnit\Framework\TestCase +{ + public function testAwaitReturnsRejectionWithoutUnwrapByDefault(): void + { + $adapter = new AmpFutureAdapter(); + $error = new \RuntimeException('failed'); + + self::assertSame($error, $adapter->await(Future::error($error))); + } + + public function testCreateRejectedPreservesFuture(): void + { + $adapter = new AmpFutureAdapter(); + $future = Future::complete('value'); + + self::assertSame($future, $adapter->createRejected($future)); + } + + public function testCancelInvokesCancellerAndRejectsFuture(): void + { + $adapter = new AmpFutureAdapter(); + $error = new \RuntimeException('cancelled'); + $future = $adapter->create( + $resolve, + $reject, + static function () use ($error): void { + throw $error; + }, + ); + + $adapter->cancel($future); + + self::assertSame($error, $adapter->await($future)); + } + + public function testCancelRejectsInvalidFuture(): void + { + $adapter = new AmpFutureAdapter(); + + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('::cancel" method must be called with a compatible Future.'); + + $adapter->cancel(Future::complete(null)); + } + + public function testCancelWithoutCancellerRejectsFuture(): void + { + $adapter = new AmpFutureAdapter(); + $future = $adapter->create($resolve, $reject); + + $adapter->cancel($future); + + self::assertInstanceOf(CancelledException::class, $adapter->await($future)); + } + + public function testAwaitDrainsWorkEnqueuedByPendingWork(): void + { + $adapter = new AmpFutureAdapter(); + $calls = []; + $adapter->enqueue(function () use ($adapter, &$calls): void { + $calls[] = 'first'; + $adapter->enqueue(function () use (&$calls): void { + $calls[] = 'second'; + }); + }); + + $adapter->await(); + + self::assertSame(['first', 'second'], $calls); + } + + public function testFailedResolutionLeavesFutureCancellable(): void + { + $adapter = new AmpFutureAdapter(); + $future = $adapter->create($resolve, $reject); + + try { + $resolve(Future::complete('invalid nested future')); + self::fail('Expected resolving with a Future to fail.'); + } catch (\Error $error) { + self::assertSame('Cannot complete with an instance of Amp\Future', $error->getMessage()); + } + + $adapter->cancel($future); + + self::assertInstanceOf(CancelledException::class, $adapter->await($future)); + } + + public function testAwaitWaitsForWorkRunningInAnotherFiber(): void + { + $adapter = new AmpFutureAdapter(); + $completed = false; + $adapter->enqueue(function () use (&$completed): void { + delay(0.01); + $completed = true; + }); + + async(function () use ($adapter, &$completed): void { + delay(0); + $adapter->await(); + + self::assertTrue($completed); + })->await(); + } +} diff --git a/src/DataLoader.php b/src/DataLoader.php index c028899..3cf9d3b 100644 --- a/src/DataLoader.php +++ b/src/DataLoader.php @@ -11,6 +11,7 @@ namespace Overblog\DataLoader; +use Overblog\PromiseAdapter\AsyncPromiseAdapterInterface; use Overblog\PromiseAdapter\PromiseAdapterInterface; /** @@ -122,13 +123,8 @@ static function () { if (!$shouldBatch) { // Otherwise dispatch the (queue of one) immediately. $this->dispatchQueue(); - } elseif ($this->getPromiseAdapter() instanceof \Overblog\PromiseAdapter\Adapter\AmpFutureAdapter) { - // On the fiber-based adapter there is no synchronous drain loop - // (SyncPromiseAdapter::onWait). Instead, schedule the batch to - // dispatch on the next event-loop tick, once the current call - // stack has finished enqueuing all loads for this frame. This - // mirrors the JS reference implementation's process.nextTick. - \Revolt\EventLoop::queue(function (): void { + } elseif ($this->getPromiseAdapter() instanceof AsyncPromiseAdapterInterface) { + $this->getPromiseAdapter()->enqueue(function (): void { if ($this->needProcess()) { $this->dispatchQueue(); } @@ -216,7 +212,9 @@ public function __destruct() } } - $this->getPromiseAdapter()->await(); + if (!$this->getPromiseAdapter() instanceof AsyncPromiseAdapterInterface) { + $this->getPromiseAdapter()->await(); + } } } @@ -229,7 +227,9 @@ protected function process() { if ($this->needProcess()) { $this->getPromiseAdapter()->await(); - $this->dispatchQueue(); + if ($this->needProcess()) { + $this->dispatchQueue(); + } $this->getPromiseAdapter()->await(); } } @@ -425,17 +425,17 @@ private function dispatchQueueBatch(array $queue) return; } - // amp v3 futures are fiber-based and expose no `then()`; drive the - // fan-out inside a fiber via `Amp\async` instead. - if ($batchPromise instanceof \Amp\Future) { - \Amp\async(function () use ($batchPromise, $queue) { - try { - $values = $batchPromise->await(); + $promiseAdapter = $this->getPromiseAdapter(); + if ($promiseAdapter instanceof AsyncPromiseAdapterInterface && $promiseAdapter->isPromise($batchPromise, true)) { + $promiseAdapter->observe( + $batchPromise, + function ($values) use ($queue): void { $this->resolveDispatchedBatch($values, $queue); - } catch (\Throwable $error) { + }, + function (\Throwable $error) use ($queue): void { $this->failedDispatch($queue, $error); - } - }); + }, + ); return; } diff --git a/tests/AmpFutureDataLoaderTest.php b/tests/AmpFutureDataLoaderTest.php index 7548066..f860927 100644 --- a/tests/AmpFutureDataLoaderTest.php +++ b/tests/AmpFutureDataLoaderTest.php @@ -15,6 +15,7 @@ use Error; use Overblog\DataLoader\DataLoader; use Overblog\PromiseAdapter\Adapter\AmpFutureAdapter; +use Overblog\PromiseAdapter\AsyncPromiseAdapterInterface; use Overblog\PromiseAdapter\PromiseAdapterInterface; use function Amp\async; @@ -55,4 +56,149 @@ public function testRejectsEveryQueuedLoadWhenBatchFutureFailsWithAnError() } } } + + public function testAwaitWithoutPromiseCompletesQueuedLoads(): void + { + $loader = new DataLoader( + static fn (array $keys): Future => Future::complete($keys), + new AmpFutureAdapter(), + ); + $future = $loader->load('value'); + + DataLoader::await(); + + self::assertTrue($future->isComplete()); + self::assertSame('value', $future->await()); + } + + public function testDestructionRejectsQueuedLoads(): void + { + $loader = new DataLoader( + static fn (array $keys): Future => Future::complete($keys), + new AmpFutureAdapter(), + ); + $future = $loader->load('value'); + + $loader->__destruct(); + unset($loader); + + $error = DataLoader::await($future, false); + self::assertInstanceOf(\RuntimeException::class, $error); + self::assertSame('DataLoader destroyed before promise complete.', $error->getMessage()); + } + + public function testSupportsDelegatingAsyncPromiseAdapter(): void + { + $adapter = new DelegatingAsyncPromiseAdapter(); + $loader = new DataLoader( + static fn (array $keys): Future => Future::complete($keys), + $adapter, + ); + + self::assertSame(['first', 'second'], DataLoader::await($loader->loadMany(['first', 'second']))); + } + + public function testDestructionDoesNotDispatchUnrelatedLoader(): void + { + $adapter = new AmpFutureAdapter(); + $loader = new DataLoader( + static fn (array $keys): Future => Future::complete($keys), + $adapter, + ); + $loader->load('cancelled')->ignore(); + + $unrelatedLoadCalls = []; + $unrelatedLoader = new DataLoader( + static function (array $keys) use (&$unrelatedLoadCalls): Future { + $unrelatedLoadCalls[] = $keys; + + return Future::complete($keys); + }, + $adapter, + ); + $unrelatedFuture = $unrelatedLoader->load('unrelated'); + + $loader->__destruct(); + unset($loader); + + self::assertSame([], $unrelatedLoadCalls); + self::assertFalse($unrelatedFuture->isComplete()); + self::assertSame('unrelated', DataLoader::await($unrelatedFuture)); + } + + public function testBatchLoaderCanAwaitAnotherLoader(): void + { + $adapter = new AmpFutureAdapter(); + $innerLoader = new DataLoader( + static fn (array $keys): Future => Future::complete($keys), + $adapter, + ); + $outerLoader = new DataLoader( + static function (array $keys) use ($innerLoader): Future { + return Future::complete(array_map( + static fn ($key) => DataLoader::await($innerLoader->load($key)), + $keys, + )); + }, + $adapter, + ); + + self::assertSame('value', DataLoader::await($outerLoader->load('value'))); + } +} + +/** @implements AsyncPromiseAdapterInterface> */ +final class DelegatingAsyncPromiseAdapter implements AsyncPromiseAdapterInterface +{ + private AmpFutureAdapter $adapter; + + public function __construct() + { + $this->adapter = new AmpFutureAdapter(); + } + + public function create(&$resolve = null, &$reject = null, ?callable $canceller = null): Future + { + return $this->adapter->create($resolve, $reject, $canceller); + } + + public function createFulfilled($promiseOrValue = null): Future + { + return $this->adapter->createFulfilled($promiseOrValue); + } + + public function createRejected($reason): Future + { + return $this->adapter->createRejected($reason); + } + + public function createAll($promisesOrValues): Future + { + return $this->adapter->createAll($promisesOrValues); + } + + public function isPromise($value, $strict = false): bool + { + return $this->adapter->isPromise($value, $strict); + } + + public function await($promise = null, $unwrap = false): mixed + { + return $this->adapter->await($promise, $unwrap); + } + + public function cancel($promise): void + { + $this->adapter->cancel($promise); + } + + public function enqueue(callable $callback): void + { + $this->adapter->enqueue($callback); + } + + public function observe($promise, callable $onFulfilled, callable $onRejected): void + { + $this->adapter->observe($promise, $onFulfilled, $onRejected); + } } From e895270832151933d032ab31dd554192c33a577b Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Wed, 19 Aug 2026 16:45:13 +0200 Subject: [PATCH 7/7] docs: remove async adapter ADR --- ...001-support-event-loop-promise-adapters.md | 34 ------------------- 1 file changed, 34 deletions(-) delete mode 100644 docs/adr/0001-support-event-loop-promise-adapters.md diff --git a/docs/adr/0001-support-event-loop-promise-adapters.md b/docs/adr/0001-support-event-loop-promise-adapters.md deleted file mode 100644 index aaaa70f..0000000 --- a/docs/adr/0001-support-event-loop-promise-adapters.md +++ /dev/null @@ -1,34 +0,0 @@ -# Support Event-Loop Promise Adapters - -## Status - -Accepted - -## Context - -DataLoader dispatches promise batches through adapters. Existing adapters expose -promises with a `then()` method or provide a synchronous drain operation. Amp v3 -Futures expose neither behavior. They require event-loop scheduling and explicit -observation. - -Putting Amp and Revolt checks in DataLoader would couple its batching logic to one -promise implementation. Other event-loop promise adapters would require more -concrete checks in the same module. - -## Decision - -Add `AsyncPromiseAdapterInterface` as an optional extension of -`PromiseAdapterInterface`. It lets an adapter enqueue a batch dispatch and observe -a promise without exposing implementation-specific promise or event-loop types to -DataLoader. - -Existing adapters continue to implement `PromiseAdapterInterface` without -changes. `AmpFutureAdapter` implements the optional interface and owns its -scheduling, observation, pending-work draining, and cancellation behavior. - -## Consequences - -DataLoader does not depend directly on Amp or Revolt classes. New event-loop -adapters can use the same seam without changes to DataLoader. Async adapters must -track scheduled and observed work so a no-argument `await()` drains all work and -does not leave unhandled failures.