Fix/bidi streaming cancellation

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"]

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

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/<ts>/exclusions.new.yaml` lists everything that
 failed/was skipped — copy entries into `exclusions.yaml` to baseline new gaps.

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<string, string>} $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<string, string>} $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<string, string>} $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);
         });
     }
 

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'] ?? ''));
+    }
 }

conformance/Services/VirtualObjectCommandInterpreter.php

@@ -25,15 +25,16 @@
  *   - "awk-{awakeableKey}"  the awakeable id registered for that key (string)
  *   - "results"             the ordered list<string> 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:<value>` / `err:<reason>`.
+                    $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:<value>` or
+     * `err:<reason>`, 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<array<string, mixed>> $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<string, mixed> $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",
                 );
         }
     }