fix: seven correctness and lifecycle findings from the bidi-streaming review

.gitignore

@@ -9,3 +9,4 @@
 .DS_Store
 /coverage/
 /build/
+/graphify-out/

README.md

@@ -233,6 +233,19 @@ and scale across cores like a Swoole worker pool:
 (new AmpStreamingServer($endpoint))->listen('0.0.0.0', 9080, workers: 8); // 0 = one per CPU
 ```
 
+Its connection, concurrency and idle ceilings default to values sized for the Restate
+runtime as the only peer — many long-lived, deliberately idle streams from one IP, which
+amphp's own defaults reject. That is not a general hardening posture: if the endpoint is
+reachable by anything other than your runtime (especially without `identityKey()`),
+tighten them:
+
+```php
+new AmpStreamingServer($endpoint, limits: new ServerLimits(
+    connectionLimitPerIp: 200,
+    streamIdleTimeoutSeconds: 300,
+));
+```
+
 The same framework-agnostic core is also hostable request/response via the **Swoole**
 server (`Qcodr\Restate\Sdk\Server\SwooleServer`, needs `ext-swoole`), a **PSR-15** adapter
 (`Qcodr\Restate\Sdk\Server\Psr15Handler`) in any Slim/Mezzio stack, on **AWS Lambda**

src/Context/Context.php

@@ -69,9 +69,14 @@ public function run(string $name, callable $action, ?RunOptions $options = null)
      * concurrently (e.g. raced via {@see select} / {@see awaitAll} against timers,
      * calls or signals). Await the returned future to obtain the value.
      *
+     * Failure handling matches {@see run}: a {@see TerminalException} resolves the future
+     * with that failure, and a non-terminal throwable is governed by `$options`' retry
+     * policy — without a bounded policy it propagates from this call and the whole
+     * attempt is retried by the runtime.
+     *
      * @param callable():mixed $action
      */
-    public function runAsync(string $name, callable $action): DurableFuture;
+    public function runAsync(string $name, callable $action, ?RunOptions $options = null): DurableFuture;
 
     /** Suspends the invocation for the given duration using a durable timer. */
     public function sleep(float $seconds): void;

src/Context/RestateContext.php

@@ -88,46 +88,26 @@ public function logger(): LoggerInterface
 
     public function run(string $name, callable $action, ?RunOptions $options = null): mixed
     {
-        $completionId = $this->vm->sysRun($name);
-
-        if ($this->vm->isCompletionReady($completionId)) {
-            return $this->completionFuture($completionId)->await();
-        }
-
-        try {
-            $result = $action();
-        } catch (TerminalException $e) {
-            $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
-            // the runtime echoes the completion, where await() raises the terminal
-            // failure carried by it.
-            return $this->completionFuture($completionId)->await();
-        } catch (Throwable $e) {
-            return $this->handleRunFailure($completionId, $e, $options?->retryPolicy);
-        }
-
-        try {
-            $serialized = $this->serde->serialize($result);
-        } catch (SerializationException $e) {
-            // A non-serializable result must fail terminally, not retry forever: the
-            // RunCommand is already journaled, so without a proposed completion the
-            // invocation would re-run 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)->await();
-        }
-
-        $this->vm->proposeRunCompletionSuccess($completionId, $serialized);
+        return $this->runFuture($name, $action, $options)->await();
+    }
 
-        return $this->completionFuture($completionId)->await();
+    public function runAsync(string $name, callable $action, ?RunOptions $options = null): DurableFuture
+    {
+        return $this->runFuture($name, $action, $options);
     }
 
-    public function runAsync(string $name, callable $action): DurableFuture
+    /**
+     * Journals the RunCommand, executes the closure at most once (never on replay) and
+     * proposes its result, returning the future that resolves to it.
+     *
+     * {@see run} awaits that future immediately; {@see runAsync} hands it back so the run
+     * can be composed with other futures. Sharing this body is what keeps the two from
+     * drifting — in particular the non-terminal branch below, which applies the retry
+     * policy identically for both.
+     *
+     * @param callable():mixed $action
+     */
+    private function runFuture(string $name, callable $action, ?RunOptions $options): DurableFuture
     {
         $completionId = $this->vm->sysRun($name);
 
@@ -139,20 +119,26 @@ public function runAsync(string $name, callable $action): DurableFuture
         try {
             $result = $action();
         } catch (TerminalException $e) {
+            // In request/response, awaiting the just-proposed (not-yet-replayed)
+            // completion writes the suspension and unwinds; in streaming it parks until
+            // the runtime echoes the completion, where await() raises the terminal
+            // failure carried by it.
             $this->vm->proposeRunCompletionFailure(
                 $completionId,
                 new Failure($e->statusCode(), $e->getMessage(), $e->metadata),
             );
 
             return $this->completionFuture($completionId);
+        } catch (Throwable $e) {
+            return $this->handleRunFailure($completionId, $e, $options?->retryPolicy);
         }
 
         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.
+            // A non-serializable result must fail terminally, not retry forever: the
+            // RunCommand is already journaled, so without a proposed completion the
+            // invocation would re-run the closure on every attempt.
             $this->vm->proposeRunCompletionFailure(
                 $completionId,
                 new Failure(TerminalException::DEFAULT_CODE, 'run result is not serializable: ' . $e->getMessage()),
@@ -175,7 +161,7 @@ public function runAsync(string $name, callable $action): DurableFuture
      * a terminal failure once attempts are exhausted, or reports a retryable attempt
      * failure carrying a computed backoff so the whole invocation re-runs the closure.
      */
-    private function handleRunFailure(int $completionId, Throwable $error, ?RetryPolicy $policy): mixed
+    private function handleRunFailure(int $completionId, Throwable $error, ?RetryPolicy $policy): DurableFuture
     {
         if ($policy === null || $policy->maxAttempts === null) {
             throw $error;
@@ -188,9 +174,10 @@ private function handleRunFailure(int $completionId, Throwable $error, ?RetryPol
                 new Failure(TerminalException::DEFAULT_CODE, $error->getMessage()),
             );
 
-            // The proposed failure is terminal; in request/response await() writes the
-            // suspension and unwinds, in streaming it raises the carried failure.
-            return $this->completionFuture($completionId)->await();
+            // The proposed failure is terminal; awaiting the returned future writes the
+            // suspension and unwinds (request/response) or raises the carried failure
+            // (streaming). runAsync() defers that await to the caller.
+            return $this->completionFuture($completionId);
         }
 
         $this->logger->warning('Durable run failed (retryable): ' . $error->getMessage(), ['exception' => $error]);

src/Protocol/Message/ProposeRunCompletionAck.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Qcodr\Restate\Sdk\Protocol\Message;
+
+use Qcodr\Restate\Sdk\Protocol\Protobuf\Reader;
+
+/**
+ * `ProposeRunCompletionAckMessage` (0x0007, service protocol V7): the runtime's
+ * confirmation that a `ProposeRunCompletion` was durably stored.
+ *
+ * Over bidirectional streaming the runtime acks a proposal with this control frame
+ * instead of echoing the value back as a notification, so the SDK promotes the result it
+ * stashed at propose time (see {@see \Qcodr\Restate\Sdk\Vm\StateMachine}). It is its own
+ * message type, not a notification: `completion_id` (field 1) is all it carries.
+ */
+final class ProposeRunCompletionAck
+{
+    public function __construct(public readonly ?int $completionId)
+    {
+    }
+
+    public static function decode(string $bytes): self
+    {
+        $reader = new Reader($bytes);
+        $completionId = null;
+        while (!$reader->atEnd()) {
+            [$field, $wire] = $reader->readTag();
+            if ($field === 1) {
+                $completionId = $reader->readVarint();
+            } else {
+                $reader->skip($wire);
+            }
+        }
+
+        return new self($completionId);
+    }
+}

src/Server/AmpStreamingServer.php

@@ -56,50 +56,31 @@
  */
 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;
+    private readonly ServerLimits $limits;
 
+    /**
+     * @param ?ServerLimits $limits connection / concurrency / idle ceilings handed to
+     *                              amphp. Defaults are tuned for the Restate runtime as
+     *                              the only peer — see {@see ServerLimits} before exposing
+     *                              this endpoint more widely.
+     */
     public function __construct(
         private readonly Endpoint $endpoint,
         ?Serde $serde = null,
         ?Clock $clock = null,
         ?LoggerInterface $logger = null,
         bool $debug = false,
+        ?ServerLimits $limits = null,
     ) {
         if (!\class_exists(SocketHttpServer::class)) {
             throw new RuntimeException(
                 'AmpStreamingServer requires amphp/http-server; run composer require amphp/http-server',
             );
         }
 
+        $this->limits = $limits ?? new ServerLimits();
         $this->logger = $logger ?? new NullLogger();
         $this->processor = new RequestProcessor(
             $endpoint,
@@ -150,7 +131,9 @@ public function listen(string $host = '0.0.0.0', int $port = 9080, int $workers
             if ($pid === 0) {
                 $this->runServer($host, $port, reusePort: true, announce: false);
 
-                return;
+                // Exit rather than return: returning would run the caller's post-listen()
+                // code, its destructors and its shutdown functions once per worker.
+                exit(0);
             }
             if ($pid > 0) {
                 $childPids[] = $pid;
@@ -171,9 +154,23 @@ public function listen(string $host = '0.0.0.0', int $port = 9080, int $workers
         }
 
         // The parent serves too; when it is asked to stop it returns, then we stop the
-        // workers and reap them so none is left as a zombie.
-        $this->runServer($host, $port, reusePort: true, announce: false);
+        // workers and reap them so none is left as a zombie. The finally matters: if the
+        // parent's own server throws (bind failure, driver error) the children would
+        // otherwise survive it, still holding the port open through SO_REUSEPORT.
+        try {
+            $this->runServer($host, $port, reusePort: true, announce: false);
+        } finally {
+            self::stopWorkers($childPids);
+        }
+    }
 
+    /**
+     * Signals each worker to stop and waits for it, so none is left as a zombie.
+     *
+     * @param list<int> $childPids
+     */
+    private static function stopWorkers(array $childPids): void
+    {
         foreach ($childPids as $pid) {
             if (\function_exists('posix_kill')) {
                 \posix_kill($pid, \SIGTERM);
@@ -200,13 +197,13 @@ private function runServer(string $host, int $port, bool $reusePort, bool $annou
         // 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,
+            connectionLimit: $this->limits->connectionLimit,
+            connectionLimitPerIp: $this->limits->connectionLimitPerIp,
+            concurrencyLimit: $this->limits->concurrencyLimit,
             httpDriverFactory: new DefaultHttpDriverFactory(
                 $this->logger,
-                streamTimeout: self::STREAM_IDLE_TIMEOUT_SECONDS,
-                connectionTimeout: self::CONNECTION_IDLE_TIMEOUT_SECONDS,
+                streamTimeout: $this->limits->streamIdleTimeoutSeconds,
+                connectionTimeout: $this->limits->connectionIdleTimeoutSeconds,
                 allowHttp2Upgrade: true,
             ),
         );

src/Server/ServerLimits.php

@@ -0,0 +1,81 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Qcodr\Restate\Sdk\Server;
+
+use RuntimeException;
+
+/**
+ * Connection, concurrency and idle ceilings for {@see AmpStreamingServer}.
+ *
+ * The defaults are deliberately far above amphp's own (1000 connections, 10 per IP,
+ * 1000 concurrent, 15 s stream / 60 s connection idle), because the Restate runtime is a
+ * single trusted peer that opens one long-lived bidi connection per in-flight invocation,
+ * all from the same IP, and legitimately keeps that stream open and idle while the handler
+ * is parked awaiting a completion, a signal, or a cancel the runtime may deliver much
+ * later. The runtime never half-closes the request (so amphp never `suspend()`s the stream
+ * timer) and its HTTP/2 keep-alive PINGs only refresh the connection timer: at amphp's
+ * 15 s the driver would `releaseStream(..., "Closing stream due to inactivity")`, making
+ * the body read throw mid-invocation and silently dropping a pending cancel. At ~10
+ * connections per IP the runtime is denied new connections ("too many existing
+ * connections"), surfacing as broken-pipe / unexpected-frame errors under load.
+ *
+ * Those defaults suit an endpoint reachable only by the runtime. They are NOT a general
+ * hardening posture: with a per-IP ceiling this high, an endpoint exposed beyond the
+ * runtime — especially one built without
+ * {@see \Qcodr\Restate\Sdk\Endpoint\EndpointBuilder::identityKey} — can be held at many
+ * idle connections for an hour each. Pass tightened values in that case; a dead peer is
+ * still detected immediately by the socket closing.
+ *
+ * Every ceiling is validated to be positive: amphp's own signatures require `int<1, max>`,
+ * and a zero or negative ceiling would silently mean "accept nothing". The constructor
+ * takes plain ints so values may come from configuration or the environment, and fails
+ * fast naming the offending field.
+ */
+final class ServerLimits
+{
+    /** @var int<1, max> */
+    public readonly int $connectionLimit;
+
+    /** @var int<1, max> */
+    public readonly int $connectionLimitPerIp;
+
+    /** @var int<1, max> */
+    public readonly int $concurrencyLimit;
+
+    /** @var int<1, max> */
+    public readonly int $streamIdleTimeoutSeconds;
+
+    /** @var int<1, max> */
+    public readonly int $connectionIdleTimeoutSeconds;
+
+    public function __construct(
+        int $connectionLimit = 100_000,
+        int $connectionLimitPerIp = 100_000,
+        int $concurrencyLimit = 100_000,
+        int $streamIdleTimeoutSeconds = 3600,
+        int $connectionIdleTimeoutSeconds = 3600,
+    ) {
+        $this->connectionLimit = self::positive('connectionLimit', $connectionLimit);
+        $this->connectionLimitPerIp = self::positive('connectionLimitPerIp', $connectionLimitPerIp);
+        $this->concurrencyLimit = self::positive('concurrencyLimit', $concurrencyLimit);
+        $this->streamIdleTimeoutSeconds = self::positive('streamIdleTimeoutSeconds', $streamIdleTimeoutSeconds);
+        $this->connectionIdleTimeoutSeconds = self::positive(
+            'connectionIdleTimeoutSeconds',
+            $connectionIdleTimeoutSeconds,
+        );
+    }
+
+    /**
+     * @return int<1, max>
+     */
+    private static function positive(string $name, int $value): int
+    {
+        if ($value < 1) {
+            throw new RuntimeException("ServerLimits::\${$name} must be positive, got {$value}");
+        }
+
+        return $value;
+    }
+}