diff --git a/conformance/Dockerfile.amp b/conformance/Dockerfile.amp new file mode 100644 index 0000000..9c981b6 --- /dev/null +++ b/conformance/Dockerfile.amp @@ -0,0 +1,36 @@ +# Cross-SDK conformance test-services image, BIDIRECTIONAL variant: plain PHP 8.4 CLI + +# amphp/http-server (NO ext-swoole), serving the conformance services over HTTP/2 cleartext +# (h2c) so the Restate runtime opens a true bidirectional invocation stream. Built by the +# `conformance-amp` make target and driven by restatedev/e2e (sdk-tests.jar). +# +# The request/response Swoole conformance image lives at conformance/Dockerfile and is +# unaffected. Mirrors docker/php-amp/Dockerfile but ships the conformance services. +FROM php:8.4-cli + +# git + unzip let Composer extract dist archives (the slim php:8.4-cli image ships neither, +# unlike phpswoole/swoole). ext-pcntl provides the signal handling AmpStreamingServer::listen() +# traps for graceful shutdown; mbstring + sodium are already bundled in php:8.4-cli, and amphp +# uses non-blocking sockets so no event-loop extension is required. +RUN apt-get update \ + && apt-get install -y --no-install-recommends git unzip \ + && rm -rf /var/lib/apt/lists/* \ + && docker-php-ext-install pcntl + +WORKDIR /app + +COPY --from=composer:2 /usr/bin/composer /usr/bin/composer + +# amphp/http-server is a require-dev / suggest dependency, so dev deps are required here +# (--no-dev would skip it). composer.lock is gitignored for a library and excluded by +# .dockerignore, so — like conformance/Dockerfile — only composer.json is copied and +# Composer resolves the stable deps fresh. +COPY composer.json ./ +RUN composer install --no-interaction --no-progress --optimize-autoloader + +COPY src ./src +COPY conformance ./conformance +RUN composer dump-autoload --optimize + +EXPOSE 9080 + +CMD ["php", "conformance/main-amp.php"] diff --git a/conformance/Dockerfile.restate-v7 b/conformance/Dockerfile.restate-v7 new file mode 100644 index 0000000..0413a69 --- /dev/null +++ b/conformance/Dockerfile.restate-v7 @@ -0,0 +1,11 @@ +# Restate runtime with service protocol V7 enabled, for the bidirectional (amp) conformance. +# +# This SDK speaks service protocol V7 (signals, signal-backed awakeables, the Future-based +# SuspensionMessage, AwaitingOnMessage). Restate 1.7.0 supports V7 but it is OPT-IN — by +# default it negotiates V6, on which the SDK's signal/awakeable model is invalid. The flag +# below makes the runtime offer V7 so cancellation, awakeables and kill work over bidi. +# +# Build: docker build -f conformance/Dockerfile.restate-v7 -t localhost/restatedev/restate-v7:latest . +# Use: --restate-container-image=localhost/restatedev/restate-v7:latest --image-pull-policy=CACHED +FROM docker.io/restatedev/restate:latest +ENV RESTATE_EXPERIMENTAL_ENABLE_PROTOCOL_V7=true diff --git a/conformance/README.md b/conformance/README.md index 6cea229..f1922c9 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -18,11 +18,20 @@ ProxyRequestSigning (Ed25519 request identity) · Combinators (awakeable-or-time SleepWithFailures · StopRuntime / KillRuntime (durability across restarts) · UpgradeWithNewInvocation · KafkaIngress · Ingress (header pass-through). -8 tests are excluded with documented reasons in `exclusions.yaml` (cancellation / -kill of *suspended* invocations, which the Rust SDK passes over the bidirectional -transport this SDK does not implement — see `../docs/adr/0001-request-response-transport.md`; -`awaitAny` combinator edge cases the Rust SDK also excludes; per-handler raw serde; -one fan-out ordering case; in-flight deployment upgrade). +A few tests are excluded with documented reasons in `exclusions.yaml` (`awaitAny` +combinator edge cases the Rust SDK also excludes; per-handler raw serde; one fan-out +ordering case; in-flight deployment upgrade; V7 scoped concurrency). + +### Bidirectional (amp) transport on service protocol V7 + +The `AmpStreamingServer` transport speaks service protocol **V7** (signals, signal-backed +awakeables, named signals, the Future-based suspension/`AwaitingOn`). Against a V7-enabled +runtime (`Dockerfile.restate-v7`) the `default` suite passes **48 / 49**, including +`Cancellation` 6/6, `KillInvocation` 1/1, `Signals` 2/2, `Combinators` 9/9, +`RunRetry` 3/3, `UserErrors` 10/10, and `ServiceToServiceCommunication` 5/5. The remaining +exclusions are the same documented gaps as above plus `ServiceToServiceScopeConcurrency` +(V7 scoped concurrency / virtual queues — not yet implemented). See the run instructions +below and `../docs/adr/0001-cancellation-over-bidirectional-streaming.md`. ## Run it @@ -43,5 +52,24 @@ java -jar build/restate-sdk-test-suite.jar run \ discovery handshake flaky on a single host. The PHP server speaks HTTP/2 cleartext (h2c), which the runtime uses for discovery + invocation. +### Bidirectional (HTTP/2) streaming + service protocol V7 + +The `AmpStreamingServer` transport (`conformance/Dockerfile.amp` → +`localhost/restatedev/php-amp-test-services`) serves true bidi h2c and speaks service +protocol **V7** (signals, signal-backed awakeables, `AwaitingOnMessage`). Restate 1.7.0 +supports V7 but negotiates V6 by default, so build a V7-enabled runtime image first: + +```bash +docker build -f conformance/Dockerfile.restate-v7 -t localhost/restatedev/restate-v7:latest . +docker build -f conformance/Dockerfile.amp -t localhost/restatedev/php-amp-test-services:latest . + +java -jar build/sdk-tests.jar run \ + --restate-container-image=localhost/restatedev/restate-v7:latest \ + --service-container-image=localhost/restatedev/php-amp-test-services:latest \ + --test-suite=default --test-name=Cancellation \ + --exclusions-file=conformance/exclusions.yaml \ + --image-pull-policy=CACHED --report-dir=build/conformance-amp-report --sequential +``` + After a run, `build/conformance-report//exclusions.new.yaml` lists everything that failed/was skipped — copy entries into `exclusions.yaml` to baseline new gaps. diff --git a/conformance/Services/Failing.php b/conformance/Services/Failing.php index 7b17949..4a60f51 100644 --- a/conformance/Services/Failing.php +++ b/conformance/Services/Failing.php @@ -29,21 +29,30 @@ final class Failing private int $eventualSuccessSideEffects = 0; private int $eventualFailureSideEffects = 0; + /** + * @param array{errorMessage?: string, metadata?: array} $failureToPropagate + */ #[Handler] - public function terminallyFailingCall(ObjectContext $ctx, string $errorMessage): void + public function terminallyFailingCall(ObjectContext $ctx, array $failureToPropagate): void { - throw new TerminalException($errorMessage); + throw new TerminalException( + (string) ($failureToPropagate['errorMessage'] ?? ''), + metadata: $failureToPropagate['metadata'] ?? [], + ); } + /** + * @param array{errorMessage?: string, metadata?: array} $failureToPropagate + */ #[Handler] - public function callTerminallyFailingCall(ObjectContext $ctx, string $errorMessage): string + public function callTerminallyFailingCall(ObjectContext $ctx, array $failureToPropagate): string { $uuid = $ctx->random()->uuidV4(); - // The callee fails terminally; awaiting the call rethrows that terminal - // failure here, so it propagates to our caller and the line below is never - // reached (mirrors the Rust `unreachable!`). - $ctx->objectCall('Failing', $uuid, 'terminallyFailingCall', $errorMessage); + // The callee fails terminally; awaiting the call rethrows that terminal failure + // here (metadata included), so it propagates to our caller and the line below is + // never reached (mirrors the Rust `unreachable!`). + $ctx->objectCall('Failing', $uuid, 'terminallyFailingCall', $failureToPropagate); throw new TerminalException('This should be unreachable'); } @@ -64,13 +73,19 @@ public function failingCallWithEventualSuccess(ObjectContext $ctx): int throw new RuntimeException('Failed at attempt ${current_attempt}'); } + /** + * @param array{errorMessage?: string, metadata?: array} $failureToPropagate + */ #[Handler] - public function terminallyFailingSideEffect(ObjectContext $ctx, string $errorMessage): void + public function terminallyFailingSideEffect(ObjectContext $ctx, array $failureToPropagate): void { + $errorMessage = (string) ($failureToPropagate['errorMessage'] ?? ''); + $metadata = $failureToPropagate['metadata'] ?? []; + // A terminal failure raised inside a run is not retried: it is journaled and // propagates out of the handler. - $ctx->run('sideEffect', static function () use ($errorMessage): void { - throw new TerminalException($errorMessage); + $ctx->run('sideEffect', static function () use ($errorMessage, $metadata): void { + throw new TerminalException($errorMessage, metadata: $metadata); }); } diff --git a/conformance/Services/TestUtilsService.php b/conformance/Services/TestUtilsService.php index 3ef031a..1c0e6ec 100644 --- a/conformance/Services/TestUtilsService.php +++ b/conformance/Services/TestUtilsService.php @@ -103,4 +103,26 @@ public function cancelInvocation(Context $ctx, string $invocationId): void { $ctx->cancel($invocationId); } + + /** + * Resolves a named signal on another invocation with a value. + * + * @param array{invocationId: string, signalName: string, value?: mixed} $req + */ + #[Handler] + public function resolveSignal(Context $ctx, array $req): void + { + $ctx->resolveSignal((string) $req['invocationId'], (string) $req['signalName'], $req['value'] ?? ''); + } + + /** + * Rejects a named signal on another invocation with a terminal failure reason. + * + * @param array{invocationId: string, signalName: string, reason?: string} $req + */ + #[Handler] + public function rejectSignal(Context $ctx, array $req): void + { + $ctx->rejectSignal((string) $req['invocationId'], (string) $req['signalName'], (string) ($req['reason'] ?? '')); + } } diff --git a/conformance/Services/VirtualObjectCommandInterpreter.php b/conformance/Services/VirtualObjectCommandInterpreter.php index eb795ea..3dc5a5a 100644 --- a/conformance/Services/VirtualObjectCommandInterpreter.php +++ b/conformance/Services/VirtualObjectCommandInterpreter.php @@ -25,15 +25,16 @@ * - "awk-{awakeableKey}" the awakeable id registered for that key (string) * - "results" the ordered list of per-command results * - * Limitations (vs. the contract) — both apply only to the `awaitAny` / - * `awaitAnySuccessful` combinators, whose conformance tests are excluded (they are - * unsupported in the Rust SDK too): - * 1. A `runThrowTerminalException` awaitable cannot be raced concurrently, because - * the PHP SDK's run() is blocking and exposes no future handle; it raises a - * terminal failure if used inside those combinators. It IS supported in `awaitOne`. - * 2. A `sleep` awaitable yields "" instead of "sleep" inside those combinators, - * because a bare durable timer future resolves to null with no way to remap it. - * In `awaitOne`, `sleep` correctly yields "sleep". + * Awaitable commands (createAwakeable, createSignal, sleep, runReturns, + * runThrowTerminalException) are all raceable inside the combinators: a run is driven + * through {@see Context::runAsync}, which proposes the side effect and returns its + * completion future without blocking, and a named signal through + * {@see Context::createSignal}. + * + * Limitation (vs. the contract): a `sleep` awaitable yields "" instead of "sleep" + * inside the `select`-based combinators (`awaitAny` / `awaitFirstCompleted`), because a + * bare durable timer future resolves to null with no way to remap it. In `awaitOne` and + * `awaitAllCompleted` (which renders by command type), `sleep` correctly yields "sleep". */ #[VirtualObject(name: 'VirtualObjectCommandInterpreter')] final class VirtualObjectCommandInterpreter @@ -64,12 +65,30 @@ public function interpretCommands(ObjectContext $ctx, array $request): string break; case 'awaitAnySuccessful': + case 'awaitFirstSucceededOrAllFailed': // First awaitable to SUCCEED wins. $lastResult = self::asString( $ctx->awaitAny(...$this->awaitableFutures($ctx, $command['commands'])), ); break; + case 'awaitFirstCompleted': + // First awaitable to COMPLETE (success or failure) wins — like awaitAny. + [, $firstValue] = $ctx->select(...$this->awaitableFutures($ctx, $command['commands'])); + $lastResult = self::asString($firstValue); + break; + + case 'awaitAllCompleted': + // Settle EVERY awaitable, rendering each `ok:` / `err:`. + $lastResult = $this->awaitAllCompleted($ctx, $command['commands']); + break; + + case 'awaitAllSucceededOrFirstFailed': + // All must succeed (values joined); the first failure short-circuits. + $values = $ctx->awaitAllSucceeded($this->awaitableFutures($ctx, $command['commands'])); + $lastResult = \implode('|', \array_map([self::class, 'asString'], $values)); + break; + case 'awaitAwakeableOrTimeout': $awakeable = $ctx->awakeable(); $ctx->set(self::awkKey((string) $command['awakeableKey']), $awakeable->id()); @@ -183,11 +202,19 @@ private function runAwaitableCommand(ObjectContext $ctx, array $command): string return self::asString($awakeable->await()); + case 'createSignal': + return self::asString($ctx->createSignal((string) $command['signalName'])->await()); + case 'sleep': $ctx->sleep(((float) $command['timeoutMillis']) / 1000); return 'sleep'; + case 'runReturns': + $value = (string) ($command['value'] ?? ''); + + return self::asString($ctx->run('cmd', static fn (): string => $value)); + case 'runThrowTerminalException': $reason = (string) ($command['reason'] ?? ''); @@ -200,6 +227,43 @@ private function runAwaitableCommand(ObjectContext $ctx, array $command): string } } + /** + * Settles every awaitable (success or failure) and renders each as `ok:` or + * `err:`, joined with `|`. All futures are started up front, then awaited in + * turn (they run concurrently, so this waits for the slowest, not the sum). + * + * @param list> $commands + */ + private function awaitAllCompleted(ObjectContext $ctx, array $commands): string + { + $pairs = []; + foreach ($commands as $command) { + $pairs[] = [$command, $this->awaitableCommandFuture($ctx, $command)]; + } + + $parts = []; + foreach ($pairs as [$command, $future]) { + try { + $parts[] = 'ok:' . $this->renderAwaitableValue($command, $future->await()); + } catch (TerminalException $e) { + $parts[] = 'err:' . $e->getMessage(); + } + } + + return \implode('|', $parts); + } + + /** + * Renders a settled awaitable's value: a bare durable timer resolves to null, so a + * `sleep` command is rendered by its name rather than the empty value. + * + * @param array $command + */ + private function renderAwaitableValue(array $command, mixed $value): string + { + return ($command['type'] ?? '') === 'sleep' ? 'sleep' : self::asString($value); + } + /** * Builds durable futures for a list of awaitable commands, for concurrent racing. * @@ -231,13 +295,27 @@ private function awaitableCommandFuture(ObjectContext $ctx, array $command): Dur return self::awakeableFuture($awakeable); + case 'createSignal': + return $ctx->createSignal((string) $command['signalName']); + case 'sleep': return $ctx->timer(((float) $command['timeoutMillis']) / 1000); + case 'runReturns': + $value = (string) ($command['value'] ?? ''); + + return $ctx->runAsync('cmd', static fn (): string => $value); + + case 'runThrowTerminalException': + $reason = (string) ($command['reason'] ?? ''); + + return $ctx->runAsync('cmd', static function () use ($reason): string { + throw new TerminalException($reason); + }); + default: - // See the class-level limitation note: a blocking run cannot be raced. throw new TerminalException( - "AwaitableCommand '{$type}' is not supported inside awaitAny/awaitAnySuccessful in the PHP SDK", + "AwaitableCommand '{$type}' is not supported inside combinators in the PHP SDK", ); } } diff --git a/conformance/exclusions.yaml b/conformance/exclusions.yaml index 9d7ab87..d8b87e6 100644 --- a/conformance/exclusions.yaml +++ b/conformance/exclusions.yaml @@ -4,79 +4,68 @@ # skips. Like the Rust SDK's own exclusions.yaml these are documented, legitimate gaps — # everything else passes against a real Restate runtime. # -# Reasons: -# * Cancellation / KillInvocation — cancelling a *suspended* invocation requires the -# runtime to wake it to deliver the CANCEL signal. The SDK declares the cancel signal -# in its suspension await tree, but reliable cancel-on-suspend in Restate's -# request/response transport is not yet matched. The Rust SDK passes these over the -# bidirectional transport, which this SDK does not implement (see docs/adr/0001). +# Cancellation / KillInvocation now PASS over the bidirectional (amp) transport against a +# V7-enabled runtime (build conformance/Dockerfile.restate-v7 — Restate negotiates service +# protocol V6 by default, on which the SDK's signal/awakeable model is invalid). They are +# therefore no longer excluded. See docs/adr/0001. +# +# Reasons (remaining): # * Combinators.awaitAny* — awaitAny/awaitAnySuccessful over a mix of awakeables/timers/ # runs has a documented limitation (the Rust SDK excludes the same tests). # * Ingress.rawHandler — per-handler raw (octet-stream) serde is not yet implemented. # * CallOrdering.ordering — one fan-out ordering case needs concurrent non-blocking raw # calls, which the proxy approximates with sequential awaited calls. # * UpgradeWithInFlightInvocation — deployment upgrade while an invocation is in flight. +# * ServiceToServiceScopeConcurrency — V7 scoped-concurrency (call scope + limit key) is a +# distinct feature needing both SDK support and the runtime's experimental virtual-queues +# subsystem; not yet implemented. exclusions: "default": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" "alwaysSuspending": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" "singleThreadSinglePartition": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" "threeNodes": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" "threeNodesAlwaysSuspending": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" "lazyState": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" "persistedTimers": - - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext" - - "dev.restate.sdktesting.tests.Cancellation.cancelFromAdminAPI" - - "dev.restate.sdktesting.tests.KillInvocation.kill" - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny" - "dev.restate.sdktesting.tests.Combinators.firstSuccessfulCompletedAwakeable" - "dev.restate.sdktesting.tests.Ingress.rawHandler" - "dev.restate.sdktesting.tests.CallOrdering.ordering" - "dev.restate.sdktesting.tests.UpgradeWithInFlightInvocation.inFlightInvocation" + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" diff --git a/conformance/main-amp.php b/conformance/main-amp.php new file mode 100644 index 0000000..372b1e2 --- /dev/null +++ b/conformance/main-amp.php @@ -0,0 +1,100 @@ + $catalog name => instance */ +$catalog = [ + 'Counter' => new Counter(), + 'Proxy' => new Proxy(), + 'MapObject' => new MapObject(), + 'ListObject' => new ListObject(), + 'AwakeableHolder' => new AwakeableHolder(), + 'BlockAndWaitWorkflow' => new BlockAndWaitWorkflow(), + 'CancelTestRunner' => new CancelTestRunner(), + 'CancelTestBlockingService' => new CancelTestBlockingService(), + 'Failing' => new Failing(), + 'KillTestRunner' => new KillTestRunner(), + 'KillTestSingleton' => new KillTestSingleton(), + 'NonDeterministic' => new NonDeterministic(), + 'TestUtilsService' => new TestUtilsService(), + 'VirtualObjectCommandInterpreter' => new VirtualObjectCommandInterpreter(), +]; + +$selection = \getenv('SERVICES') ?: '*'; +$builder = Endpoint::builder()->protocolMode(ProtocolMode::BidiStream); +foreach ($catalog as $name => $instance) { + if ($selection === '*' || \str_contains($selection, $name)) { + $builder->bind($instance); + } +} + +// A real STDERR logger so a streaming-driver error surfaces in the container logs (and the +// conformance suite's captured service log) instead of being swallowed by a NullLogger. +// Filters amphp's chatty notice/info so only warnings and errors show. +$logger = new class () extends AbstractLogger { + private const VISIBLE = [ + LogLevel::WARNING, + LogLevel::ERROR, + LogLevel::CRITICAL, + LogLevel::ALERT, + LogLevel::EMERGENCY, + ]; + + public function log($level, string|Stringable $message, array $context = []): void + { + if (!\in_array($level, self::VISIBLE, true)) { + return; + } + \fwrite(\STDERR, '[' . $level . '] ' . $message . "\n"); + $exception = $context['exception'] ?? null; + if ($exception instanceof Throwable) { + \fwrite(\STDERR, $exception . "\n"); + } + } +}; + +$port = (int) (\getenv('PORT') ?: 9080); +(new AmpStreamingServer($builder->build(), logger: $logger))->listen('0.0.0.0', $port); diff --git a/docs/adr/0001-cancellation-over-bidirectional-streaming.md b/docs/adr/0001-cancellation-over-bidirectional-streaming.md new file mode 100644 index 0000000..d14af45 --- /dev/null +++ b/docs/adr/0001-cancellation-over-bidirectional-streaming.md @@ -0,0 +1,64 @@ +# ADR 0001 — Cancellation over the bidirectional (HTTP/2) streaming transport + +**Status:** Accepted + +## Context + +The SDK gained an AMPHP HTTP/2 bidirectional streaming transport (`AmpStreamingServer`) +alongside the original request/response Swoole server. The cross-SDK conformance suite's +`Cancellation.*` and `KillInvocation.kill` classes failed (0/6 and 0/1) and were excluded. + +Cancelling a *parked* invocation is the hard case: the runtime must wake the invocation to +deliver the built-in CANCEL signal, the handler's pending await must fail, and the cancel +must propagate so the invocation's children and locks are released. Getting this right +turned out to require the **service protocol V7** model end to end, plus several SDK-side +corrections that the conformance suite exercises but unit tests did not. + +## Decision + +Implement cancellation against **service protocol V7**, the version the SDK already targets +(signals, signal-backed awakeables, the Future-based `SuspensionMessage`, and +`AwaitingOnMessage`). Concretely: + +1. **Require a V7-capable runtime.** Restate 1.7.0 supports V7 but negotiates **V6 by + default**; on V6 the SDK's signal/awakeable model is invalid (an awakeable resolution is + applied as a non-existent completion index and crashes the partition). V7 is enabled with + the runtime flag `experimental-enable-protocol-v7` (env + `RESTATE_EXPERIMENTAL_ENABLE_PROTOCOL_V7=true`). The bidi conformance uses + `conformance/Dockerfile.restate-v7`, which bakes that flag onto `restate:latest`. + +2. **Signal-backed awakeable id.** Awakeable ids use the `sign_1` prefix (signal-backed), + not the legacy completion-backed `prom_1` (`src/Vm/AwakeableId.php`). + +3. **Announce await points.** On every streaming park the SDK emits an `AwaitingOnMessage` + (`FiberSuspender` → `StateMachine::writeAwaitingOn`) so the runtime knows what a parked + invocation awaits and pushes the matching completion/signal onto the open stream. + +4. **Canonical cancel-guard await tree.** A single-leaf await flattens its ids next to the + CANCEL signal under a `FirstCompleted` node (`StateMachine::guardWithCancel`) — the flat + shape the runtime keys its cancel wake-up off — rather than nesting the await beneath a + top-level signal. + +5. **Implicit cancellation propagation.** A handler tracks the invocation ids of the calls + it issues and, when cancelled at an await, sends the CANCEL signal to each known child + before failing with 409 (`StateMachine::raiseCancellation`). A cancelled parent therefore + tears down the calls it was blocked on, so children release their virtual-object locks. + +6. **Raised amphp connection limits.** The runtime opens one long-lived bidi connection per + in-flight invocation, all from one IP; amphp's defaults (1000 total, 10 per IP, 1000 + concurrent) starve under load ("too many existing connections"). The limits are raised so + the runtime governs concurrency (`AmpStreamingServer`). + +Two supporting streaming-driver fixes also landed: the await/cancel combinators raise +`CancelledException` (409) when woken only by a cancel, and the driver drains every +already-resolvable park per inbound chunk rather than suspending spuriously. + +## Consequences + +- Against a V7 runtime over bidi: `Cancellation` 6/6, `KillInvocation` 1/1, with no + regression — `State` 3/3 and `ServiceToServiceCommunication` 5/5 (the latter previously + 4/5; V7 also fixed `oneWayCallWithDelay`). The exclusions are removed + (`conformance/exclusions.yaml`). +- The bidi transport now **requires a V7-enabled runtime** for awakeables and cancellation. + Against a default (V6) runtime those features do not work; basic features still do. +- The request/response (Swoole) transport is unchanged and not covered by this ADR. diff --git a/src/Context/Context.php b/src/Context/Context.php index 63f6dce..10b9ef3 100644 --- a/src/Context/Context.php +++ b/src/Context/Context.php @@ -62,6 +62,17 @@ public function traceContext(): ?TraceContext; */ public function run(string $name, callable $action, ?RunOptions $options = null): mixed; + /** + * Executes a side effect durably WITHOUT awaiting it, returning a future that + * resolves to the journaled result. Like {@see run} the closure runs once and its + * result is persisted, but control returns immediately so the run can be composed + * concurrently (e.g. raced via {@see select} / {@see awaitAll} against timers, + * calls or signals). Await the returned future to obtain the value. + * + * @param callable():mixed $action + */ + public function runAsync(string $name, callable $action): DurableFuture; + /** Suspends the invocation for the given duration using a durable timer. */ public function sleep(float $seconds): void; @@ -299,6 +310,20 @@ public function resolveAwakeable(string $id, mixed $value = null): void; public function rejectAwakeable(string $id, string $message): void; + /** + * Creates a future that resolves when a named signal is delivered to THIS + * invocation. The signal is addressed by the chosen `$name`: another invocation + * resolves it via {@see resolveSignal} (or rejects it via {@see rejectSignal}) + * targeting this invocation's id and the same name. + */ + public function createSignal(string $name): DurableFuture; + + /** Resolves a named signal on another invocation with a value. */ + public function resolveSignal(string $invocationId, string $name, mixed $value = null): void; + + /** Rejects a named signal on another invocation with a terminal failure reason. */ + public function rejectSignal(string $invocationId, string $name, string $reason): void; + /** Deterministic, replay-stable randomness seeded by the runtime. */ public function random(): ContextRand; diff --git a/src/Context/DurableFuture.php b/src/Context/DurableFuture.php index 321edb3..ed1171b 100644 --- a/src/Context/DurableFuture.php +++ b/src/Context/DurableFuture.php @@ -11,23 +11,30 @@ use Qcodr\Restate\Sdk\Vm\StateMachine; /** - * A pending durable result (a call result, a timer, or an awakeable). + * A pending durable result (a call result, a timer, an awakeable, or a named signal). * * Awaiting it returns the value if the completion is already in the replayed * journal, decoding the payload via the supplied decoder; otherwise the state * machine suspends the invocation. A failure result is raised as a * {@see TerminalException}. + * + * Three addressing modes: a completion id (calls, timers, runs), a signal index + * (awakeables), or a user-chosen signal name (named signals). When {@see $signalName} + * is set the future routes through the VM's named-signal table regardless of $id. */ final class DurableFuture { /** - * @param (Closure(string): mixed)|null $decoder maps the raw value bytes to a PHP value + * @param (Closure(string): mixed)|null $decoder maps the raw value bytes to a PHP value + * @param ?string $signalName when set, the future awaits this named + * signal instead of a completion/signal id */ public function __construct( private readonly StateMachine $vm, private readonly int $id, private readonly bool $isSignal, private readonly ?Closure $decoder = null, + private readonly ?string $signalName = null, ) { } @@ -41,9 +48,24 @@ public function isSignal(): bool return $this->isSignal; } + /** Whether this future awaits a user-chosen named signal (rather than a completion/signal id). */ + public function isNamedSignal(): bool + { + return $this->signalName !== null; + } + + public function signalName(): ?string + { + return $this->signalName; + } + /** Whether the result is already available (in the replayed journal). */ public function isReady(): bool { + if ($this->signalName !== null) { + return $this->vm->isNamedSignalReady($this->signalName); + } + return $this->isSignal ? $this->vm->isSignalReady($this->id) : $this->vm->isCompletionReady($this->id); @@ -63,18 +85,16 @@ public function isFailed(): bool return false; } - $notification = $this->isSignal - ? $this->vm->peekSignal($this->id) - : $this->vm->peekCompletion($this->id); - - return $notification->resultKind === NotificationResult::Failure; + return $this->peek()->resultKind === NotificationResult::Failure; } public function await(): mixed { - $notification = $this->isSignal - ? $this->vm->awaitSignal($this->id) - : $this->vm->awaitCompletion($this->id); + $notification = match (true) { + $this->signalName !== null => $this->vm->awaitNamedSignal($this->signalName), + $this->isSignal => $this->vm->awaitSignal($this->id), + default => $this->vm->awaitCompletion($this->id), + }; return $this->resolve($notification); } @@ -82,11 +102,19 @@ public function await(): mixed /** Resolves an already-ready future without suspending (peeks; does not consume). */ public function take(): mixed { - $notification = $this->isSignal + return $this->resolve($this->peek()); + } + + /** Reads the ready notification from the matching VM table without consuming it. */ + private function peek(): Notification + { + if ($this->signalName !== null) { + return $this->vm->peekNamedSignal($this->signalName); + } + + return $this->isSignal ? $this->vm->peekSignal($this->id) : $this->vm->peekCompletion($this->id); - - return $this->resolve($notification); } private function resolve(Notification $notification): mixed @@ -95,6 +123,7 @@ private function resolve(Notification $notification): mixed NotificationResult::Failure => throw new TerminalException( $notification->failure->message ?? 'terminal failure', $notification->failure->code ?? TerminalException::DEFAULT_CODE, + metadata: $notification->failure->metadata ?? [], ), NotificationResult::Value => $this->decoder !== null ? ($this->decoder)($notification->value ?? '') diff --git a/src/Context/RestateContext.php b/src/Context/RestateContext.php index 2cb886c..f69e823 100644 --- a/src/Context/RestateContext.php +++ b/src/Context/RestateContext.php @@ -12,6 +12,8 @@ use Qcodr\Restate\Sdk\Protocol\Message\CompleteAwakeableCommand; use Qcodr\Restate\Sdk\Protocol\Message\Failure; use Qcodr\Restate\Sdk\Protocol\Message\Header; +use Qcodr\Restate\Sdk\Protocol\Message\SendSignalCommand; +use Qcodr\Restate\Sdk\Protocol\Message\Value; use Qcodr\Restate\Sdk\Serde\Serde; use Qcodr\Restate\Sdk\Serde\SerializationException; use Qcodr\Restate\Sdk\Vm\InvocationInput; @@ -95,7 +97,7 @@ public function run(string $name, callable $action, ?RunOptions $options = null) try { $result = $action(); } catch (TerminalException $e) { - $this->vm->proposeRunCompletionFailure($completionId, new Failure($e->statusCode(), $e->getMessage())); + $this->vm->proposeRunCompletionFailure($completionId, new Failure($e->statusCode(), $e->getMessage(), $e->metadata)); // In request/response, awaiting the just-proposed (not-yet-replayed) // completion writes the suspension and unwinds; in streaming it parks until @@ -125,6 +127,45 @@ public function run(string $name, callable $action, ?RunOptions $options = null) return $this->completionFuture($completionId)->await(); } + public function runAsync(string $name, callable $action): DurableFuture + { + $completionId = $this->vm->sysRun($name); + + // Replay: the result is already journaled, so the closure must NOT re-run. + if ($this->vm->isCompletionReady($completionId)) { + return $this->completionFuture($completionId); + } + + try { + $result = $action(); + } catch (TerminalException $e) { + $this->vm->proposeRunCompletionFailure( + $completionId, + new Failure($e->statusCode(), $e->getMessage(), $e->metadata), + ); + + return $this->completionFuture($completionId); + } + + try { + $serialized = $this->serde->serialize($result); + } catch (SerializationException $e) { + // A non-serializable result fails terminally rather than re-running forever + // (see run()): the RunCommand is journaled, so a missing completion would + // re-execute the closure on every attempt. + $this->vm->proposeRunCompletionFailure( + $completionId, + new Failure(TerminalException::DEFAULT_CODE, 'run result is not serializable: ' . $e->getMessage()), + ); + + return $this->completionFuture($completionId); + } + + $this->vm->proposeRunCompletionSuccess($completionId, $serialized); + + return $this->completionFuture($completionId); + } + /** * Handles a non-terminal failure inside a run closure. * @@ -312,12 +353,21 @@ public function select(DurableFuture ...$futures): array } } - // No future is ready yet: request/response unwinds inside suspendAny(); streaming - // parks until the predicate holds, then the rescan below is guaranteed a winner. - [$completions, $signals] = self::partitionFutures($futures); + // No future is ready yet. A pending cancel must surface as a 409 rather than + // (re-)suspend: in request/response this is the re-invocation carrying the CANCEL + // signal in the replayed journal, where suspending again would loop forever. + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + + // request/response unwinds inside suspendAny(); streaming parks until the + // predicate holds, then either the rescan below finds a winner or — if only the + // cancel guard fired — the invocation is cancelled. + [$completions, $signals, $namedSignals] = self::partitionFutures($futures); $this->vm->suspendAny( $completions, $signals, + $namedSignals, fn (): bool => self::anyReady($futures) || $this->vm->isCancelled(), ); @@ -327,6 +377,10 @@ public function select(DurableFuture ...$futures): array } } + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + throw new LogicException('select resumed without a ready future'); } @@ -334,17 +388,27 @@ public function awaitAll(array $futures): array { $unresolved = self::unresolved($futures); if ($unresolved !== []) { + // A pending cancel surfaces as a 409 rather than (re-)suspend (see select()). + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + // Request/response unwinds inside suspendAll(); streaming parks until every // future is ready (or a cancel arrives), then the guard below holds. - [$completions, $signals] = self::partitionFutures($unresolved); + [$completions, $signals, $namedSignals] = self::partitionFutures($unresolved); $this->vm->suspendAll( $completions, $signals, + $namedSignals, fn (): bool => self::allReady($futures) || $this->vm->isCancelled(), ); } if (!self::allReady($futures)) { + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + throw new LogicException('awaitAll resumed without every future ready'); } @@ -359,10 +423,21 @@ public function awaitAny(DurableFuture ...$futures): mixed $isResolved = fn (): bool => self::anySucceededOrAllReady($futures) || $this->vm->isCancelled(); if (!self::anySucceededOrAllReady($futures)) { + // A pending cancel surfaces as a 409 rather than (re-)suspend (see select()). + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + // Request/response unwinds inside suspendAnySucceeded(); streaming parks until // a future succeeds or every future has resolved, then the scan below settles. - [$completions, $signals] = self::partitionFutures(self::unresolved($futures)); - $this->vm->suspendAnySucceeded($completions, $signals, $isResolved); + [$completions, $signals, $namedSignals] = self::partitionFutures(self::unresolved($futures)); + $this->vm->suspendAnySucceeded($completions, $signals, $namedSignals, $isResolved); + + // Streaming resumed: a cancel that woke the park surfaces as a 409 (cancel + // wins the race here, exactly as StateMachine::awaitCompletion does post-park). + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } } $lastFailure = null; @@ -391,10 +466,21 @@ public function awaitAllSucceeded(array $futures): array $isResolved = fn (): bool => self::anyFailedOrAllSucceeded($futures) || $this->vm->isCancelled(); if (!self::anyFailedOrAllSucceeded($futures)) { + // A pending cancel surfaces as a 409 rather than (re-)suspend (see select()). + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + // Request/response unwinds inside suspendAllSucceeded(); streaming parks until // one future fails or all succeed, then the scan below settles. - [$completions, $signals] = self::partitionFutures(self::unresolved($futures)); - $this->vm->suspendAllSucceeded($completions, $signals, $isResolved); + [$completions, $signals, $namedSignals] = self::partitionFutures(self::unresolved($futures)); + $this->vm->suspendAllSucceeded($completions, $signals, $namedSignals, $isResolved); + + // Streaming resumed: a cancel that woke the park surfaces as a 409 (cancel + // wins the race here, exactly as StateMachine::awaitCompletion does post-park). + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } } foreach ($futures as $future) { @@ -405,6 +491,10 @@ public function awaitAllSucceeded(array $futures): array } if (!self::allReady($futures)) { + if ($this->vm->isCancelled()) { + $this->vm->raiseCancellation(); + } + throw new LogicException('awaitAllSucceeded resumed without resolution'); } @@ -494,6 +584,34 @@ public function rejectAwakeable(string $id, string $message): void ); } + public function createSignal(string $name): DurableFuture + { + // A named signal is addressed by its user-chosen name, so nothing is allocated + // on the VM (unlike an awakeable, which reserves a signal index): the future just + // awaits the named-signal table, filled when another invocation sends to it. + return new DurableFuture( + $this->vm, + 0, + isSignal: true, + decoder: fn (string $bytes): mixed => $this->serde->deserialize($bytes), + signalName: $name, + ); + } + + public function resolveSignal(string $invocationId, string $name, mixed $value = null): void + { + $this->vm->sysSendSignal( + SendSignalCommand::resolveNamed($invocationId, $name, new Value($this->serde->serialize($value))), + ); + } + + public function rejectSignal(string $invocationId, string $name, string $reason): void + { + $this->vm->sysSendSignal( + SendSignalCommand::rejectNamed($invocationId, $name, new Failure(TerminalException::DEFAULT_CODE, $reason)), + ); + } + public function get(string $key): mixed { [$found, $value] = $this->vm->sysGetState($key); @@ -688,23 +806,30 @@ private static function anyFailedOrAllSucceeded(array $futures): bool } /** + * Splits futures into the three await-tree buckets the combinator suspends on: a + * named signal awaits by name (populating `waitingNamedSignals`), an awakeable by + * its signal index, and everything else by its completion id. + * * @param array $futures * - * @return array{0: list, 1: list} [completionIds, signalIds] + * @return array{0: list, 1: list, 2: list} [completionIds, signalIds, namedSignals] */ private static function partitionFutures(array $futures): array { $completions = []; $signals = []; + $namedSignals = []; foreach ($futures as $future) { - if ($future->isSignal()) { + if ($future->isNamedSignal()) { + $namedSignals[] = (string) $future->signalName(); + } elseif ($future->isSignal()) { $signals[] = $future->id(); } else { $completions[] = $future->id(); } } - return [$completions, $signals]; + return [$completions, $signals, $namedSignals]; } /** diff --git a/src/Endpoint/InvocationDriver.php b/src/Endpoint/InvocationDriver.php index 18332df..3b7c756 100644 --- a/src/Endpoint/InvocationDriver.php +++ b/src/Endpoint/InvocationDriver.php @@ -123,8 +123,10 @@ public function driveStreaming( $this->invocationProcessor->process($service, $handler, $vm); }); - // Run to the first park (a ParkSignal) or straight to a terminal frame (null). - $park = $fiber->start(); + // Run to the first park (a ParkSignal) or straight to a terminal frame, then drain + // any await already satisfiable from journal-buffered notifications before we ever + // block on the stream. + $park = $this->drainResolved($fiber, $fiber->start()); while (!$fiber->isTerminated()) { $chunk = $io->read(); @@ -139,16 +141,40 @@ public function driveStreaming( break; } - // Routes late completions/signals (and skips ack/control frames). Resume the - // fiber only once the parked await's own predicate is satisfied, otherwise - // keep feeding frames — a frame that does not make the await resolvable must - // not wake the handler prematurely. + // Routes late completions/signals (and skips ack/control frames), then resumes + // the fiber for every await this chunk satisfies — not just one — before + // blocking on the next read. A single chunk can carry several notifications + // (batched completions, or a completion plus the cancel), and each resumed + // await may run straight on to the next whose result is already present; a + // frame that does not make the current await resolvable still does not wake it. $vm->notifyInput($chunk); - if (!$park instanceof ParkSignal || ($park->isResolved)()) { - $park = $fiber->resume(); - } + $park = $this->drainResolved($fiber, $park); } $io->close(); } + + /** + * Resumes the fiber while the current park's awaited result is already present, so a + * single inbound chunk drives every await it satisfies before the driver blocks on the + * next read. Returns the park the fiber is left on, or null once it terminates. + * + * Each iteration advances `$park = $fiber->resume()` and re-tests the new park, so a + * resumed await either returns, throws (terminating the fiber), or re-parks on a + * still-unresolved await whose predicate is false — the loop always makes progress. + * + * `$park` is typed `mixed` because {@see \Fiber::start()}/{@see \Fiber::resume()} return + * mixed; the suspender only ever yields a {@see ParkSignal} and the fiber body returns + * void, so in practice the value is always `ParkSignal|null`. + * + * @param Fiber $fiber + */ + private function drainResolved(Fiber $fiber, mixed $park): mixed + { + while ($park instanceof ParkSignal && ($park->isResolved)()) { + $park = $fiber->resume(); + } + + return $park; + } } diff --git a/src/Endpoint/InvocationProcessor.php b/src/Endpoint/InvocationProcessor.php index 88e5d43..62f9f45 100644 --- a/src/Endpoint/InvocationProcessor.php +++ b/src/Endpoint/InvocationProcessor.php @@ -83,7 +83,7 @@ public function process(ServiceDefinition $service, HandlerDefinition $handler, } catch (SuspendException) { // The suspension message was already written by the state machine. } catch (TerminalException $e) { - $vm->sysWriteOutputFailure(new Failure($e->statusCode(), $e->getMessage())); + $vm->sysWriteOutputFailure(new Failure($e->statusCode(), $e->getMessage(), $e->metadata)); $vm->sysEnd(); } catch (RetryableException $e) { $this->logger->warning('Invocation attempt failed (retryable): ' . $e->getMessage(), ['exception' => $e]); diff --git a/src/Error/TerminalException.php b/src/Error/TerminalException.php index 0430fde..867df61 100644 --- a/src/Error/TerminalException.php +++ b/src/Error/TerminalException.php @@ -18,8 +18,16 @@ class TerminalException extends RuntimeException { public const DEFAULT_CODE = 500; - public function __construct(string $message, int $code = self::DEFAULT_CODE, ?Throwable $previous = null) - { + /** + * @param array $metadata user error metadata propagated with the + * failure (service protocol V7) + */ + public function __construct( + string $message, + int $code = self::DEFAULT_CODE, + ?Throwable $previous = null, + public readonly array $metadata = [], + ) { parent::__construct($message, $code, $previous); } diff --git a/src/Protocol/Message/AwaitingOnMessage.php b/src/Protocol/Message/AwaitingOnMessage.php new file mode 100644 index 0000000..ede4f46 --- /dev/null +++ b/src/Protocol/Message/AwaitingOnMessage.php @@ -0,0 +1,45 @@ +writeMessage(1, $this->awaitingOn->encode())->toString(); + } + + public function requestedAck(): bool + { + return false; + } +} diff --git a/src/Protocol/Message/Failure.php b/src/Protocol/Message/Failure.php index 84103fb..c684af5 100644 --- a/src/Protocol/Message/Failure.php +++ b/src/Protocol/Message/Failure.php @@ -8,26 +8,40 @@ use Qcodr\Restate\Sdk\Protocol\Protobuf\Writer; /** - * Nested `Failure { uint32 code = 1; string message = 2; }`. + * Nested `Failure { uint32 code = 1; string message = 2; repeated FailureMetadata + * metadata = 3; }` with `FailureMetadata { string key = 1; string value = 2; }`. * - * Carries user-visible terminal failures (an invocation's failure result or a - * failed call result). The repeated `metadata` field (3) is decode-tolerant but - * not produced by this SDK. + * Carries user-visible terminal failures (an invocation's failure result or a failed + * call result). The `metadata` map is round-tripped so user error context propagates + * across calls and to the caller (service protocol V7). */ final class Failure { + /** + * @param array $metadata + */ public function __construct( public readonly int $code, public readonly string $message, + public readonly array $metadata = [], ) { } public function encode(): string { - return (new Writer()) + $writer = (new Writer()) ->writeUint32(1, $this->code) - ->writeString(2, $this->message) - ->toString(); + ->writeString(2, $this->message); + + foreach ($this->metadata as $key => $value) { + $entry = (new Writer()) + ->writeString(1, (string) $key) + ->writeString(2, $value) + ->toString(); + $writer->writeMessage(3, $entry); + } + + return $writer->toString(); } public static function decode(string $bytes): self @@ -35,6 +49,7 @@ public static function decode(string $bytes): self $reader = new Reader($bytes); $code = 0; $message = ''; + $metadata = []; while (!$reader->atEnd()) { [$field, $wire] = $reader->readTag(); switch ($field) { @@ -44,11 +59,37 @@ public static function decode(string $bytes): self case 2: $message = $reader->readLengthDelimited(); break; + case 3: + [$key, $value] = self::decodeMetadataEntry($reader->readLengthDelimited()); + $metadata[$key] = $value; + break; default: $reader->skip($wire); } } - return new self($code, $message); + return new self($code, $message, $metadata); + } + + /** + * @return array{0: string, 1: string} [key, value] + */ + private static function decodeMetadataEntry(string $bytes): array + { + $reader = new Reader($bytes); + $key = ''; + $value = ''; + while (!$reader->atEnd()) { + [$field, $wire] = $reader->readTag(); + if ($field === 1) { + $key = $reader->readLengthDelimited(); + } elseif ($field === 2) { + $value = $reader->readLengthDelimited(); + } else { + $reader->skip($wire); + } + } + + return [$key, $value]; } } diff --git a/src/Protocol/Message/Future.php b/src/Protocol/Message/Future.php index 573c6fc..795c4e2 100644 --- a/src/Protocol/Message/Future.php +++ b/src/Protocol/Message/Future.php @@ -40,6 +40,11 @@ public static function forSignal(int $signalId): self return new self(waitingSignals: [$signalId]); } + public static function forNamedSignal(string $name): self + { + return new self(waitingNamedSignals: [$name]); + } + public function encode(): string { $writer = new Writer(); diff --git a/src/Protocol/Message/SendSignalCommand.php b/src/Protocol/Message/SendSignalCommand.php index a0bdd07..3fae095 100644 --- a/src/Protocol/Message/SendSignalCommand.php +++ b/src/Protocol/Message/SendSignalCommand.php @@ -13,7 +13,9 @@ * index (`idx`) or by a custom `name`, and carries a result (void/value/failure). * * The {@see cancel} factory builds the cancellation signal: built-in CANCEL index 1 - * with a void result. + * with a void result. The {@see resolveNamed} / {@see rejectNamed} factories deliver + * a custom-named signal carrying a value or a failure — the send side of named + * signals (the receive side awaits via {@see Future::forNamedSignal}). */ final class SendSignalCommand implements OutgoingMessage { @@ -26,6 +28,8 @@ private function __construct( public readonly ?string $signalName, public readonly bool $void, public readonly string $name = '', + public readonly ?Value $value = null, + public readonly ?Failure $failure = null, ) { } @@ -35,6 +39,18 @@ public static function cancel(string $targetInvocationId): self return new self($targetInvocationId, self::CANCEL_SIGNAL_INDEX, null, true); } + /** Resolves a custom-named signal on the target invocation with a value. */ + public static function resolveNamed(string $targetInvocationId, string $signalName, Value $value): self + { + return new self($targetInvocationId, null, $signalName, false, '', $value, null); + } + + /** Rejects a custom-named signal on the target invocation with a terminal failure. */ + public static function rejectNamed(string $targetInvocationId, string $signalName, Failure $failure): self + { + return new self($targetInvocationId, null, $signalName, false, '', null, $failure); + } + public function messageType(): MessageType { return MessageType::SendSignalCommand; @@ -50,7 +66,11 @@ public function encode(): string $writer->writeUint32Present(2, $this->signalIdx ?? 0); } - if ($this->void) { + if ($this->value !== null) { + $writer->writeMessage(5, $this->value->encode()); // Value result + } elseif ($this->failure !== null) { + $writer->writeMessage(6, $this->failure->encode()); // Failure result + } elseif ($this->void) { $writer->writeMessage(4, ''); // Void result } diff --git a/src/Server/AmpStreamingServer.php b/src/Server/AmpStreamingServer.php index 748c070..bdd3f5e 100644 --- a/src/Server/AmpStreamingServer.php +++ b/src/Server/AmpStreamingServer.php @@ -7,6 +7,7 @@ use Amp\ByteStream\ReadableIterableStream; use Amp\Http\HttpStatus; use Amp\Http\Server\DefaultErrorHandler; +use Amp\Http\Server\Driver\DefaultHttpDriverFactory; use Amp\Http\Server\Request; use Amp\Http\Server\RequestHandler\ClosureRequestHandler; use Amp\Http\Server\Response; @@ -52,6 +53,34 @@ */ final class AmpStreamingServer { + /** + * Per-stream and whole-connection idle ceilings handed to amphp's HTTP/2 driver, in + * seconds. amphp defaults these to 15s / 60s, which is wrong for this transport: a + * Restate invocation legitimately keeps its bidi stream open and idle while the handler + * is parked awaiting a completion, a signal, or a cancel the runtime may deliver much + * later, and the runtime never half-closes the request (so amphp never `suspend()`s the + * stream timer) while its HTTP/2 keep-alive PINGs only refresh the connection timer. + * At 15s amphp would otherwise `releaseStream(..., "Closing stream due to inactivity")`, + * making the body read throw mid-invocation and silently dropping a pending cancel. + * Raised well above the runtime's own inactivity/abort windows so Restate governs + * suspension; a dead peer is still detected immediately by the socket closing. + */ + private const STREAM_IDLE_TIMEOUT_SECONDS = 3600; + private const CONNECTION_IDLE_TIMEOUT_SECONDS = 3600; + + /** + * Connection / concurrency ceilings handed to amphp. The Restate runtime is a single + * trusted peer that opens one long-lived bidi connection per in-flight invocation, all + * from the same IP, so amphp's defaults (1000 total, 10 per IP, 1000 concurrent) are + * far too low: at ~10 the runtime is denied new connections ("too many existing + * connections"), which surfaces as broken-pipe / unexpected-frame errors and dropped + * invocations under load. Raised so the runtime — not amphp — governs how many + * invocations run at once. + */ + private const CONNECTION_LIMIT = 100_000; + private const CONNECTION_LIMIT_PER_IP = 100_000; + private const CONCURRENCY_LIMIT = 100_000; + private readonly RequestProcessor $processor; private readonly LoggerInterface $logger; @@ -85,9 +114,25 @@ public function listen(string $host = '0.0.0.0', int $port = 9080): void throw new RuntimeException("Port {$port} is out of range (0-65535)"); } - // createForDirectAccess wires up the HTTP/2 driver (incl. h2c prior-knowledge), - // which the Restate runtime uses to open the bidirectional invocation stream. - $server = SocketHttpServer::createForDirectAccess($this->logger); + // The Restate runtime opens the invocation stream with HTTP/2 cleartext (h2c) + // PRIOR KNOWLEDGE — it writes the HTTP/2 connection preface straight onto the + // socket, with no TLS (so no ALPN) and no `Upgrade: h2c` handshake. amphp only + // honours that preface when its HTTP/1 driver is built with HTTP/2 upgrade allowed; + // the default createForDirectAccess factory leaves it off and answers the preface + // with `505 Unsupported version 2.0`. Enable it explicitly so the bidi stream is + // accepted (verified against a real runtime in the conformance suite). + $server = SocketHttpServer::createForDirectAccess( + $this->logger, + connectionLimit: self::CONNECTION_LIMIT, + connectionLimitPerIp: self::CONNECTION_LIMIT_PER_IP, + concurrencyLimit: self::CONCURRENCY_LIMIT, + httpDriverFactory: new DefaultHttpDriverFactory( + $this->logger, + streamTimeout: self::STREAM_IDLE_TIMEOUT_SECONDS, + connectionTimeout: self::CONNECTION_IDLE_TIMEOUT_SECONDS, + allowHttp2Upgrade: true, + ), + ); $server->expose(new InternetAddress($host, $port)); $server->start(new ClosureRequestHandler($this->handleRequest(...)), new DefaultErrorHandler()); diff --git a/src/Vm/AwakeableId.php b/src/Vm/AwakeableId.php index 8c0fa38..6cd5c3b 100644 --- a/src/Vm/AwakeableId.php +++ b/src/Vm/AwakeableId.php @@ -7,13 +7,16 @@ /** * Builds the public awakeable identifier handed to other invocations. * - * Per the protocol, the id is the literal `prom_1` followed by the Base64 URL-safe - * (unpadded) encoding of the invocation id concatenated with the awakeable's signal - * index as a 32-bit big-endian unsigned integer. + * Per the protocol (service protocol V7), the id is the literal `sign_1` followed by + * the Base64 URL-safe (unpadded) encoding of the invocation id concatenated with the + * awakeable's signal index as a 32-bit big-endian unsigned integer. The `sign_` prefix + * marks it as a *signal*-backed awakeable; the older `prom_` prefix denoted the + * completion-backed form, which makes the runtime route a resolution to a non-existent + * completion id — a journal "storage corruption" error that crashes the partition. */ final class AwakeableId { - private const PREFIX = 'prom_1'; + private const PREFIX = 'sign_1'; public static function encode(string $invocationId, int $signalId): string { diff --git a/src/Vm/FiberSuspender.php b/src/Vm/FiberSuspender.php index 3930528..f8b2618 100644 --- a/src/Vm/FiberSuspender.php +++ b/src/Vm/FiberSuspender.php @@ -19,11 +19,18 @@ * The handler runs inside a {@see \Fiber} the driver started; calling this outside * a fiber raises a {@see \FiberError}, which is the correct signal that streaming * was wired up without a driver. + * + * Before parking it announces the await tree to the runtime with an + * {@see \Qcodr\Restate\Sdk\Protocol\Message\AwaitingOnMessage} (V7): on an open bidi + * stream the runtime only pushes a completion/signal once it knows the invocation is + * waiting on it, so without this a parked invocation never receives an external CANCEL + * signal or an awakeable resolved by another invocation. */ final class FiberSuspender implements Suspender { public function park(StateMachine $vm, Future $awaitTree, Closure $isResolved): void { + $vm->writeAwaitingOn($awaitTree); Fiber::suspend(new ParkSignal($awaitTree, $isResolved)); } } diff --git a/src/Vm/StateMachine.php b/src/Vm/StateMachine.php index 51d8b89..a22fb23 100644 --- a/src/Vm/StateMachine.php +++ b/src/Vm/StateMachine.php @@ -8,6 +8,7 @@ use Qcodr\Restate\Sdk\Error\CancelledException; use Qcodr\Restate\Sdk\Protocol\ErrorBehavior; use Qcodr\Restate\Sdk\Protocol\Frame; +use Qcodr\Restate\Sdk\Protocol\Message\AwaitingOnMessage; use Qcodr\Restate\Sdk\Protocol\Message\CallCommand; use Qcodr\Restate\Sdk\Protocol\Message\ClearAllStateCommand; use Qcodr\Restate\Sdk\Protocol\Message\ClearStateCommand; @@ -26,6 +27,7 @@ use Qcodr\Restate\Sdk\Protocol\Message\Header; use Qcodr\Restate\Sdk\Protocol\Message\InputCommand; use Qcodr\Restate\Sdk\Protocol\Message\Notification; +use Qcodr\Restate\Sdk\Protocol\Message\NotificationResult; use Qcodr\Restate\Sdk\Protocol\Message\OneWayCallCommand; use Qcodr\Restate\Sdk\Protocol\Message\OutgoingMessage; use Qcodr\Restate\Sdk\Protocol\Message\OutputCommand; @@ -83,6 +85,25 @@ final class StateMachine private array $completions = []; /** @var array signals keyed by signal index */ private array $signals = []; + /** @var array named signals keyed by their user-chosen name */ + private array $namedSignals = []; + /** + * Invocation-id completion ids of the calls this handler has issued, in order. On + * cancellation they are used to propagate the cancel to those child invocations + * (implicit cancellation), so a cancelled parent tears down the calls it spawned. + * + * @var list + */ + private array $trackedInvocationIdCompletions = []; + /** + * Proposed `ctx.run` results awaiting the runtime's `ProposeRunCompletionAck`, keyed by + * completion id. Over streaming the runtime confirms a proposal with a lightweight ack + * (a control frame, processing-phase only) rather than echoing the value, so the result + * is stashed here at propose time and promoted into the completion table on the ack. + * + * @var array + */ + private array $pendingRunResults = []; private VmState $state = VmState::WaitingPreFlight; @@ -167,6 +188,7 @@ private function tryParse(): void $input = null; $completions = []; $signals = []; + $namedSignals = []; $commandTypes = []; while ($consumed < $start->knownEntries) { @@ -178,7 +200,7 @@ private function tryParse(): void return; // need more bytes } - $this->classifyJournalFrame($frame, $knownCommands, $input, $completions, $signals, $commandTypes); + $this->classifyJournalFrame($frame, $knownCommands, $input, $completions, $signals, $namedSignals, $commandTypes); $consumed++; } @@ -189,6 +211,7 @@ private function tryParse(): void $this->knownCommands = $knownCommands; $this->completions = $completions; $this->signals = $signals; + $this->namedSignals = $namedSignals; $this->journalCommandTypes = $commandTypes; $this->parsed = true; // Drop the (up to 64 MB) parsed journal; keep only any bytes past it. In @@ -217,7 +240,11 @@ private function ingestTrailing(): void Notification::decode($frame->payload), $this->completions, $this->signals, + $this->namedSignals, ); + } elseif ($type === MessageType::ProposeRunCompletionAck) { + // The runtime confirmed a proposed run; resolve it from the stashed value. + $this->resolveProposedRun(Notification::decode($frame->payload)->completionId); } } @@ -226,12 +253,26 @@ private function ingestTrailing(): void } } + /** + * Promotes a proposed `ctx.run` result into the completion table once the runtime acks + * it. The ack frame carries only the completion id (field 1); the value/failure was + * stashed at propose time, so this wakes the parked run with the result it proposed. + */ + private function resolveProposedRun(?int $completionId): void + { + if ($completionId !== null && isset($this->pendingRunResults[$completionId])) { + $this->completions[$completionId] = $this->pendingRunResults[$completionId]; + unset($this->pendingRunResults[$completionId]); + } + } + /** * @param int $knownCommands by-ref command counter * @param InputCommand|null $input by-ref captured input command - * @param array $completions by-ref - * @param array $signals by-ref - * @param list $commandTypes by-ref, appended per command frame + * @param array $completions by-ref + * @param array $signals by-ref + * @param array $namedSignals by-ref + * @param list $commandTypes by-ref, appended per command frame */ private function classifyJournalFrame( Frame $frame, @@ -239,6 +280,7 @@ private function classifyJournalFrame( ?InputCommand &$input, array &$completions, array &$signals, + array &$namedSignals, array &$commandTypes, ): void { $type = $frame->type(); @@ -256,7 +298,7 @@ private function classifyJournalFrame( return; } if ($type !== null && $type->isNotification()) { - $this->routeNotification(Notification::decode($frame->payload), $completions, $signals); + $this->routeNotification(Notification::decode($frame->payload), $completions, $signals, $namedSignals); return; } @@ -265,13 +307,26 @@ private function classifyJournalFrame( } /** - * @param array $completions by-ref - * @param array $signals by-ref + * Routes a decoded notification into its table. The id fields are a protocol oneof — + * a completion id, a built-in/awakeable signal index, or a user-chosen signal name — + * so exactly one branch applies. Named signals (signal_name set) are keyed by name + * rather than index, since the receive side addresses them by the name another + * invocation sent to. + * + * @param array $completions by-ref + * @param array $signals by-ref + * @param array $namedSignals by-ref */ - private function routeNotification(Notification $notification, array &$completions, array &$signals): void - { + private function routeNotification( + Notification $notification, + array &$completions, + array &$signals, + array &$namedSignals, + ): void { if ($notification->completionId !== null) { $completions[$notification->completionId] = $notification; + } elseif ($notification->signalName !== null) { + $namedSignals[$notification->signalName] = $notification; } elseif ($notification->signalId !== null) { $signals[$notification->signalId] = $notification; } @@ -417,6 +472,8 @@ public function sysCall( $idempotencyKey, $headers, )); + // Remember the callee so a cancel of this handler propagates to it. + $this->trackedInvocationIdCompletions[] = $invocationIdCompletionId; return [$invocationIdCompletionId, $resultCompletionId]; } @@ -499,16 +556,36 @@ public function sysRejectPromise(string $key, Failure $failure): void public function proposeRunCompletionSuccess(int $completionId, string $value): void { $this->appendOutput(ProposeRunCompletion::success($completionId, $value)); + $this->pendingRunResults[$completionId] = new Notification( + $completionId, + null, + null, + NotificationResult::Value, + $value, + null, + null, + null, + ); } public function proposeRunCompletionFailure(int $completionId, Failure $failure): void { $this->appendOutput(ProposeRunCompletion::failure($completionId, $failure)); + $this->pendingRunResults[$completionId] = new Notification( + $completionId, + null, + null, + NotificationResult::Failure, + null, + $failure, + null, + null, + ); } /** * Creates an awakeable: a signal slot plus the public id another invocation can - * use to complete it. The id is `prom_1` + base64url(invocationId ++ uint32be(idx)). + * use to complete it. The id is `sign_1` + base64url(invocationId ++ uint32be(idx)). * * @return array{0: string, 1: int} [awakeableId, signalId] */ @@ -526,6 +603,17 @@ public function sysCompleteAwakeable(CompleteAwakeableCommand $command): void $this->recordCommand($command); } + /** + * Sends a named signal to another invocation (resolve or reject), journaling the + * command. Mirrors {@see sysCompleteAwakeable}: the {@see SendSignalCommand} carries + * the target invocation id, the signal name and the value/failure result. + */ + public function sysSendSignal(SendSignalCommand $command): void + { + $this->ensureParsed(); + $this->recordCommand($command); + } + /** Cancels another invocation by sending it the built-in CANCEL signal. */ public function sysCancel(string $invocationId): void { @@ -568,6 +656,26 @@ public function isCancelled(): bool return isset($this->signals[self::CANCEL_SIGNAL_ID]); } + /** + * Fails the current await with {@see CancelledException}, first propagating the cancel + * to the calls this handler issued (implicit cancellation): every tracked callee whose + * invocation id is already known is sent the built-in CANCEL signal, so a cancelled + * parent tears down the children it is blocked on rather than leaking them. Mirrors the + * cancellation branch of sdk-shared-core's `do_await`. + */ + public function raiseCancellation(): never + { + foreach ($this->trackedInvocationIdCompletions as $completionId) { + $invocationId = ($this->completions[$completionId] ?? null)?->invocationId; + if ($invocationId !== null) { + $this->recordCommand(SendSignalCommand::cancel($invocationId)); + } + } + $this->trackedInvocationIdCompletions = []; + + throw new CancelledException(); + } + /** Returns the completion if ready, otherwise parks (or fails if cancelled). */ public function awaitCompletion(int $completionId): Notification { @@ -575,7 +683,7 @@ public function awaitCompletion(int $completionId): Notification return $this->completions[$completionId]; } if ($this->isCancelled()) { - throw new CancelledException(); + $this->raiseCancellation(); } // Request/response parks by throwing (the lines below are unreachable there); @@ -587,7 +695,7 @@ public function awaitCompletion(int $completionId): Notification ); if ($this->isCancelled()) { - throw new CancelledException(); // cancel won the race + $this->raiseCancellation(); // cancel won the race } return $this->peekCompletion($completionId); // the driver guarantees its presence @@ -604,7 +712,7 @@ public function awaitSignal(int $signalId): Notification return $this->signals[$signalId]; } if ($this->isCancelled()) { - throw new CancelledException(); + $this->raiseCancellation(); } $this->parkOn( @@ -613,7 +721,7 @@ public function awaitSignal(int $signalId): Notification ); if ($this->isCancelled()) { - throw new CancelledException(); // cancel won the race + $this->raiseCancellation(); // cancel won the race } return $this->peekSignal($signalId); // the driver guarantees its presence @@ -629,18 +737,56 @@ public function peekSignal(int $signalId): Notification return $this->signals[$signalId]; } + public function isNamedSignalReady(string $name): bool + { + return isset($this->namedSignals[$name]); + } + + /** Returns the named signal if ready, otherwise parks (or fails if cancelled). Mirrors {@see awaitSignal}. */ + public function awaitNamedSignal(string $name): Notification + { + if (isset($this->namedSignals[$name])) { + return $this->namedSignals[$name]; + } + if ($this->isCancelled()) { + $this->raiseCancellation(); + } + + $this->parkOn( + Future::forNamedSignal($name), + fn (): bool => isset($this->namedSignals[$name]) || $this->isCancelled(), + ); + + if ($this->isCancelled()) { + $this->raiseCancellation(); // cancel won the race + } + + return $this->peekNamedSignal($name); // the driver guarantees its presence + } + + /** Reads a ready named signal without consuming it (non-destructive; see {@see peekSignal}). */ + public function peekNamedSignal(string $name): Notification + { + if (!isset($this->namedSignals[$name])) { + throw new ProtocolException("Named signal {$name} is not available"); + } + + return $this->namedSignals[$name]; + } + /** * Parks awaiting the first of several results to complete (race semantics). * * @param list $completionIds * @param list $signalIds + * @param list $namedSignals * @param Closure(): bool $isResolved the combinator's readiness predicate, supplied * by the caller; the streaming driver resumes only * once it holds */ - public function suspendAny(array $completionIds, array $signalIds, Closure $isResolved): void + public function suspendAny(array $completionIds, array $signalIds, array $namedSignals, Closure $isResolved): void { - $this->parkOn(new Future($completionIds, $signalIds, [], [], CombinatorType::FirstCompleted), $isResolved); + $this->parkOn(new Future($completionIds, $signalIds, $namedSignals, [], CombinatorType::FirstCompleted), $isResolved); } /** @@ -648,11 +794,12 @@ public function suspendAny(array $completionIds, array $signalIds, Closure $isRe * * @param list $completionIds * @param list $signalIds + * @param list $namedSignals * @param Closure(): bool $isResolved */ - public function suspendAll(array $completionIds, array $signalIds, Closure $isResolved): void + public function suspendAll(array $completionIds, array $signalIds, array $namedSignals, Closure $isResolved): void { - $this->parkOn(new Future($completionIds, $signalIds, [], [], CombinatorType::AllCompleted), $isResolved); + $this->parkOn(new Future($completionIds, $signalIds, $namedSignals, [], CombinatorType::AllCompleted), $isResolved); } /** @@ -662,11 +809,12 @@ public function suspendAll(array $completionIds, array $signalIds, Closure $isRe * * @param list $completionIds * @param list $signalIds + * @param list $namedSignals * @param Closure(): bool $isResolved */ - public function suspendAnySucceeded(array $completionIds, array $signalIds, Closure $isResolved): void + public function suspendAnySucceeded(array $completionIds, array $signalIds, array $namedSignals, Closure $isResolved): void { - $this->parkOn(new Future($completionIds, $signalIds, [], [], CombinatorType::FirstSucceededOrAllFailed), $isResolved); + $this->parkOn(new Future($completionIds, $signalIds, $namedSignals, [], CombinatorType::FirstSucceededOrAllFailed), $isResolved); } /** @@ -676,11 +824,12 @@ public function suspendAnySucceeded(array $completionIds, array $signalIds, Clos * * @param list $completionIds * @param list $signalIds + * @param list $namedSignals * @param Closure(): bool $isResolved */ - public function suspendAllSucceeded(array $completionIds, array $signalIds, Closure $isResolved): void + public function suspendAllSucceeded(array $completionIds, array $signalIds, array $namedSignals, Closure $isResolved): void { - $this->parkOn(new Future($completionIds, $signalIds, [], [], CombinatorType::AllSucceededOrFirstFailed), $isResolved); + $this->parkOn(new Future($completionIds, $signalIds, $namedSignals, [], CombinatorType::AllSucceededOrFirstFailed), $isResolved); } /** @@ -702,12 +851,32 @@ public function suspendAllSucceeded(array $completionIds, array $signalIds, Clos */ private function parkOn(Future $inner, Closure $isResolved): void { - $awaitOn = new Future( + $this->suspender->park($this, $this->guardWithCancel($inner), $isResolved); + } + + /** + * Wraps an awaited future as `FirstCompleted([inner, CANCEL signal])` so the runtime + * wakes a suspended invocation when it is cancelled. Matching the canonical encoding + * matters: a single-leaf await (one completion or signal, no combinator) flattens its + * ids up next to the CANCEL signal — the runtime keys its cancel wake-up off that flat + * shape — while a real combinator stays nested under the guard. + */ + private function guardWithCancel(Future $inner): Future + { + if ($inner->combinatorType === CombinatorType::Unknown && $inner->nestedFutures === []) { + return new Future( + waitingCompletions: $inner->waitingCompletions, + waitingSignals: [...$inner->waitingSignals, self::CANCEL_SIGNAL_ID], + waitingNamedSignals: $inner->waitingNamedSignals, + combinatorType: CombinatorType::FirstCompleted, + ); + } + + return new Future( waitingSignals: [self::CANCEL_SIGNAL_ID], nestedFutures: [$inner], combinatorType: CombinatorType::FirstCompleted, ); - $this->suspender->park($this, $awaitOn, $isResolved); } /** @@ -721,6 +890,18 @@ public function writeSuspension(Future $awaitTree): void $this->state = VmState::Closed; } + /** + * Streaming only: announces the current await tree to the runtime with an + * {@see AwaitingOnMessage} so it pushes the awaited completions/signals — including + * external ones the SDK cannot pull, like the CANCEL signal or an awakeable another + * invocation resolves — onto the open bidi stream. Unlike {@see writeSuspension} it + * leaves the VM open: the handler stays parked until the driver feeds a result. + */ + public function writeAwaitingOn(Future $awaitTree): void + { + $this->appendOutput(new AwaitingOnMessage($awaitTree)); + } + // --- Termination --- public function sysWriteOutputSuccess(string $value): void diff --git a/tests/Support/Fixtures/AwakeableService.php b/tests/Support/Fixtures/AwakeableService.php new file mode 100644 index 0000000..bf6ba1d --- /dev/null +++ b/tests/Support/Fixtures/AwakeableService.php @@ -0,0 +1,27 @@ +awakeable()->await(); + + return \is_string($value) ? $value : ''; + } +} diff --git a/tests/Support/Fixtures/CallOptionsService.php b/tests/Support/Fixtures/CallOptionsService.php index 664bd1a..f18083c 100644 --- a/tests/Support/Fixtures/CallOptionsService.php +++ b/tests/Support/Fixtures/CallOptionsService.php @@ -40,6 +40,21 @@ public function callAndReturnInvocationId(Context $ctx): string return \is_string($invocationId) ? $invocationId : ''; } + /** + * Awaits the callee's invocation id (completion 1) and then its result (completion 2) + * in sequence, returning the result. Used to prove the streaming driver drains both + * awaits when the runtime batches both completions into a single inbound chunk. + */ + #[Handler] + public function callAwaitIdThenResult(Context $ctx): string + { + $handle = $ctx->serviceCallHandle('Target', 'receive', 'payload'); + $handle->invocationId()->await(); + $result = $handle->result()->await(); + + return \is_string($result) ? $result : ''; + } + /** * Echoes the request metadata captured from the StartMessage / InputCommand. * diff --git a/tests/Support/Fixtures/CancellationService.php b/tests/Support/Fixtures/CancellationService.php index 35d1076..f83d505 100644 --- a/tests/Support/Fixtures/CancellationService.php +++ b/tests/Support/Fixtures/CancellationService.php @@ -5,15 +5,17 @@ namespace Qcodr\Restate\Sdk\Tests\Support\Fixtures; use Qcodr\Restate\Sdk\Context\Context; +use Qcodr\Restate\Sdk\Error\CancelledException; use Qcodr\Restate\Sdk\Error\RetryableException; use Qcodr\Restate\Sdk\Service\Attribute\Handler; use Qcodr\Restate\Sdk\Service\Attribute\Service; /** * Fixture exercising durable-error tuning and cancellation: a handler that throws a - * pausing {@see RetryableException}, one that cancels another invocation, and one - * that awaits a never-arriving timer (so a delivered CANCEL signal surfaces as a - * terminal 409). + * pausing {@see RetryableException}, one that cancels another invocation, one that + * awaits a never-arriving timer (so a delivered CANCEL signal surfaces as a terminal + * 409), and the four combinators reached while a CANCEL is pending (each must surface a + * 409 rather than re-suspend or throw a misleading error). */ #[Service] final class CancellationService @@ -42,4 +44,59 @@ public function awaitThenSleep(Context $ctx): string return 'slept'; } + + /** Races a never-completing call; a pending CANCEL must surface as a 409, not a suspension. */ + #[Handler] + public function raceWhileCancelled(Context $ctx): string + { + $ctx->awaitAny($ctx->serviceCallAsync('Backend', 'never')); + + return 'done'; + } + + /** select() over a never-completing call; a pending CANCEL must surface as a 409. */ + #[Handler] + public function selectWhileCancelled(Context $ctx): string + { + $ctx->select($ctx->serviceCallAsync('Backend', 'never')); + + return 'done'; + } + + /** awaitAll() over a never-completing call; a pending CANCEL must surface as a 409. */ + #[Handler] + public function awaitAllWhileCancelled(Context $ctx): string + { + $ctx->awaitAll([$ctx->serviceCallAsync('Backend', 'never')]); + + return 'done'; + } + + /** awaitAllSucceeded() over a never-completing call; a pending CANCEL must surface as a 409. */ + #[Handler] + public function awaitAllSucceededWhileCancelled(Context $ctx): string + { + $ctx->awaitAllSucceeded([$ctx->serviceCallAsync('Backend', 'never')]); + + return 'done'; + } + + /** + * Observes a cancel at the sleep await (CancelledException), then reaches a combinator. + * The combinator must also surface the still-pending cancel as a 409 rather than + * re-parking — proves the driver drains a re-park whose predicate already holds. + */ + #[Handler] + public function raceAfterObservedCancel(Context $ctx): string + { + try { + $ctx->sleep(60.0); + } catch (CancelledException) { + // observed; fall through to a combinator that is still cancelled + } + + $ctx->awaitAny($ctx->serviceCallAsync('Backend', 'never')); + + return 'done'; + } } diff --git a/tests/Support/JournalBuilder.php b/tests/Support/JournalBuilder.php index 41fcede..6998925 100644 --- a/tests/Support/JournalBuilder.php +++ b/tests/Support/JournalBuilder.php @@ -145,6 +145,59 @@ public function cancelSignal(): self return $this; } + /** + * Adds a signal notification carrying a value (notification field 5), as the runtime + * delivers when an awakeable is resolved by another invocation. Awakeable signals + * start at idx 17 (built-in signals reserve 0..16); mirrors {@see cancelSignal} but + * with a value payload instead of a void cancel. + */ + public function awakeableSignal(int $signalId, string $value): self + { + $payload = (new Writer()) + ->writeUint32Present(2, $signalId) + ->writeMessage(5, (new Writer())->writeBytes(1, $value)->toString()) + ->toString(); + $this->journal[] = [MessageType::SignalNotification, $payload]; + + return $this; + } + + /** + * Adds a named-signal notification carrying a value: signal_name (field 3) plus a + * value (field 5), as the runtime delivers when another invocation sends to + * (targetInvocationId, name). Mirrors {@see awakeableSignal} but keyed by name + * instead of signal index. + */ + public function namedSignal(string $name, string $value): self + { + $payload = (new Writer()) + ->writeStringPresent(3, $name) + ->writeMessage(5, (new Writer())->writeBytes(1, $value)->toString()) + ->toString(); + $this->journal[] = [MessageType::SignalNotification, $payload]; + + return $this; + } + + /** + * Adds a named-signal notification carrying a {@see \Qcodr\Restate\Sdk\Protocol\Message\Failure} + * (field 6), as the runtime delivers when a named signal is rejected. + */ + public function failedNamedSignal(string $name, string $message, int $code = 500): self + { + $failure = (new Writer()) + ->writeUint32(1, $code) + ->writeString(2, $message) + ->toString(); + $payload = (new Writer()) + ->writeStringPresent(3, $name) + ->writeMessage(6, $failure) + ->toString(); + $this->journal[] = [MessageType::SignalNotification, $payload]; + + return $this; + } + public function runCompletion(int $completionId, string $value): self { $payload = (new Writer()) diff --git a/tests/Unit/Context/RestateContextTest.php b/tests/Unit/Context/RestateContextTest.php index 905b823..3847ae2 100644 --- a/tests/Unit/Context/RestateContextTest.php +++ b/tests/Unit/Context/RestateContextTest.php @@ -658,7 +658,7 @@ public function testAwakeableExposesPublicId(): void $awakeable = $ctx->awakeable(); self::assertInstanceOf(Awakeable::class, $awakeable); - self::assertStringStartsWith('prom_1', $awakeable->id()); + self::assertStringStartsWith('sign_1', $awakeable->id()); } public function testAwakeableAwaitSuspendsAsSignal(): void @@ -675,10 +675,12 @@ public function testAwakeableAwaitSuspendsAsSignal(): void } // An awakeable is completed via a signal, so the await tree must wait on a - // signal id, never a completion id. - $inner = self::innerAwaitTree($this->frames($vm)); - self::assertSame([], $inner['completions']); - self::assertNotSame([], $inner['signals']); + // signal id, never a completion id. A single-signal await flattens next to the + // CANCEL signal, so the signals live on the outer (cancel-guarded) node. + $outer = self::outerAwaitTree($this->frames($vm)); + self::assertSame([], $outer['completions']); + self::assertNotSame([], $outer['signals']); + self::assertSame([], $outer['nested'], 'a single await flattens rather than nesting'); } public function testResolveAndRejectAwakeableEmitCompletionCommands(): void @@ -1048,15 +1050,25 @@ private static function proposedFailure(array $frames): Failure * @return array{completions: list, signals: list, named: list, nested: list, combinator: int} */ private static function innerAwaitTree(array $frames): array + { + $outer = self::outerAwaitTree($frames); + self::assertCount(1, $outer['nested']); + + return self::decodeFuture($outer['nested'][0]); + } + + /** + * @param list<\Qcodr\Restate\Sdk\Protocol\Frame> $frames + * + * @return array{completions: list, signals: list, nested: list} + */ + private static function outerAwaitTree(array $frames): array { $suspension = self::frameOfType($frames, MessageType::Suspension); $outerBytes = self::fields($suspension->payload)[4]; self::assertIsString($outerBytes); - $outer = self::decodeFuture($outerBytes); - self::assertCount(1, $outer['nested']); - - return self::decodeFuture($outer['nested'][0]); + return self::decodeFuture($outerBytes); } /** diff --git a/tests/Unit/Context/RunAsyncAndNamedSignalsTest.php b/tests/Unit/Context/RunAsyncAndNamedSignalsTest.php new file mode 100644 index 0000000..cbd1215 --- /dev/null +++ b/tests/Unit/Context/RunAsyncAndNamedSignalsTest.php @@ -0,0 +1,369 @@ +build( + (new JournalBuilder()) + ->input('1') + ->command(MessageType::RunCommand) + ->runCompletion(1, '"stored"'), + ); + + $invoked = false; + $future = $ctx->runAsync('step', static function () use (&$invoked): string { + $invoked = true; + + return 'fresh'; + }); + + self::assertInstanceOf(DurableFuture::class, $future); + self::assertSame('stored', $future->await()); + self::assertFalse($invoked, 'a replayed run must not re-execute its action'); + } + + public function testRunAsyncLiveProposesSuccessAndReturnsAFutureWithoutSuspending(): void + { + [$vm, $ctx] = $this->build((new JournalBuilder())->input('1')); + + $future = $ctx->runAsync('step', static fn (): string => 'value'); + + self::assertInstanceOf(DurableFuture::class, $future); + + // runAsync proposes but does NOT await: no suspension is emitted by it. + $frames = $this->frames($vm); + self::assertSame( + [MessageType::RunCommand, MessageType::ProposeRunCompletion], + $this->typesOf($frames), + ); + + $proposal = self::fields(self::frameOfType($frames, MessageType::ProposeRunCompletion)->payload); + self::assertSame('"value"', $proposal[14]); + self::assertArrayNotHasKey(15, $proposal, 'a success proposal carries no failure'); + } + + public function testRunAsyncWithTerminalExceptionProposesFailureWithoutSuspending(): void + { + [$vm, $ctx] = $this->build((new JournalBuilder())->input('1')); + + $ctx->runAsync('step', static fn (): mixed => throw new TerminalException('boom', 418)); + + $frames = $this->frames($vm); + self::assertSame( + [MessageType::RunCommand, MessageType::ProposeRunCompletion], + $this->typesOf($frames), + ); + + $proposal = self::fields(self::frameOfType($frames, MessageType::ProposeRunCompletion)->payload); + $failureBytes = $proposal[15]; + self::assertIsString($failureBytes); + $failure = Failure::decode($failureBytes); + self::assertSame(418, $failure->code); + self::assertSame('boom', $failure->message); + } + + // --- createSignal() (receive) --- + + public function testCreateSignalReturnsANamedSignalFuture(): void + { + [, $ctx] = $this->build((new JournalBuilder())->input('1')); + + $future = $ctx->createSignal('my-signal'); + + self::assertTrue($future->isNamedSignal()); + self::assertSame('my-signal', $future->signalName()); + } + + public function testCreateSignalReplayResolvesToDeserializedValue(): void + { + [, $ctx] = $this->build( + (new JournalBuilder())->input('1')->namedSignal('my-signal', '"hello"'), + ); + + self::assertSame('hello', $ctx->createSignal('my-signal')->await()); + } + + public function testCreateSignalSuspendsWhenTheNamedSignalIsAbsent(): void + { + [$vm, $ctx] = $this->build((new JournalBuilder())->input('1')); + + try { + $ctx->createSignal('pending')->await(); + self::fail('expected suspension on an unresolved named signal'); + } catch (SuspendException) { + // expected + } + + // The await tree waits on the named signal (flattened next to the CANCEL signal), + // never on a completion or signal index. + $tree = self::awaitTree($this->frames($vm)); + self::assertSame(['pending'], $tree['named']); + self::assertSame([self::CANCEL_SIGNAL_ID], $tree['signals']); + self::assertSame([], $tree['completions']); + } + + // --- resolveSignal() / rejectSignal() (send) --- + + public function testResolveSignalEmitsSendSignalCommandWithNameAndValue(): void + { + [$vm, $ctx] = $this->build((new JournalBuilder())->input('1')); + + $ctx->resolveSignal('inv-target', 'my-signal', 'val'); + + $frames = $this->frames($vm); + self::assertSame([MessageType::SendSignalCommand], $this->typesOf($frames)); + + $signal = self::fields($frames[0]->payload); + self::assertSame('inv-target', $signal[1], 'target invocation id in field 1'); + self::assertSame('my-signal', $signal[3], 'signal name in field 3'); + self::assertArrayNotHasKey(2, $signal, 'a named signal must not emit the built-in idx'); + + $valueBytes = $signal[5]; + self::assertIsString($valueBytes); + self::assertSame('"val"', Value::decode($valueBytes)->content); + } + + public function testRejectSignalEmitsSendSignalCommandWithNameAndFailure(): void + { + [$vm, $ctx] = $this->build((new JournalBuilder())->input('1')); + + $ctx->rejectSignal('inv-target', 'my-signal', 'because'); + + $frames = $this->frames($vm); + self::assertSame([MessageType::SendSignalCommand], $this->typesOf($frames)); + + $signal = self::fields($frames[0]->payload); + self::assertSame('inv-target', $signal[1]); + self::assertSame('my-signal', $signal[3]); + self::assertArrayNotHasKey(5, $signal, 'a reject carries no value'); + + $failureBytes = $signal[6]; + self::assertIsString($failureBytes); + $failure = Failure::decode($failureBytes); + self::assertSame(TerminalException::DEFAULT_CODE, $failure->code); + self::assertSame('because', $failure->message); + } + + // --- combinator partitioning --- + + public function testSelectCarriesANamedSignalFutureIntoTheNamedSignalBucket(): void + { + [$vm, $ctx] = $this->build((new JournalBuilder())->input('1')); + $signal = $ctx->createSignal('combined'); + + try { + $ctx->select($signal); + self::fail('expected suspension'); + } catch (SuspendException) { + // expected + } + + // partitionFutures must route the named signal into waitingNamedSignals (field 3 + // of the inner combinator node), not into completions or signals. + $inner = self::innerAwaitTree($this->frames($vm)); + self::assertSame(['combined'], $inner['named']); + self::assertSame([], $inner['completions']); + self::assertSame([], $inner['signals']); + } + + // --- Helpers --- + + /** + * @return array{0: StateMachine, 1: RestateContext} + */ + private function build(JournalBuilder $builder): array + { + $vm = new StateMachine(ServiceProtocolVersion::V7); + $vm->notifyInput($builder->build()); + $vm->notifyInputClosed(); + $input = $vm->sysInput(); + + $ctx = new RestateContext( + $vm, + $input, + new JsonSerde(), + new SystemClock(), + ContextRand::fromSeed($input->randomSeed), + writable: true, + logger: new NullLogger(), + ); + + return [$vm, $ctx]; + } + + /** + * @return list + */ + private function frames(StateMachine $vm): array + { + return MessageCodec::decodeAll($vm->takeOutput()); + } + + /** + * @param list $frames + * + * @return list + */ + private function typesOf(array $frames): array + { + return \array_map(static fn (Frame $frame): ?MessageType => $frame->type(), $frames); + } + + /** + * @param list $frames + */ + private static function frameOfType(array $frames, MessageType $type): Frame + { + foreach ($frames as $frame) { + if ($frame->type() === $type) { + return $frame; + } + } + + self::fail(\sprintf('no %s frame was emitted', $type->name)); + } + + /** + * The outer (cancel-guarded) await tree carried in the suspension frame. + * + * @param list $frames + * + * @return array{completions: list, signals: list, named: list, nested: list, combinator: int} + */ + private static function awaitTree(array $frames): array + { + $suspension = self::frameOfType($frames, MessageType::Suspension); + $treeBytes = self::fields($suspension->payload)[4]; + self::assertIsString($treeBytes); + + return self::decodeFuture($treeBytes); + } + + /** + * The single nested await point, peeling off the outer cancel-aware wrapper. + * + * @param list $frames + * + * @return array{completions: list, signals: list, named: list, nested: list, combinator: int} + */ + private static function innerAwaitTree(array $frames): array + { + $outer = self::awaitTree($frames); + self::assertCount(1, $outer['nested']); + + return self::decodeFuture($outer['nested'][0]); + } + + /** + * @return array + */ + private static function fields(string $payload): array + { + $reader = new Reader($payload); + $fields = []; + while (!$reader->atEnd()) { + [$field, $wire] = $reader->readTag(); + if ($wire === WireType::VARINT) { + $fields[$field] = $reader->readVarint(); + } elseif ($wire === WireType::LENGTH_DELIMITED) { + $fields[$field] = $reader->readLengthDelimited(); + } else { + $reader->skip($wire); + } + } + + return $fields; + } + + /** + * @return array{completions: list, signals: list, named: list, nested: list, combinator: int} + */ + private static function decodeFuture(string $bytes): array + { + $reader = new Reader($bytes); + $completions = []; + $signals = []; + $named = []; + $nested = []; + $combinator = 0; + while (!$reader->atEnd()) { + [$field, $wire] = $reader->readTag(); + switch ($field) { + case 1: + $completions = self::unpackVarints($reader->readLengthDelimited()); + break; + case 2: + $signals = self::unpackVarints($reader->readLengthDelimited()); + break; + case 3: + $named[] = $reader->readLengthDelimited(); + break; + case 4: + $nested[] = $reader->readLengthDelimited(); + break; + case 5: + $combinator = $reader->readVarint(); + break; + default: + $reader->skip($wire); + } + } + + return [ + 'completions' => $completions, + 'signals' => $signals, + 'named' => $named, + 'nested' => $nested, + 'combinator' => $combinator, + ]; + } + + /** + * @return list + */ + private static function unpackVarints(string $packed): array + { + $reader = new Reader($packed); + $values = []; + while (!$reader->atEnd()) { + $values[] = $reader->readVarint(); + } + + return $values; + } +} diff --git a/tests/Unit/Endpoint/InvocationDriverTest.php b/tests/Unit/Endpoint/InvocationDriverTest.php index 87eb5c2..2b1d2ad 100644 --- a/tests/Unit/Endpoint/InvocationDriverTest.php +++ b/tests/Unit/Endpoint/InvocationDriverTest.php @@ -4,6 +4,7 @@ namespace Qcodr\Restate\Sdk\Tests\Unit\Endpoint; +use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use Qcodr\Restate\Sdk\Endpoint\Endpoint; use Qcodr\Restate\Sdk\Endpoint\InvocationDriver; @@ -19,6 +20,7 @@ use Qcodr\Restate\Sdk\Service\HandlerDefinition; use Qcodr\Restate\Sdk\Service\ServiceDefinition; use Qcodr\Restate\Sdk\Tests\Support\BufferedStreamTransport; +use Qcodr\Restate\Sdk\Tests\Support\Fixtures\AwakeableService; use Qcodr\Restate\Sdk\Tests\Support\Fixtures\CallOptionsService; use Qcodr\Restate\Sdk\Tests\Support\Fixtures\CancellationService; use Qcodr\Restate\Sdk\Tests\Support\Fixtures\Greeter; @@ -111,13 +113,14 @@ public function testLateCompletionResolvesParkedCallWithoutSuspending(): void ); $output = $transport->written(); - self::assertSame([MessageType::CallCommand, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); - self::assertNotContains(MessageType::Suspension, $this->frameTypes($output), 'a parked await must not write a suspension'); + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($output), 'a parked await announces an AwaitingOn, not a suspension'); self::assertSame('"inv-xyz"', $this->successValue($output)); self::assertTrue($transport->isClosed(), 'the driver closes the channel after the terminal frame'); - // Read #1 is the completion read; the CallCommand was already written by then. - self::assertSame([MessageType::CallCommand], $this->frameTypes($transport->outputAtRead(1))); + // Read #1 is the completion read; the CallCommand and the park's AwaitingOn were + // already written by then. + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn], $this->frameTypes($transport->outputAtRead(1))); } public function testSpuriousFrameDoesNotResumeParkedHandlerPrematurely(): void @@ -140,18 +143,18 @@ public function testSpuriousFrameDoesNotResumeParkedHandlerPrematurely(): void ); $output = $transport->written(); - self::assertSame([MessageType::CallCommand, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); self::assertNotContains(MessageType::Suspension, $this->frameTypes($output)); // The awaited id (not the spurious frame's value) is returned: had the spurious // frame resumed the now straight-line await, it would have observed an absent // completion and failed instead of returning this value. self::assertSame('"inv-xyz"', $this->successValue($output)); - // The spurious frame (read #1) and the resolving frame (read #2) were both - // consumed while only the CallCommand had been written: the handler stayed parked - // across the spurious frame and emitted its terminal output only after read #2. - self::assertSame([MessageType::CallCommand], $this->frameTypes($transport->outputAtRead(1))); - self::assertSame([MessageType::CallCommand], $this->frameTypes($transport->outputAtRead(2))); + // The spurious frame (read #1) and the resolving frame (read #2) were both consumed + // while only the CallCommand and the park's AwaitingOn had been written: the handler + // stayed parked across the spurious frame and emitted its output only after read #2. + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn], $this->frameTypes($transport->outputAtRead(1))); + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn], $this->frameTypes($transport->outputAtRead(2))); } public function testPromptCancelTurnsParkedAwaitIntoTerminal409(): void @@ -167,7 +170,7 @@ public function testPromptCancelTurnsParkedAwaitIntoTerminal409(): void ); $output = $transport->written(); - self::assertSame([MessageType::SleepCommand, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); + self::assertSame([MessageType::SleepCommand, MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); self::assertNotContains(MessageType::Suspension, $this->frameTypes($output)); $failure = $this->failure($output); @@ -175,7 +178,7 @@ public function testPromptCancelTurnsParkedAwaitIntoTerminal409(): void self::assertSame('cancelled', $failure->message); } - public function testRunWithAckResolvesOnRunCompletion(): void + public function testRunResolvesFromProposeRunCompletionAck(): void { $service = new RunService(); $endpoint = Endpoint::builder()->bind($service)->build(); @@ -186,9 +189,10 @@ public function testRunWithAckResolvesOnRunCompletion(): void $transport = new BufferedStreamTransport([ (new JournalBuilder())->input('')->build(), - // The ack control frame is read and ignored; the value arrives separately. + // Over streaming the runtime confirms a proposed run with only this ack control + // frame (processing phase) — it does NOT echo a RunCompletion notification. The + // SDK resolves the parked run from the value it proposed. (new JournalBuilder())->proposeRunCompletionAck(1)->frames(), - (new JournalBuilder())->runCompletion(1, '"effect-result"')->frames(), ]); $vm = new StateMachine(ServiceProtocolVersion::V7, new FiberSuspender(), new StreamingOutputSink($transport)); @@ -199,6 +203,7 @@ public function testRunWithAckResolvesOnRunCompletion(): void self::assertSame([ MessageType::RunCommand, MessageType::ProposeRunCompletion, + MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End, ], $this->frameTypes($output)); @@ -224,9 +229,9 @@ public function testHandlerCompletingWithoutAwaitingEmitsOneShotOutputAndEnd(): self::assertSame('"Greetings world"', $this->successValue($output)); self::assertTrue($transport->isClosed()); - // The handler never parked: only the single journal read happened (read #0), - // so the resume loop body never ran. - self::assertSame('', $transport->outputAtRead(1), 'no second read occurred — the loop body never ran'); + // The handler never parked — it emitted no AwaitingOn — so the one-shot Output/End + // ran straight through without yielding to the streaming loop. + self::assertNotContains(MessageType::AwaitingOn, $this->frameTypes($output), 'a non-awaiting handler never parks'); } public function testEofWhileParkedSuspendsGracefully(): void @@ -241,11 +246,185 @@ public function testEofWhileParkedSuspendsGracefully(): void ); $output = $transport->written(); - self::assertSame([MessageType::SleepCommand, MessageType::Suspension], $this->frameTypes($output)); + self::assertSame([MessageType::SleepCommand, MessageType::AwaitingOn, MessageType::Suspension], $this->frameTypes($output)); self::assertNotContains(MessageType::OutputCommand, $this->frameTypes($output), 'no terminal output on a graceful suspend'); self::assertTrue($transport->isClosed()); } + /** + * Each of the four combinators reached while a CANCEL signal is already pending must + * surface a terminal 409 — not re-suspend (the old bug streamed a SuspensionMessage) + * and not throw a misleading LogicException/TerminalException. The cancel rides in the + * same chunk as the journal, so it is in the signal table before the handler runs. + * + * @return array + */ + public static function combinatorHandlers(): array + { + return [ + 'awaitAny' => ['raceWhileCancelled'], + 'select' => ['selectWhileCancelled'], + 'awaitAll' => ['awaitAllWhileCancelled'], + 'awaitAllSucceeded' => ['awaitAllSucceededWhileCancelled'], + ]; + } + + #[DataProvider('combinatorHandlers')] + public function testCombinatorWhileCancelledTerminatesWith409(string $handlerName): void + { + $transport = $this->drive( + new CancellationService(), + 'CancellationService', + $handlerName, + [ + // The CANCEL signal arrives in the same chunk as the journal, so it is + // pending before the combinator is reached; no further chunk follows. + (new JournalBuilder())->input('')->build() . (new JournalBuilder())->cancelSignal()->frames(), + ], + ); + + $output = $transport->written(); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($output), 'a pending cancel must not re-suspend a combinator'); + self::assertSame(CancelledException::CODE, $this->failure($output)->code, 'a cancelled combinator fails with HTTP 409'); + self::assertTrue($transport->isClosed()); + } + + public function testCancelObservedMidStreamThenCombinatorDrains(): void + { + // The sleep await observes the cancel (CancelledException, caught by the handler); + // the handler then reaches a combinator that is still cancelled. The driver must + // drain that re-park (its predicate already holds) into the terminal 409 rather + // than block for a chunk that never comes and suspend. + $transport = $this->drive( + new CancellationService(), + 'CancellationService', + 'raceAfterObservedCancel', + [ + (new JournalBuilder())->input('')->build(), + (new JournalBuilder())->cancelSignal()->frames(), + ], + ); + + $output = $transport->written(); + // SleepCommand + the sleep park's AwaitingOn, then the call, then 409 — the + // combinator re-park drains synchronously (its predicate already holds) so it + // never parks again and emits no second AwaitingOn. + self::assertSame( + [MessageType::SleepCommand, MessageType::AwaitingOn, MessageType::CallCommand, MessageType::OutputCommand, MessageType::End], + $this->frameTypes($output), + ); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($output)); + self::assertSame(CancelledException::CODE, $this->failure($output)->code); + } + + public function testCombinatorParkedThenCancelledMidStreamTerminatesWith409(): void + { + // No cancel at entry: the combinator parks, then a CANCEL arrives on the open + // stream. The post-suspend rescan must surface a 409 (not the misleading + // TerminalException the rescan used to throw when no future was ready). + $transport = $this->drive( + new CancellationService(), + 'CancellationService', + 'raceWhileCancelled', + [ + (new JournalBuilder())->input('')->build(), + (new JournalBuilder())->cancelSignal()->frames(), + ], + ); + + $output = $transport->written(); + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($output)); + self::assertSame(CancelledException::CODE, $this->failure($output)->code); + } + + public function testBatchedCompletionsResolveTwoSequentialAwaitsInOneChunk(): void + { + // The runtime batches both the invocation-id (completion 1) and the call result + // (completion 2) into a single chunk. The handler awaits them in sequence; the + // driver must run both awaits off that one chunk, never suspending. + $transport = $this->drive( + new CallOptionsService(), + 'CallOptionsService', + 'callAwaitIdThenResult', + [ + (new JournalBuilder())->input('')->build(), + (new JournalBuilder())->invocationIdCompletion(1, 'inv-xyz')->frames() + . (new JournalBuilder())->callCompletion(2, '"call-result"')->frames(), + ], + ); + + $output = $transport->written(); + // One AwaitingOn for the first (parked) await; the second await finds its + // completion already buffered and returns on the fast path without parking. + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($output)); + self::assertSame('"call-result"', $this->successValue($output)); + // Only the CallCommand + the park's AwaitingOn were out at read #1, which carried + // both completions: the handler drove both awaits off that single notification chunk. + self::assertSame( + [MessageType::CallCommand, MessageType::AwaitingOn], + $this->frameTypes($transport->outputAtRead(1)), + ); + } + + public function testAwakeableSignalOnOpenStreamResolvesParkedAwait(): void + { + // An awakeable emits no command; the handler parks on signal idx 17, which the + // runtime resolves by streaming a SignalNotification on the open channel. + $transport = $this->drive( + new AwakeableService(), + 'AwakeableService', + 'awaitOne', + [ + (new JournalBuilder())->input('')->build(), + (new JournalBuilder())->awakeableSignal(17, '"resolved"')->frames(), + ], + ); + + $output = $transport->written(); + // The awakeable park announces an AwaitingOn (it carries no command of its own), + // which is exactly what lets the runtime push the resolving signal back. + self::assertSame([MessageType::AwaitingOn, MessageType::OutputCommand, MessageType::End], $this->frameTypes($output)); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($output)); + self::assertSame('"resolved"', $this->successValue($output)); + } + + public function testEofWhileParkedOnAwakeableSuspendsWaitingOnCancelAndSignal(): void + { + // No resolving signal arrives: parked on the awakeable (signal idx 17), an EOF + // suspends. The await tree must declare the CANCEL signal (idx 1) at the outer + // node and nest the awakeable signal (idx 17) — so the runtime re-invokes on + // either the awakeable resolution or a cancel. + $transport = $this->drive( + new AwakeableService(), + 'AwakeableService', + 'awaitOne', + [(new JournalBuilder())->input('')->build()], + ); + + // Parking first announced an AwaitingOn (await tree in field 1); the EOF then + // wrote the Suspension (await tree in field 4). Both carry the same tree. + $frames = MessageCodec::decodeAll($transport->written()); + self::assertSame([MessageType::AwaitingOn, MessageType::Suspension], \array_map(static fn ($f) => $f->type(), $frames)); + + $awaitingOnReader = new Reader($frames[0]->payload); + [$awaitingOnField] = $awaitingOnReader->readTag(); + self::assertSame(1, $awaitingOnField, 'the AwaitingOn carries its await tree in field 1'); + + $reader = new Reader($frames[1]->payload); + [$field] = $reader->readTag(); + self::assertSame(4, $field, 'the suspension carries its await tree in field 4'); + $outer = $this->decodeFuture($reader->readLengthDelimited()); + + // A single-signal await flattens: the awakeable signal (idx 17) and the CANCEL + // signal (idx 1) sit together on the FirstCompleted node, with nothing nested — + // the flat shape the runtime keys its cancel wake-up off. + self::assertSame([17, 1], $outer['signals'], 'the await waits on the awakeable signal and the CANCEL signal'); + self::assertSame([], $outer['completions']); + self::assertSame([], $outer['nested'], 'a single await is flattened, not nested'); + } + private function frameOfType(string $output, MessageType $type): Frame { foreach (MessageCodec::decodeAll($output) as $frame) { @@ -255,4 +434,51 @@ private function frameOfType(string $output, MessageType $type): Frame } self::fail("No {$type->name} frame in streamed output"); } + + /** + * Decodes a {@see \Qcodr\Restate\Sdk\Protocol\Message\Future} payload, returning its + * leaf ids and the raw bytes of each nested future (mirrors the helper in + * {@see \Qcodr\Restate\Sdk\Tests\Unit\Vm\StateMachineTest}). + * + * @return array{completions: list, signals: list, nested: list} + */ + private function decodeFuture(string $payload): array + { + $reader = new Reader($payload); + $completions = []; + $signals = []; + $nested = []; + while (!$reader->atEnd()) { + [$field, $wire] = $reader->readTag(); + switch ($field) { + case 1: + $completions = $this->unpackVarints($reader->readLengthDelimited()); + break; + case 2: + $signals = $this->unpackVarints($reader->readLengthDelimited()); + break; + case 4: + $nested[] = $reader->readLengthDelimited(); + break; + default: + $reader->skip($wire); + } + } + + return ['completions' => $completions, 'signals' => $signals, 'nested' => $nested]; + } + + /** + * @return list + */ + private function unpackVarints(string $packed): array + { + $reader = new Reader($packed); + $values = []; + while (!$reader->atEnd()) { + $values[] = $reader->readVarint(); + } + + return $values; + } } diff --git a/tests/Unit/Protocol/Message/FailureTest.php b/tests/Unit/Protocol/Message/FailureTest.php index 012e804..40c13fb 100644 --- a/tests/Unit/Protocol/Message/FailureTest.php +++ b/tests/Unit/Protocol/Message/FailureTest.php @@ -18,20 +18,34 @@ public function testRoundTripPreservesCodeAndMessage(): void self::assertSame('boom', $failure->message); } - public function testDecodeSkipsTheUnusedMetadataField(): void + public function testEncodeDecodeRoundTripsMetadata(): void { - // The repeated `metadata` field (3) is decode-tolerant but never produced by - // this SDK; the decoder must skip it and still read code (1) and message (2). + // The repeated `metadata` field (3) carries user error context (V7) and must + // survive an encode/decode round trip alongside code (1) and message (2). + $failure = new Failure(42, 'oops', ['key1' => 'v1', 'key2' => 'v2']); + + $decoded = Failure::decode($failure->encode()); + + self::assertSame(42, $decoded->code); + self::assertSame('oops', $decoded->message); + self::assertSame(['key1' => 'v1', 'key2' => 'v2'], $decoded->metadata); + } + + public function testDecodeToleratesAnUnknownTrailingField(): void + { + // An unknown field (here field 5) is skipped; code/message still decode and + // metadata stays empty. $bytes = (new Writer()) ->writeUint32(1, 42) ->writeString(2, 'oops') - ->writeBytesPresent(3, 'extra-metadata') + ->writeBytesPresent(5, 'future-field') ->toString(); $failure = Failure::decode($bytes); self::assertSame(42, $failure->code); self::assertSame('oops', $failure->message); + self::assertSame([], $failure->metadata); } public function testDefaultCodeDecodesToZero(): void diff --git a/tests/Unit/Protocol/Message/FutureTest.php b/tests/Unit/Protocol/Message/FutureTest.php index a91c974..5616171 100644 --- a/tests/Unit/Protocol/Message/FutureTest.php +++ b/tests/Unit/Protocol/Message/FutureTest.php @@ -25,6 +25,19 @@ public function testForSignalEncodesPackedSignalIdInFieldTwo(): void self::assertTrue($reader->atEnd(), 'the default combinator must not be emitted'); } + public function testForNamedSignalEncodesNameInFieldThree(): void + { + // A lone named-signal await: the name is a present string in field 3 and the + // default (Unknown = 0) combinator is omitted by proto3 scalar rules. + $reader = new Reader(Future::forNamedSignal('my-signal')->encode()); + + [$field, $wire] = $reader->readTag(); + self::assertSame(3, $field); + self::assertSame(WireType::LENGTH_DELIMITED, $wire); + self::assertSame('my-signal', $reader->readLengthDelimited()); + self::assertTrue($reader->atEnd(), 'the default combinator must not be emitted'); + } + public function testForCompletionEncodesPackedCompletionInFieldOne(): void { $reader = new Reader(Future::forCompletion(42)->encode()); diff --git a/tests/Unit/Protocol/Message/SendSignalCommandNamedTest.php b/tests/Unit/Protocol/Message/SendSignalCommandNamedTest.php index 9308f44..8269158 100644 --- a/tests/Unit/Protocol/Message/SendSignalCommandNamedTest.php +++ b/tests/Unit/Protocol/Message/SendSignalCommandNamedTest.php @@ -5,7 +5,10 @@ namespace Qcodr\Restate\Sdk\Tests\Unit\Protocol\Message; use PHPUnit\Framework\TestCase; +use Qcodr\Restate\Sdk\Protocol\Message\Failure; use Qcodr\Restate\Sdk\Protocol\Message\SendSignalCommand; +use Qcodr\Restate\Sdk\Protocol\Message\Value; +use Qcodr\Restate\Sdk\Protocol\MessageType; use Qcodr\Restate\Sdk\Protocol\Protobuf\Reader; use Qcodr\Restate\Sdk\Protocol\Protobuf\WireType; use ReflectionClass; @@ -40,6 +43,43 @@ public function testNamedSignalWithVoidResultEmitsFieldFour(): void self::assertSame('', $fields[4]); } + public function testResolveNamedEncodesNameAndValueResult(): void + { + $command = SendSignalCommand::resolveNamed('inv-target', 'my-signal', new Value('"v"')); + + self::assertSame(MessageType::SendSignalCommand, $command->messageType()); + self::assertFalse($command->requestedAck()); + + $fields = self::fields($command->encode()); + self::assertSame('inv-target', $fields[1]); + self::assertSame('my-signal', $fields[3]); + self::assertArrayNotHasKey(2, $fields, 'a named signal must not emit the built-in idx'); + self::assertArrayNotHasKey(4, $fields, 'a value result must not also emit a void'); + self::assertArrayNotHasKey(6, $fields, 'a resolve carries no failure'); + + $valueBytes = $fields[5]; + self::assertIsString($valueBytes); + self::assertSame('"v"', Value::decode($valueBytes)->content); + } + + public function testRejectNamedEncodesNameAndFailureResult(): void + { + $command = SendSignalCommand::rejectNamed('inv-target', 'my-signal', new Failure(409, 'denied')); + + $fields = self::fields($command->encode()); + self::assertSame('inv-target', $fields[1]); + self::assertSame('my-signal', $fields[3]); + self::assertArrayNotHasKey(2, $fields); + self::assertArrayNotHasKey(4, $fields); + self::assertArrayNotHasKey(5, $fields, 'a reject carries no value'); + + $failureBytes = $fields[6]; + self::assertIsString($failureBytes); + $failure = Failure::decode($failureBytes); + self::assertSame(409, $failure->code); + self::assertSame('denied', $failure->message); + } + private static function namedSignal(string $target, string $name, bool $void = false): SendSignalCommand { $class = new ReflectionClass(SendSignalCommand::class); diff --git a/tests/Unit/Vm/AwakeableIdTest.php b/tests/Unit/Vm/AwakeableIdTest.php index d9acb40..68b1210 100644 --- a/tests/Unit/Vm/AwakeableIdTest.php +++ b/tests/Unit/Vm/AwakeableIdTest.php @@ -8,7 +8,7 @@ use Qcodr\Restate\Sdk\Vm\AwakeableId; /** - * Verifies the public awakeable id encoding: the literal `prom_1` prefix followed by + * Verifies the public awakeable id encoding: the literal `sign_1` prefix followed by * the unpadded url-safe Base64 of the invocation id concatenated with the signal * index as a 32-bit big-endian unsigned integer. */ @@ -16,26 +16,26 @@ final class AwakeableIdTest extends TestCase { public function testEncodesAKnownVector(): void { - self::assertSame('prom_1aW52LTEAAAAR', AwakeableId::encode('inv-1', 17)); + self::assertSame('sign_1aW52LTEAAAAR', AwakeableId::encode('inv-1', 17)); } public function testEncodesAKnownVectorForALongerInvocationId(): void { self::assertSame( - 'prom_1bXktaW52b2NhdGlvbi1pZAAAABE', + 'sign_1bXktaW52b2NhdGlvbi1pZAAAABE', AwakeableId::encode('my-invocation-id', 17), ); } public function testStartsWithTheProtocolPrefix(): void { - self::assertStringStartsWith('prom_1', AwakeableId::encode('inv-1', 1)); + self::assertStringStartsWith('sign_1', AwakeableId::encode('inv-1', 1)); } public function testSuffixDecodesBackToInvocationIdAndBigEndianSignalIndex(): void { $id = AwakeableId::encode('inv-7', 259); - $suffix = \substr($id, \strlen('prom_1')); + $suffix = \substr($id, \strlen('sign_1')); // Restore the standard alphabet and padding, then decode and check the bytes: // invocation id followed by the signal index as uint32 big-endian. diff --git a/tests/Unit/Vm/NamedSignalStateMachineTest.php b/tests/Unit/Vm/NamedSignalStateMachineTest.php new file mode 100644 index 0000000..540e086 --- /dev/null +++ b/tests/Unit/Vm/NamedSignalStateMachineTest.php @@ -0,0 +1,207 @@ +notifyInput($journal); + $vm->notifyInputClosed(); + self::assertTrue($vm->isReadyToExecute()); + + return $vm; + } + + public function testNamedSignalNotificationInJournalResolvesAwaitWithoutSuspending(): void + { + $journal = (new JournalBuilder()) + ->input('1') + ->namedSignal('ready', '"value"') + ->build(); + $vm = $this->machine($journal); + $vm->sysInput(); + + self::assertTrue($vm->isNamedSignalReady('ready')); + self::assertFalse($vm->isNamedSignalReady('absent')); + + $notification = $vm->awaitNamedSignal('ready'); // returns without suspending + self::assertSame('value', \json_decode($notification->value ?? '', true)); + self::assertSame('ready', $notification->signalName); + self::assertSame('"value"', $vm->peekNamedSignal('ready')->value); + + // A resolved named signal is keyed by name, never by an index in the signals table. + self::assertSame([], $this->outputTypes($vm), 'a replayed named signal emits nothing'); + } + + public function testUnresolvedNamedSignalAwaitSuspendsWithNamedSignalAwaitTree(): void + { + $vm = $this->machine((new JournalBuilder())->input('1')->build()); + $vm->sysInput(); + + try { + $vm->awaitNamedSignal('pending'); + self::fail('expected suspension on an unresolved named signal'); + } catch (SuspendException) { + // expected + } + + $frames = MessageCodec::decodeAll($vm->takeOutput()); + self::assertSame([MessageType::Suspension], \array_map(static fn ($f) => $f->type(), $frames)); + + // The await tree flattens the named signal next to the CANCEL signal under a + // FirstCompleted node (the canonical single-await shape), exactly like an awakeable. + $tree = $this->decodeSuspensionFuture($frames[0]->payload); + self::assertSame(['pending'], $tree['named'], 'the awaited named signal sits on the cancel-guarded node'); + self::assertSame([self::CANCEL_SIGNAL_ID], $tree['signals'], 'the suspension also waits on the CANCEL signal'); + self::assertSame([], $tree['completions']); + self::assertSame([], $tree['nested'], 'a single await is flattened, not nested'); + self::assertSame(CombinatorType::FirstCompleted->value, $tree['combinator']); + } + + public function testRejectedNamedSignalRaisesTheCarriedFailureOnAwait(): void + { + $journal = (new JournalBuilder()) + ->input('1') + ->failedNamedSignal('boom', 'denied', 409) + ->build(); + $vm = $this->machine($journal); + $vm->sysInput(); + + $notification = $vm->awaitNamedSignal('boom'); + self::assertNotNull($notification->failure); + self::assertSame(409, $notification->failure->code); + self::assertSame('denied', $notification->failure->message); + } + + public function testParkedNamedSignalAwaitResumesWhenSignalStreamedIn(): void + { + $sink = new RecordingOutputSink(); + $vm = new StateMachine(ServiceProtocolVersion::V7, new FiberSuspender(), $sink); + $vm->notifyInput((new JournalBuilder())->input('1')->build()); + self::assertTrue($vm->isReadyToExecute()); + $vm->sysInput(); + + $result = null; + $fiber = new Fiber(static function () use ($vm, &$result): void { + $result = $vm->awaitNamedSignal('streamed'); + }); + + $park = $fiber->start(); + self::assertTrue($fiber->isSuspended(), 'an unresolved named-signal await parks the fiber'); + self::assertInstanceOf(ParkSignal::class, $park); + self::assertSame(['streamed'], $park->awaitTree->waitingNamedSignals); + self::assertSame([self::CANCEL_SIGNAL_ID], $park->awaitTree->waitingSignals); + + // The driver streams the named signal in (as the runtime would) and resumes. + $vm->notifyInput((new JournalBuilder())->namedSignal('streamed', '"hi"')->frames()); + $fiber->resume(); + + self::assertTrue($fiber->isTerminated(), 'the fiber finished once the named signal arrived'); + self::assertInstanceOf(Notification::class, $result); + self::assertSame('"hi"', $result->value); + self::assertSame('streamed', $result->signalName); + + // Parking announced the await tree (AwaitingOn) but wrote no suspension frame. + self::assertSame([MessageType::AwaitingOn], $sink->frameTypes()); + } + + /** @return list */ + private function outputTypes(StateMachine $vm): array + { + return \array_map( + static fn ($frame) => $frame->type(), + MessageCodec::decodeAll($vm->takeOutput()), + ); + } + + /** + * Decodes the await-point {@see \Qcodr\Restate\Sdk\Protocol\Message\Future} carried in + * a suspension frame (field 4), capturing its named-signal leaves (field 3) too. + * + * @return array{completions: list, signals: list, named: list, nested: list, combinator: int} + */ + private function decodeSuspensionFuture(string $suspensionPayload): array + { + $reader = new Reader($suspensionPayload); + [$field] = $reader->readTag(); + self::assertSame(4, $field, 'the suspension carries the await tree in field 4'); + + $tree = new Reader($reader->readLengthDelimited()); + $completions = []; + $signals = []; + $named = []; + $nested = []; + $combinator = 0; + while (!$tree->atEnd()) { + [$treeField, $wire] = $tree->readTag(); + switch ($treeField) { + case 1: + $completions = $this->unpackVarints($tree->readLengthDelimited()); + break; + case 2: + $signals = $this->unpackVarints($tree->readLengthDelimited()); + break; + case 3: + $named[] = $tree->readLengthDelimited(); + break; + case 4: + $nested[] = $tree->readLengthDelimited(); + break; + case 5: + $combinator = $tree->readVarint(); + break; + default: + $tree->skip($wire); + } + } + + return [ + 'completions' => $completions, + 'signals' => $signals, + 'named' => $named, + 'nested' => $nested, + 'combinator' => $combinator, + ]; + } + + /** + * @return list + */ + private function unpackVarints(string $packed): array + { + $reader = new Reader($packed); + $values = []; + while (!$reader->atEnd()) { + $values[] = $reader->readVarint(); + } + + return $values; + } +} diff --git a/tests/Unit/Vm/StateMachineTest.php b/tests/Unit/Vm/StateMachineTest.php index 75fe647..01eb338 100644 --- a/tests/Unit/Vm/StateMachineTest.php +++ b/tests/Unit/Vm/StateMachineTest.php @@ -457,12 +457,12 @@ public function testSuspensionAwaitsBothTheResultAndTheCancelSignal(): void self::assertSame(4, $field); $outer = $this->decodeFuture($reader->readLengthDelimited()); - // The outer node waits on the built-in CANCEL signal (idx 1) so a cancel wakes it... + // A single awaited completion flattens next to the built-in CANCEL signal (idx 1) + // under a FirstCompleted node — matching the canonical await-tree the runtime keys + // its cancel wake-up off — rather than nesting the completion below the guard. + self::assertSame([$completionId], $outer['completions'], 'the awaited completion sits on the cancel-guarded node'); self::assertSame([1], $outer['signals'], 'the suspension also waits on the CANCEL signal'); - // ...and nests the actual awaited completion. - self::assertCount(1, $outer['nested'], 'the real await point is nested under the cancel guard'); - $inner = $this->decodeFuture($outer['nested'][0]); - self::assertSame([$completionId], $inner['completions']); + self::assertSame([], $outer['nested'], 'a single await is flattened, not nested'); } /** diff --git a/tests/Unit/Vm/StreamingStateMachineTest.php b/tests/Unit/Vm/StreamingStateMachineTest.php index 4dd72c8..5b8ecfa 100644 --- a/tests/Unit/Vm/StreamingStateMachineTest.php +++ b/tests/Unit/Vm/StreamingStateMachineTest.php @@ -62,14 +62,17 @@ public function testUnresolvedAwaitParksWithoutEmittingSuspension(): void // SuspensionMessage) plus the predicate the driver evaluates before resuming. self::assertInstanceOf(ParkSignal::class, $park); self::assertFalse(($park->isResolved)(), 'the await is unresolved while the completion is absent'); + // A single completion await flattens next to the CANCEL signal under a + // FirstCompleted node (no nesting), matching the canonical await tree. $awaitTree = $park->awaitTree; + self::assertSame([$resultId], $awaitTree->waitingCompletions); self::assertSame([self::CANCEL_SIGNAL_ID], $awaitTree->waitingSignals); self::assertSame(CombinatorType::FirstCompleted, $awaitTree->combinatorType); - self::assertCount(1, $awaitTree->nestedFutures); - self::assertSame([$resultId], $awaitTree->nestedFutures[0]->waitingCompletions); + self::assertSame([], $awaitTree->nestedFutures, 'a single await is flattened, not nested'); - // Only the CallCommand was emitted: no suspension frame is written while parked. - self::assertSame([MessageType::CallCommand], $sink->frameTypes()); + // Parking announces the await tree with an AwaitingOn (so the runtime pushes the + // awaited completions/signals on the open stream) but writes NO suspension frame. + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn], $sink->frameTypes()); self::assertNotContains(MessageType::Suspension, $sink->frameTypes()); // A non-buffering sink has nothing to drain via the legacy takeOutput() path. @@ -99,8 +102,9 @@ public function testParkedAwaitResumesWithValueWhenCompletionStreamedIn(): void self::assertSame('"hello"', $result->value); self::assertSame($resultId, $result->completionId); - // The park/resume round trip never wrote a suspension frame. - self::assertSame([MessageType::CallCommand], $sink->frameTypes()); + // The park/resume round trip announced the await with an AwaitingOn but never + // wrote a suspension frame. + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn], $sink->frameTypes()); } public function testCombinatorParksOnFirstCompletedTreeInStreaming(): void @@ -113,7 +117,7 @@ public function testCombinatorParksOnFirstCompletedTreeInStreaming(): void // is resumed directly (no driver), so a trivial always-true closure suffices. $finished = false; $fiber = new Fiber(static function () use ($vm, &$finished): void { - $vm->suspendAny([2, 4], [], static fn (): bool => true); + $vm->suspendAny([2, 4], [], [], static fn (): bool => true); $finished = true; });