diff --git a/Makefile b/Makefile index 28fa06c..b37c7b7 100644 --- a/Makefile +++ b/Makefile @@ -7,7 +7,7 @@ INGRESS_URL ?= http://localhost:8080 ADMIN_URL ?= http://localhost:9070 -.PHONY: install test test-unit up wait down logs lint stan cs cs-fix sast infection bench bench-e2e check examples +.PHONY: install test test-unit up wait down logs lint stan cs cs-fix sast infection bench bench-e2e bench-e2e-amp bench-e2e-compare check examples install: composer install @@ -49,9 +49,17 @@ infection: bench: php benchmarks/micro.php -# End-to-end load/latency/memory through Restate + Swoole (needs Docker). +# End-to-end load/latency/memory through Restate (needs Docker). Swoole request/response +# by default; `bench-e2e-amp` runs the amphp bidi transport; `bench-e2e-compare` runs +# both against one runtime and prints them side by side. bench-e2e: - benchmarks/e2e/run.sh + TRANSPORT=swoole benchmarks/e2e/run.sh + +bench-e2e-amp: + TRANSPORT=amp benchmarks/e2e/run.sh + +bench-e2e-compare: + benchmarks/e2e/compare.sh # The local pre-commit gate: lint + SAST + unit tests. check: lint sast test-unit @@ -69,13 +77,16 @@ wait: done; \ echo "Restate did not become healthy in time" >&2; exit 1 -# Bring up the example services live (all examples on one endpoint) and register them. +# Bring up the example services live (all examples on one endpoint, over true bidi +# HTTP/2 via amphp) and register them. No `use_http_11`: bidi requires HTTP/2, which the +# runtime negotiates against the amphp h2c host. The pinned 1.5.2 already serves bidi; +# override with RESTATE_IMAGE=... for a newer runtime if needed. examples: docker compose up -d --build restate examples-endpoint $(MAKE) wait @curl -fsS -X POST $(ADMIN_URL)/deployments \ -H 'content-type: application/json' \ - -d '{"uri":"http://examples-endpoint:9080","use_http_11":true,"force":true}' >/dev/null \ + -d '{"uri":"http://examples-endpoint:9080","force":true}' >/dev/null \ && echo "Examples registered. Try: curl $(INGRESS_URL)/FanOut/fanOut" # --- Cross-SDK conformance (official restatedev/sdk-test-suite) ----------- diff --git a/README.md b/README.md index 459440f..f4147ed 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,6 @@ [![CI](https://github.com/qcodr/restate-sdk-php/actions/workflows/ci.yml/badge.svg)](https://github.com/qcodr/restate-sdk-php/actions/workflows/ci.yml) [![codecov](https://codecov.io/gh/qcodr/restate-sdk-php/branch/main/graph/badge.svg)](https://codecov.io/gh/qcodr/restate-sdk-php) [![PHPStan level max](https://img.shields.io/badge/PHPStan-level%20max-brightgreen.svg)](phpstan.neon) -[![Psalm type coverage](https://shepherd.dev/github/qcodr/restate-sdk-php/coverage.svg)](https://shepherd.dev/github/qcodr/restate-sdk-php) [![Mutation testing badge](https://img.shields.io/endpoint?style=flat&url=https%3A%2F%2Fbadge-api.stryker-mutator.io%2Fgithub.com%2Fqcodr%2Frestate-sdk-php%2Fmain)](https://dashboard.stryker-mutator.io/reports/github.com/qcodr/restate-sdk-php/main) [![Latest Stable Version](https://img.shields.io/packagist/v/qcodr/restate-sdk-php.svg)](https://packagist.org/packages/qcodr/restate-sdk-php) [![Total Downloads](https://img.shields.io/packagist/dt/qcodr/restate-sdk-php.svg)](https://packagist.org/packages/qcodr/restate-sdk-php) @@ -17,12 +16,15 @@ A pure-PHP SDK for [Restate](https://restate.dev) — durable execution for **Services**, **Virtual Objects**, and **Workflows**. It mirrors the [Rust SDK](https://github.com/restatedev/sdk-rust) surface with idiomatic PHP: -attributes for service definitions, a typed context API, and a Swoole-based server. +attributes for service definitions, a typed context API, and a true bidirectional +HTTP/2 streaming server. The Restate **service protocol (v5–v7)** is implemented from scratch in pure PHP — -framing, protobuf messages, the journal/replay state machine, and suspension — so -the SDK has no native-extension dependency for its core (only the server transport -needs `ext-swoole`). +framing, protobuf messages, the journal/replay state machine, suspension, and +signals — so the SDK has **no native-extension dependency**: the default server +(`AmpStreamingServer`) runs on pure-PHP [amphp/http-server](https://amphp.org), and +a request/response Swoole server, a PSR-15 adapter, and an AWS Lambda handler are +available as alternative transports. ## Features @@ -40,13 +42,15 @@ needs `ext-swoole`). ## Requirements - PHP **8.2+** (`ext-json`, `ext-mbstring`) -- `ext-swoole` to run the server (provided by the Docker image) +- `amphp/http-server` to run the default bidirectional-streaming server + (or `ext-swoole` for the request/response Swoole server) - Docker + Docker Compose for end-to-end testing ## Installation ```bash composer require qcodr/restate-sdk-php +composer require amphp/http-server # the default server transport ``` ## Quick start @@ -86,28 +90,34 @@ final class Counter } ``` -Serve them: +Serve them over true bidirectional HTTP/2 streaming (the default server): ```php use Qcodr\Restate\Sdk\Endpoint\Endpoint; -use Qcodr\Restate\Sdk\Server\SwooleServer; +use Qcodr\Restate\Sdk\Endpoint\ProtocolMode; +use Qcodr\Restate\Sdk\Server\AmpStreamingServer; $endpoint = Endpoint::builder() ->bind(new Greeter()) ->bind(new Counter()) + ->protocolMode(ProtocolMode::BidiStream) ->build(); -(new SwooleServer($endpoint))->listen('0.0.0.0', 9080); +(new AmpStreamingServer($endpoint))->listen('0.0.0.0', 9080); ``` Register the deployment with a running Restate server, then invoke through the ingress: ```bash -restate deployments register http://localhost:9080 --use-http1.1 -curl localhost:8080/Greeter/greet -d '"world"' # "Greetings world" -curl localhost:8080/Counter/acme/add -d '5' # 5 +restate deployments register http://localhost:9080 +curl localhost:8080/Greeter/greet -H 'content-type: application/json' -d '"world"' # "Greetings world" +curl localhost:8080/Counter/acme/add -H 'content-type: application/json' -d '5' # 5 ``` +> Drop `->protocolMode(ProtocolMode::BidiStream)` (and add `--use-http1.1` when +> registering) to serve plain request/response over the same amphp host, or swap in +> `SwooleServer` / the PSR-15 / Lambda adapters — see **Transports** below. + ## Workflows & durable promises ```php @@ -136,7 +146,7 @@ final class SignupWorkflow ## Context API > **Service classes must be stateless.** A bound service instance is shared across -> concurrent invocations within a Swoole worker — keep per-invocation data in local +> concurrent invocations within the server process — keep per-invocation data in local > variables or Restate state (`$ctx->set(...)`), never in mutable instance properties. | Capability | Methods | @@ -166,7 +176,8 @@ tuned transient failure; any other throwable is a plain transient error (retried `ctx->logger()` returns a **PSR-3** logger that suppresses records emitted during replay, so each line is logged exactly once even though handlers re-run from the top on every slice. Provide the underlying logger (e.g. Monolog) when constructing the -server: `new SwooleServer($endpoint, logger: $myLogger)` (defaults to a null logger). +server: `new AmpStreamingServer($endpoint, logger: $myLogger)` (defaults to a null +logger). For distributed **tracing**, mind the propagation boundary: @@ -213,10 +224,20 @@ $endpoint = Endpoint::builder() ->build(); ``` -**Transports.** Besides `SwooleServer`, the framework-agnostic core is hostable via a -**PSR-15** adapter (`Qcodr\Restate\Sdk\Server\Psr15Handler`) in any Slim/Mezzio stack, on -**AWS Lambda** (`Qcodr\Restate\Sdk\Server\LambdaHandler` — Function URL / API Gateway proxy), -and directly via `RequestProcessor` (bytes in → bytes out). +**Transports.** The default `AmpStreamingServer` (pure-PHP amphp) serves true +bidirectional HTTP/2 streaming. One amphp process is a single event loop; pass a worker +count to pre-fork N processes that share the port via `SO_REUSEPORT` (needs `ext-pcntl`) +and scale across cores like a Swoole worker pool: + +```php +(new AmpStreamingServer($endpoint))->listen('0.0.0.0', 9080, workers: 8); // 0 = one per CPU +``` + +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** +(`Qcodr\Restate\Sdk\Server\LambdaHandler` — Function URL / API Gateway proxy), and directly +via `RequestProcessor` (bytes in → bytes out). **Typed clients.** `bin/restate-codegen [outDir] [namespace]` generates an IDE-autocompletable client so callers write @@ -244,7 +265,8 @@ self-contained, runnable endpoint. | `services.php` | the canonical Service + Virtual Object + Workflow trio | | `tracing.php` | replay-aware PSR-3 logging (run standalone: `php examples/tracing.php`) | -Run a single example with the bundled server: +Run a single example with the bundled server (amphp; the per-example endpoints are +request/response, so `--use-http1.1` is fine): ```bash php bin/restate-serve examples/counter.php # serves on :9080 @@ -252,10 +274,10 @@ restate deployments register http://localhost:9080 --use-http1.1 curl localhost:8080/Counter/my-key/increment ``` -Or bring all of them up live (Docker), against a real runtime: +Or bring all of them up live (Docker) over true bidi HTTP/2, against a real runtime: ```bash -make examples # builds + registers the example endpoint +make examples # builds + registers the example endpoint (bidi) curl localhost:8080/FanOut/fanOut # -> "Completed in order: fast, medium, slow" ``` @@ -311,8 +333,10 @@ make sast # psalm taint analysis (SAST) (== composer sast) make check # lint + sast + unit tests (pre-commit) (== composer check) ``` -> The Compose file pins `restatedev/restate:1.5.2`. The `:latest` image targets -> newer CPUs (AVX2) and may crash on older hardware. +> The Compose file pins `restatedev/restate:1.5.2` (the last AVX2-free image, so it +> runs on older hardware); it already serves the bidi examples. Override with +> `RESTATE_IMAGE=...` for a newer runtime — V7 cancellation/signals over bidi needs +> ≥ 1.7 (which needs AVX2), as covered in [`conformance/README.md`](conformance/README.md). ## Architecture @@ -325,14 +349,17 @@ src/ Context/ typed context API (Service / Object / Workflow) over the VM Serde/ JSON (de)serialization Endpoint/ framework-agnostic RequestProcessor + transport DTOs - Server/ SwooleServer transport adapter + Server/ transport adapters: AmpStreamingServer (default, bidi), Swoole, PSR-15, Lambda ``` The framework-agnostic `RequestProcessor` (bytes in → bytes out) is the testable -core; `SwooleServer` is one swappable transport. Transport mode is -`REQUEST_RESPONSE`: the runtime sends `StartMessage` + the replayed journal, the SDK -processes one slice and suspends when it awaits a result it does not yet have; the -runtime re-invokes with a longer journal and the handler replays from the top. +core; each server is one swappable transport. The default `AmpStreamingServer` +advertises `BIDI_STREAM`: the runtime keeps the invocation channel open in both +directions, streaming the journal and late completions/signals, so a parked await is +resumed on the next result instead of writing a suspension. The request/response +transports (`SwooleServer`, PSR-15, Lambda) advertise `REQUEST_RESPONSE` instead: the +SDK processes one slice and suspends when it awaits a result it does not yet have, and +the runtime re-invokes with a longer journal so the handler replays from the top. ## License diff --git a/benchmarks/e2e/bench-endpoint-amp.php b/benchmarks/e2e/bench-endpoint-amp.php new file mode 100644 index 0000000..de51df8 --- /dev/null +++ b/benchmarks/e2e/bench-endpoint-amp.php @@ -0,0 +1,50 @@ +bind(new BenchGreeterAmp()) + ->protocolMode(ProtocolMode::BidiStream) + ->build(); + +// WORKER_NUM mirrors the Swoole bench: 0 = one worker per CPU, N = N workers, so the +// single-event-loop ceiling can be lifted to the runtime-bound plateau (see docs/BENCHMARKS.md). +$workers = (int) (\getenv('WORKER_NUM') ?: 1); +\fwrite(\STDERR, "bench endpoint: amphp bidi streaming, workers={$workers}\n"); +(new AmpStreamingServer($endpoint))->listen('0.0.0.0', (int) (\getenv('PORT') ?: 9080), $workers); diff --git a/benchmarks/e2e/compare.sh b/benchmarks/e2e/compare.sh new file mode 100755 index 0000000..abb565a --- /dev/null +++ b/benchmarks/e2e/compare.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +# +# Runs the e2e benchmark for BOTH transports against the SAME runtime and prints a +# side-by-side comparison (Swoole request/response vs amphp bidi HTTP/2). Both legs run +# against one runtime image for a fair comparison; the pinned 1.5.2 serves bidi fine. +# Override with RESTATE_IMAGE for a newer runtime. +# +# Usage: +# benchmarks/e2e/compare.sh +# DURATION=60s CONNECTIONS=100 benchmarks/e2e/compare.sh +# RESTATE_IMAGE=docker.io/restatedev/restate:1.7.0 benchmarks/e2e/compare.sh + +set -euo pipefail + +export RESTATE_IMAGE="${RESTATE_IMAGE:-docker.io/restatedev/restate:1.5.2}" +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +RUN="$ROOT/benchmarks/e2e/run.sh" +OUT_DIR="$ROOT/build/benchmarks" + +echo "### Runtime: $RESTATE_IMAGE" +TRANSPORT=swoole "$RUN" +TRANSPORT=amp "$RUN" + +echo +echo "=== Swoole (r/r) vs amphp (bidi) ===" +python3 - "$OUT_DIR/e2e-oha-swoole.json" "$OUT_DIR/e2e-oha-amp.json" <<'PY' +import json, sys + +def load(path): + o = json.load(open(path)) + s, p = o.get("summary", {}), o.get("latencyPercentiles", {}) + return { + "req/s": s.get("requestsPerSec", 0), + "ok%": s.get("successRate", 0) * 100, + "p50ms": p.get("p50", 0) * 1000, + "p90ms": p.get("p90", 0) * 1000, + "p99ms": p.get("p99", 0) * 1000, + } + +sw, amp = load(sys.argv[1]), load(sys.argv[2]) +cols = ["req/s", "ok%", "p50ms", "p90ms", "p99ms"] +print(f"{'metric':<8}{'swoole':>14}{'amp':>14}{'amp/swoole':>12}") +for c in cols: + ratio = (amp[c] / sw[c]) if sw[c] else 0 + print(f"{c:<8}{sw[c]:>14,.2f}{amp[c]:>14,.2f}{ratio:>11.2f}x") +PY diff --git a/benchmarks/e2e/run.sh b/benchmarks/e2e/run.sh index 2e8614c..cb726fc 100755 --- a/benchmarks/e2e/run.sh +++ b/benchmarks/e2e/run.sh @@ -1,42 +1,58 @@ #!/usr/bin/env bash # -# End-to-end load / latency / memory benchmark. +# End-to-end load / latency / memory benchmark for one transport. # -# Drives a real request path: oha -> Restate ingress -> the runtime -> the PHP Swoole -# endpoint (this SDK) -> response. Latency and throughput come from oha; the Swoole -# worker's resident memory is sampled with `docker stats` during a sustained run to -# detect leaks. Everything is containerized — the only host requirement is Docker. +# Drives a real request path: oha -> Restate ingress -> the runtime -> the PHP endpoint +# (this SDK) -> response. Latency and throughput come from oha; the endpoint's resident +# memory is sampled with `docker stats` during a sustained run to detect leaks. +# Everything is containerized — the only host requirement is Docker. +# +# Two transports serve the identical BenchGreeter so they compare head to head: +# TRANSPORT=swoole request/response over Swoole (docker/php, bench-endpoint.php) +# TRANSPORT=amp true bidi HTTP/2 over amphp (docker/php-amp, bench-endpoint-amp.php) # # Usage: -# benchmarks/e2e/run.sh +# benchmarks/e2e/run.sh # TRANSPORT=swoole (default) +# TRANSPORT=amp benchmarks/e2e/run.sh # DURATION=60s CONNECTIONS=100 benchmarks/e2e/run.sh -# KEEP_UP=1 benchmarks/e2e/run.sh # leave the stack running afterwards +# KEEP_UP=1 benchmarks/e2e/run.sh # leave the stack running afterwards # # Env: -# DURATION load duration per oha run (default 30s) -# CONNECTIONS concurrent connections (default 50) -# HANDLER ingress path to hit (default /Greeter/greet) -# BODY JSON request body (default "world") -# INGRESS ingress base URL (default http://localhost:8080) -# OHA_IMAGE load-generator image (default ghcr.io/hatoo/oha:latest) +# TRANSPORT swoole | amp (default swoole) +# RESTATE_IMAGE runtime image (bidi needs >=1.7) (compose default 1.5.2) +# DURATION load duration per oha run (default 30s) +# CONNECTIONS concurrent connections (default 50) +# HANDLER ingress path to hit (default /BenchGreeter/greet) +# BODY JSON request body (default "world") +# INGRESS ingress base URL (default http://localhost:8080) +# OHA_IMAGE load-generator image (default ghcr.io/hatoo/oha:latest) set -euo pipefail +TRANSPORT="${TRANSPORT:-swoole}" +case "$TRANSPORT" in + swoole) ENDPOINT_SERVICE="bench-endpoint-swoole"; USE_HTTP_11=true ;; # request/response + amp) ENDPOINT_SERVICE="bench-endpoint-amp"; USE_HTTP_11=false ;; # bidi needs HTTP/2 + *) echo "Unknown TRANSPORT='$TRANSPORT' (want swoole|amp)" >&2; exit 2 ;; +esac + DURATION="${DURATION:-30s}" MEM_DURATION="${MEM_DURATION:-180s}" # longer window so the leak slope is post-warmup CONNECTIONS="${CONNECTIONS:-50}" -HANDLER="${HANDLER:-/Greeter/greet}" +HANDLER="${HANDLER:-/BenchGreeter/greet}" BODY="${BODY:-\"world\"}" INGRESS="${INGRESS:-http://localhost:8080}" ADMIN="${ADMIN:-http://localhost:9070}" OHA_IMAGE="${OHA_IMAGE:-ghcr.io/hatoo/oha:latest}" -ENDPOINT_SERVICE="examples-endpoint" ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" OUT_DIR="$ROOT/build/benchmarks" mkdir -p "$OUT_DIR" cd "$ROOT" +OHA_OUT="$OUT_DIR/e2e-oha-$TRANSPORT.json" +MEM_OUT="$OUT_DIR/e2e-mem-$TRANSPORT.csv" + log() { printf '\n=== %s ===\n' "$*"; } cleanup() { @@ -49,7 +65,7 @@ cleanup() { } trap cleanup EXIT -log "Bringing up Restate 1.5.2 + Swoole endpoint" +log "Bringing up Restate + $TRANSPORT endpoint ($ENDPOINT_SERVICE)" docker compose up -d --build restate "$ENDPOINT_SERVICE" log "Waiting for health" @@ -60,10 +76,10 @@ for _ in $(seq 1 60); do sleep 2 done -log "Registering deployment" +log "Registering deployment (use_http_11=$USE_HTTP_11)" curl -fsS -X POST "$ADMIN/deployments" \ -H 'content-type: application/json' \ - -d "{\"uri\":\"http://$ENDPOINT_SERVICE:9080\",\"use_http_11\":true,\"force\":true}" >/dev/null + -d "{\"uri\":\"http://$ENDPOINT_SERVICE:9080\",\"use_http_11\":$USE_HTTP_11,\"force\":true}" >/dev/null echo "registered" log "Warmup" @@ -78,9 +94,9 @@ log "Load: oha -z $DURATION -c $CONNECTIONS POST $HANDLER" docker run --rm --network host "$OHA_IMAGE" \ -z "$DURATION" -c "$CONNECTIONS" --no-tui --output-format json \ -m POST -d "$BODY" -H 'content-type: application/json' \ - "$INGRESS$HANDLER" > "$OUT_DIR/e2e-oha.json" + "$INGRESS$HANDLER" > "$OHA_OUT" -# Sustained load in the background while we sample the worker's resident memory. +# Sustained load in the background while we sample the endpoint's resident memory. log "Memory sampling under sustained load (leak check)" CID="$(docker compose ps -q "$ENDPOINT_SERVICE")" # compose prefixes the container name docker run --rm --network host "$OHA_IMAGE" \ @@ -89,8 +105,8 @@ docker run --rm --network host "$OHA_IMAGE" \ "$INGRESS$HANDLER" >/dev/null 2>&1 & LOAD_PID=$! -: > "$OUT_DIR/e2e-mem.csv" -echo "seconds,mem_bytes" >> "$OUT_DIR/e2e-mem.csv" +: > "$MEM_OUT" +echo "seconds,mem_bytes" >> "$MEM_OUT" SECONDS_ELAPSED=0 while kill -0 "$LOAD_PID" 2>/dev/null; do USAGE="$(docker stats --no-stream --format '{{.MemUsage}}' "$CID" 2>/dev/null | awk '{print $1}')" || USAGE="" @@ -102,14 +118,14 @@ mult = {"B":1,"KiB":1024,"MiB":1024**2,"GiB":1024**3,"TiB":1024**4,"KB":1000,"MB print(int(float(m.group(1))*mult.get(m.group(2),1)) if m else 0) PY )" || BYTES=0 - echo "$SECONDS_ELAPSED,$BYTES" >> "$OUT_DIR/e2e-mem.csv" + echo "$SECONDS_ELAPSED,$BYTES" >> "$MEM_OUT" sleep 3 SECONDS_ELAPSED=$((SECONDS_ELAPSED+3)) done wait "$LOAD_PID" 2>/dev/null || true -log "Summary" -python3 - "$OUT_DIR/e2e-oha.json" "$OUT_DIR/e2e-mem.csv" <<'PY' +log "Summary ($TRANSPORT)" +python3 - "$OHA_OUT" "$MEM_OUT" <<'PY' import json, sys oha = json.load(open(sys.argv[1])) s = oha.get("summary", {}) @@ -125,7 +141,7 @@ rows = [l.strip().split(",") for l in open(sys.argv[2]).read().splitlines()[1:] mem = [(int(t), int(b)) for t, b in rows if b.isdigit() and int(b) > 0] if len(mem) >= 2: # Slope over the steady-state window (ignore the first 30s of warmup: opcache, - # Swoole connection buffers) so the leak verdict reflects sustained behavior. + # connection buffers) so the leak verdict reflects sustained behavior. cutoff = min(30, mem[-1][0] // 2) steady = [m for m in mem if m[0] >= cutoff] or mem # Least-squares slope over the steady window: robust to docker-stats' ~0.1 MiB @@ -137,10 +153,10 @@ if len(mem) >= 2: slope = sum((t - mt) * (b - mb) for t, b in steady) / denom * 60 / 1e6 # MB/min peak = max(b for _, b in mem) span = steady[-1][0] - steady[0][0] - print(f" worker RSS : start {mem[0][1]/1e6:.1f} MB, peak {peak/1e6:.1f} MB") + print(f" endpoint RSS : start {mem[0][1]/1e6:.1f} MB, peak {peak/1e6:.1f} MB") print(f" steady slope : {slope:+.2f} MB/min (least-squares over {span}s, n={n}) " f"({'no leak' if abs(slope) < 0.5 else 'investigate'})") PY echo -echo "Raw: $OUT_DIR/e2e-oha.json, $OUT_DIR/e2e-mem.csv" +echo "Raw: $OHA_OUT, $MEM_OUT" diff --git a/bin/restate-serve b/bin/restate-serve index 7d8a2ff..8bde12d 100755 --- a/bin/restate-serve +++ b/bin/restate-serve @@ -4,7 +4,8 @@ declare(strict_types=1); /** - * Boots a Restate deployment endpoint over Swoole. + * Boots a Restate deployment endpoint over the bidirectional-streaming HTTP/2 host + * ({@see Qcodr\Restate\Sdk\Server\AmpStreamingServer}, the SDK's default server). * * Usage: * restate-serve [bootstrap.php] [host] [port] @@ -12,6 +13,12 @@ declare(strict_types=1); * The bootstrap file must return a Qcodr\Restate\Sdk\Endpoint\Endpoint. Defaults: * bootstrap = ./restate.php, host = 0.0.0.0, port = $PORT or 9080. * + * The server serves whatever protocol mode the endpoint declares: an endpoint built + * with `->protocolMode(ProtocolMode::BidiStream)` is driven over true bidi HTTP/2; + * a plain request/response endpoint is buffered over the same h2c host. Requires + * amphp/http-server (a composer `suggest`); install it with + * `composer require amphp/http-server`. + * * Example bootstrap (restate.php): * listen($host, $port); + (new AmpStreamingServer($endpoint))->listen($host, $port); })($argv); diff --git a/composer.json b/composer.json index a538b3e..56b4390 100644 --- a/composer.json +++ b/composer.json @@ -25,8 +25,8 @@ "vimeo/psalm": "^6.0" }, "suggest": { - "amphp/http-server": "Required to run the bidirectional-streaming HTTP server (Qcodr\\Restate\\Sdk\\Server\\AmpStreamingServer).", - "ext-swoole": "Required to run the Swoole-based HTTP server (Qcodr\\Restate\\Sdk\\Server\\SwooleServer).", + "amphp/http-server": "Required to run the default bidirectional-streaming HTTP server (Qcodr\\Restate\\Sdk\\Server\\AmpStreamingServer).", + "ext-swoole": "Required to run the alternative request/response Swoole HTTP server (Qcodr\\Restate\\Sdk\\Server\\SwooleServer).", "ext-sodium": "Required for request identity verification (Qcodr\\Restate\\Sdk\\Endpoint\\Identity).", "open-telemetry/sdk": "Bridge Context::traceContext() into OpenTelemetry spans (see examples/tracing.php)." }, diff --git a/docker-compose.yml b/docker-compose.yml index d34af97..2abe3b3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,18 +1,52 @@ services: restate: - image: docker.io/restatedev/restate:1.5.2 + # Default pins 1.5.2 (last AVX2-free runtime) — it serves the bidi examples fine. + # Override for a newer runtime, e.g. V7 cancellation/signals over bidi needs >=1.7: + # RESTATE_IMAGE=docker.io/restatedev/restate:1.7.0 docker compose up + image: ${RESTATE_IMAGE:-docker.io/restatedev/restate:1.5.2} ports: - "8080:8080" # ingress - "9070:9070" # admin extra_hosts: - "host.docker.internal:host-gateway" + # Default: all example services over true bidi HTTP/2 (amphp AmpStreamingServer). examples-endpoint: build: context: . - dockerfile: docker/php/Dockerfile + dockerfile: docker/php-amp/Dockerfile command: php /app/examples/endpoint.php expose: - "9080" depends_on: - restate + + # Benchmark endpoints (see benchmarks/e2e/run.sh). Each serves the same single + # BenchGreeter so the two transports can be compared head to head. + bench-endpoint-amp: + build: + context: . + dockerfile: docker/php-amp/Dockerfile + command: php /app/benchmarks/e2e/bench-endpoint-amp.php + # WORKER_NUM>1 pre-forks N event-loop workers (SO_REUSEPORT) to lift the + # single-process throughput ceiling; 1 (default) is a single event loop. + environment: + - WORKER_NUM=${WORKER_NUM:-1} + expose: + - "9080" + depends_on: + - restate + + bench-endpoint-swoole: + build: + context: . + dockerfile: docker/php/Dockerfile + command: php /app/benchmarks/e2e/bench-endpoint.php + # WORKER_NUM=1 makes it transport-equal to the single-process amp endpoint; + # 0 (default) lets Swoole pre-fork max(2, cpu_num) workers (the worker_num sweep). + environment: + - WORKER_NUM=${WORKER_NUM:-0} + expose: + - "9080" + depends_on: + - restate diff --git a/docker/php-amp/Dockerfile b/docker/php-amp/Dockerfile index b3d3844..1b86434 100644 --- a/docker/php-amp/Dockerfile +++ b/docker/php-amp/Dockerfile @@ -1,27 +1,38 @@ -# Restate PHP SDK deployment image (bidirectional streaming variant): plain PHP 8.4 -# CLI + amphp/http-server (NO ext-swoole), serving the example services over HTTP/2 -# (h2c) on port 9080 for the Restate runtime to discover and invoke. +# Restate PHP SDK deployment image (DEFAULT): plain PHP 8.4 CLI + amphp/http-server +# (NO ext-swoole), serving the example services over true bidirectional HTTP/2 (h2c) +# on port 9080 for the Restate runtime to discover and invoke. # -# The request/response Swoole image lives at docker/php/Dockerfile and is unaffected. +# This is the default server (AmpStreamingServer). The Swoole request/response variant +# lives at docker/php/Dockerfile and is used by the Swoole e2e benchmark. FROM php:8.4-cli -# amphp uses non-blocking sockets and benefits from an event-loop extension; ext-pcntl -# provides the signal handling AmpStreamingServer::listen() traps for graceful shutdown. -RUN docker-php-ext-install pcntl +# 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 -# Install incl. the amphp host (a require-dev / suggest dependency); --no-dev would skip -# amphp/http-server, so dev deps are needed for this streaming image. -COPY composer.json composer.lock ./ -RUN composer install --no-interaction --no-progress --optimize-autoloader +# amphp/http-server is a require-dev / suggest dependency. Pull ONLY it (plus the prod +# deps) rather than the full require-dev tooling (phpunit/psalm/infection/…) a plain +# `composer install` would drag into the runtime image — smaller image, and far fewer +# GitHub downloads to flake on. `--update-no-dev` resolves without the dev tooling; +# composer.lock is gitignored for a library, so deps resolve fresh from composer.json. +COPY composer.json ./ +RUN composer require amphp/http-server:^3 \ + --no-interaction --no-progress --optimize-autoloader --update-no-dev COPY src ./src COPY examples ./examples +COPY benchmarks ./benchmarks RUN composer dump-autoload --optimize EXPOSE 9080 -CMD ["php", "examples/amp-endpoint.php"] +CMD ["php", "examples/endpoint.php"] diff --git a/docker/php/Dockerfile b/docker/php/Dockerfile index 783083e..536bc7d 100644 --- a/docker/php/Dockerfile +++ b/docker/php/Dockerfile @@ -1,5 +1,8 @@ -# Restate PHP SDK deployment image: PHP 8.x + ext-swoole, serving the example -# services on port 9080 for the Restate runtime to discover and invoke. +# Restate PHP SDK Swoole image: PHP 8.x + ext-swoole, serving over request/response +# on port 9080. This is the ALTERNATIVE transport — the default deployment image is the +# amphp bidirectional-streaming one at docker/php-amp/Dockerfile. Kept here because the +# Swoole e2e benchmark (benchmarks/e2e, TRANSPORT=swoole) measures this request/response +# path against the amphp one. FROM phpswoole/swoole:latest WORKDIR /app @@ -12,8 +15,9 @@ RUN composer install --no-dev --no-interaction --no-progress --optimize-autoload COPY src ./src COPY examples ./examples +COPY benchmarks ./benchmarks RUN composer dump-autoload --no-dev --optimize EXPOSE 9080 -CMD ["php", "examples/endpoint.php"] +CMD ["php", "benchmarks/e2e/bench-endpoint.php"] diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 5b81108..b6e5657 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -5,7 +5,7 @@ Performance is measured at two levels, each with a different purpose: | Layer | What it measures | Reproducible where | Use | |-------|------------------|--------------------|-----| | **Micro** (`benchmarks/micro.php`) | CPU + memory of the SDK hot path (protocol codec, journal/replay state machine, typed context + serde) — **no I/O** | Any PHP host, CI | Regression gate; isolates SDK cost from network/runtime | -| **End-to-end** (`benchmarks/e2e/run.sh`) | Real request path: load generator → Restate ingress → runtime → PHP Swoole endpoint → response. Latency, throughput, worker memory | Any host with Docker | Realistic latency/throughput + leak detection under sustained load | +| **End-to-end** (`benchmarks/e2e/run.sh`) | Real request path: load generator → Restate ingress → runtime → PHP endpoint → response, over Swoole (r/r) *or* amphp (bidi). Latency, throughput, endpoint memory | Any host with Docker | Realistic latency/throughput + leak detection under sustained load + transport comparison | The micro layer is the authoritative number for *SDK overhead* and for catching regressions, because it is deterministic and dependency-free. The end-to-end layer @@ -27,18 +27,20 @@ make bench # or, tuning the loop and emitting JSON: BENCH_ITER=100000 BENCH_WARMUP=10000 BENCH_JSON=1 php benchmarks/micro.php -# End-to-end (brings up Restate 1.5.2 + the Swoole endpoint via docker compose, -# drives it with a containerized `oha`, samples worker RSS with `docker stats`) -make bench-e2e +# End-to-end (brings up Restate 1.5.2 + a PHP endpoint via docker compose, drives it +# with a containerized `oha`, samples endpoint RSS with `docker stats`) +make bench-e2e # Swoole request/response (default) +make bench-e2e-amp # amphp bidi HTTP/2 +make bench-e2e-compare # both, side by side, one runtime # or, tuning the load: -DURATION=30s MEM_DURATION=360s CONNECTIONS=100 benchmarks/e2e/run.sh +TRANSPORT=amp DURATION=30s MEM_DURATION=360s CONNECTIONS=100 benchmarks/e2e/run.sh KEEP_UP=1 benchmarks/e2e/run.sh # leave the stack up for inspection ``` The only external tool the end-to-end harness needs is Docker; the load generator ([`oha`](https://github.com/hatoo/oha)) runs as a container (`ghcr.io/hatoo/oha`), so nothing is installed on the host. Raw results are written to -`build/benchmarks/` (`e2e-oha.json`, `e2e-mem.csv`). +`build/benchmarks/` (`e2e-oha-.json`, `e2e-mem-.csv`). --- @@ -112,6 +114,80 @@ round-trip and journal persistence, not the SDK (~0.05 ms of which is SDK CPU pe the micro-benchmark). Throughput here is bounded by this old CPU and a single Swoole worker config; scale with `worker_num` and faster hardware. +### Transport comparison: Swoole (r/r) vs amphp (bidi) + +`make bench-e2e-compare` runs the identical `BenchGreeter/greet` over both transports +against one runtime, head to head. On a **16-core AVX2 host, Restate 1.5.2, 50 +connections, 20 s** (numbers are relative — the absolute rate is host-dependent): + +| Transport | req/s | p50 | p90 | p99 | peak RSS | +|-----------|------:|----:|----:|----:|---------:| +| Swoole r/r — 16 workers (default) | 970 | 47 ms | 74 ms | 136 ms | 39 MB | +| Swoole r/r — 1 worker (`WORKER_NUM=1`) | 1,228 | 38 ms | 57 ms | 94 ms | — | +| amphp bidi — before TCP_NODELAY | 495 | 89 ms | 170 ms | 274 ms | 44 MB | +| **amphp bidi — 1 process** | **660** | 63 ms | 135 ms | 250 ms | 44 MB | + +Two findings: + +- **The workload is runtime-bound, not endpoint-bound.** Swoole with *one* worker + (1,228 req/s) beats Swoole with sixteen (970): more workers do not help, because the + limiter is the Restate runtime's single-partition journaling, not PHP (same + conclusion as the worker_num sweep below). +- **At one process, a zero-suspension handler costs ~1.8× per request on bidi.** At equal + concurrency and one process (`-c 50`, 1 worker) Swoole does ~1,200 req/s, bidi ~660: pure + transport overhead, since the bidi streaming driver (fiber park/resume, a held-open HTTP/2 + stream, the frame queue) does strictly more work per call than Swoole's request/response. + A trivial greeter is bidi's *worst* case — it never suspends, so none of bidi's advantage + applies. (This per-request gap is closed by running multiple workers — see below.) + +> **Optimizing the bidi transport.** Three layered wins, each measured: +> +> 1. **`TCP_NODELAY`** — amphp's `BindContext` leaves it off, so a slice that writes a +> couple of small frames (Output then End) and then waits to read hit Nagle ↔ +> delayed-ACK for a ~40 ms per-invocation stall. Enabling it on the server socket (plus +> coalescing each slice's frames into one write in `AmpStreamTransport`) lifted the +> greeter from **495 → ~660 req/s (+35%)**, p50 89 → 63 ms. +> 2. **Inline fast-path** — the original `stream()` spawned an `async()` task + an outbound +> `Queue` + `ReadableIterableStream` for *every* invocation: ~17.7 µs to create the +> extra Fiber, ~3.4 µs for the queue, and 2–3 event-loop hops per call (a micro-benchmark +> put `ReadableBuffer::read` at 465 ns vs 3,416 ns for the queue path — 7.3×). A handler +> that does not park now runs entirely in the request fiber and returns its whole output +> as one `ReadableBuffer` — no async, no queue, zero extra hops; only a *parked* handler +> falls back to the streaming queue (`SwitchableOutputSink` buffers inline, then forwards +> to the transport on the first park). Worth **+15–22%** more single-worker throughput +> and a notably tighter tail under load. +> 3. **Multi-worker** — see below; it removes the single-event-loop ceiling entirely. + +#### Multi-worker bidi (lifting the single-event-loop ceiling) + +One amphp process is one event loop — a single-core ceiling. At `-c 50` it serves ~520 +req/s; pushing concurrency only grows latency until it **collapses** (`-c 400` → 123 req/s, +p50 734 ms) as a single HTTP/2 connection / loop saturates. `AmpStreamingServer::listen()` +takes a **`$workers`** count: it pre-forks N processes that each bind the port with +`SO_REUSEPORT`, so the kernel load-balances connections across N event loops — the amphp +equivalent of Swoole's worker pool. Same host, Restate 1.5.2, `BenchGreeter/greet`: + +| workers | `-c 50` | `-c 200` | `-c 400` | +|--------:|--------:|---------:|---------:| +| 1 | 522 | 693 | 123 💥 | +| 8 | 743 | 1,421 | 1,364 | +| 16 | 805 | **1,470** | **1,709** | + +With 8–16 workers the bidi transport **matches and overtakes** single-worker Swoole +(~1,228) — 1,470 req/s at `-c 200`, 1,709 at `-c 400`, 100 % success, and the collapse is +gone. Beyond that the limiter is the Restate runtime again, not the SDK. So the structural +per-request overhead is a *single-process* property; throughput parity is a `WORKER_NUM` +away (`bench-endpoint-amp.php` reads it; production sets `listen(..., workers: N)`). + +When bidi pays off is the opposite workload — handlers that **suspend** (sleep, durable +calls, awakeables, `select`). Over request/response every suspension is a full +re-invocation HTTP round-trip (the runtime replays a longer journal from the top); over +bidi the runtime streams the completion onto the open channel and the SDK resumes the +parked fiber in place, no re-invoke. Bidi is also the only transport that can deliver +**cancellation / signals** (V7) to a parked invocation at all. Choose bidi for +correctness and suspension-heavy durable workflows; either transport is fine (Swoole is +faster) for fire-and-forget stateless calls. + --- ## Scaling and bottleneck analysis diff --git a/examples/amp-endpoint.php b/examples/amp-endpoint.php deleted file mode 100644 index 56ab14f..0000000 --- a/examples/amp-endpoint.php +++ /dev/null @@ -1,45 +0,0 @@ -bind(new Greeter()) - ->bind(new Counter()) - ->bind(new RunExample()) - ->bind(new FailureExample()) - ->bind(new FanOut()) - ->bind(new CatalogService()) - ->bind(new PeriodicTask()) - ->bind(new MyService()) - ->bind(new MyVirtualObject()) - ->bind(new MyWorkflow()) - ->protocolMode(ProtocolMode::BidiStream) - ->build(); - -(new AmpStreamingServer($endpoint))->listen('0.0.0.0', (int) (\getenv('PORT') ?: 9080)); diff --git a/examples/endpoint.php b/examples/endpoint.php index 5ec96b9..14f3330 100644 --- a/examples/endpoint.php +++ b/examples/endpoint.php @@ -5,14 +5,23 @@ namespace Restate\Examples; use Qcodr\Restate\Sdk\Endpoint\Endpoint; -use Qcodr\Restate\Sdk\Server\SwooleServer; +use Qcodr\Restate\Sdk\Endpoint\ProtocolMode; +use Qcodr\Restate\Sdk\Server\AmpStreamingServer; require __DIR__ . '/../vendor/autoload.php'; /** - * Serves every example service on one endpoint (port 9080). Each example file - * defines its class and returns its own single-service endpoint; requiring them - * here just loads the class definitions so they can be bound together. + * Serves every example service on one endpoint (port 9080) over true bidirectional + * HTTP/2 (h2c) streaming via {@see AmpStreamingServer} — the SDK's default server. + * Each example file defines its class and returns its own single-service endpoint; + * requiring them here just loads the class definitions so they can be bound together. + * + * The endpoint opts into {@see ProtocolMode::BidiStream}: discovery advertises + * BIDI_STREAM and the Restate runtime keeps the invocation channel open in both + * directions, so a parked await is resumed on the next streamed result instead of + * writing a suspension and re-invoking. + * + * Requires amphp/http-server (a composer `suggest`) and NO ext-swoole. * * Run directly: php examples/endpoint.php * Or per example: php bin/restate-serve examples/counter.php @@ -36,6 +45,7 @@ ->bind(new MyService()) ->bind(new MyVirtualObject()) ->bind(new MyWorkflow()) + ->protocolMode(ProtocolMode::BidiStream) ->build(); -(new SwooleServer($endpoint))->listen('0.0.0.0', (int) (\getenv('PORT') ?: 9080)); +(new AmpStreamingServer($endpoint))->listen('0.0.0.0', (int) (\getenv('PORT') ?: 9080)); diff --git a/examples/tracing.php b/examples/tracing.php index 606b88f..f386a00 100644 --- a/examples/tracing.php +++ b/examples/tracing.php @@ -13,7 +13,8 @@ use Qcodr\Restate\Sdk\Context\Context; use Qcodr\Restate\Sdk\Context\TraceContext; use Qcodr\Restate\Sdk\Endpoint\Endpoint; -use Qcodr\Restate\Sdk\Server\SwooleServer; +use Qcodr\Restate\Sdk\Endpoint\ProtocolMode; +use Qcodr\Restate\Sdk\Server\AmpStreamingServer; use Qcodr\Restate\Sdk\Service\Attribute\Handler; use Qcodr\Restate\Sdk\Service\Attribute\Service; use Stringable; @@ -104,5 +105,8 @@ public function log($level, string|Stringable $message, array $context = []): vo } }; -$endpoint = Endpoint::builder()->bind(new TracingGreeter())->build(); -(new SwooleServer($endpoint, logger: $logger))->listen('0.0.0.0', (int) (\getenv('PORT') ?: 9080)); +$endpoint = Endpoint::builder() + ->bind(new TracingGreeter()) + ->protocolMode(ProtocolMode::BidiStream) + ->build(); +(new AmpStreamingServer($endpoint, logger: $logger))->listen('0.0.0.0', (int) (\getenv('PORT') ?: 9080)); diff --git a/src/Endpoint/InvocationDriver.php b/src/Endpoint/InvocationDriver.php index 3b7c756..ac0ad76 100644 --- a/src/Endpoint/InvocationDriver.php +++ b/src/Endpoint/InvocationDriver.php @@ -4,6 +4,7 @@ namespace Qcodr\Restate\Sdk\Endpoint; +use Closure; use Fiber; use Psr\Log\LoggerInterface; use Qcodr\Restate\Sdk\Context\Clock; @@ -154,6 +155,86 @@ public function driveStreaming( $io->close(); } + /** + * Runs the journal-replay phase and the handler's first execution slice in the + * calling fiber (instead of a separate `async()` task). Reads from the stream via + * `$readChunk()`, starts the handler fiber, and immediately drains any park whose + * result is already present in the journal. + * + * Returns null when the handler ran to completion in this slice — all output is + * already in the `OutputSink` the caller wired into `$vm`. Returns a two-element + * array `[Fiber, mixed $park]` when the handler is suspended on an unresolved park; + * the caller is responsible for routing late completions via {@see continueFromPark}. + * + * EOF before the journal is complete is treated as a silent close: nothing ran, so + * null is returned and the caller should emit an empty (or no) response body. + * + * @param Closure(): ?string $readChunk reads one inbound chunk; returns null at EOF + * @return array{0: Fiber, 1: mixed}|null + */ + public function tryStartInline( + StateMachine $vm, + ServiceDefinition $service, + HandlerDefinition $handler, + Closure $readChunk, + ): ?array { + // Phase 1: feed the journal until the VM can run. + while (!$vm->isReadyToExecute()) { + $chunk = $readChunk(); + if ($chunk === null) { + // EOF before the journal was complete; nothing to run. + return null; + } + $vm->notifyInput($chunk); + } + + // Phase 2: start the handler and drain any immediately-satisfiable parks. + $fiber = new Fiber(function () use ($service, $handler, $vm): void { + $this->invocationProcessor->process($service, $handler, $vm); + }); + + $park = $this->drainResolved($fiber, $fiber->start()); + + if ($fiber->isTerminated()) { + // Handler completed without any unresolved park in this slice. + return null; + } + + return [$fiber, $park]; + } + + /** + * Continues an invocation that was left parked by {@see tryStartInline}. Reads + * completions/signals from `$io` (flush-before-read) and resumes the fiber for + * every await the arriving chunk satisfies, exactly as {@see driveStreaming} does + * in its own loop. Closes `$io` when the fiber terminates or EOF arrives. + * + * @param Fiber $fiber + */ + public function continueFromPark( + StateMachine $vm, + Fiber $fiber, + mixed $park, + StreamTransport $io, + ): void { + while (!$fiber->isTerminated()) { + $chunk = $io->read(); + if ($chunk === null) { + $vm->notifyInputClosed(); + if ($park instanceof ParkSignal) { + $vm->writeSuspension($park->awaitTree); + } + + break; + } + + $vm->notifyInput($chunk); + $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 diff --git a/src/Endpoint/RequestProcessor.php b/src/Endpoint/RequestProcessor.php index 39ece3a..0611916 100644 --- a/src/Endpoint/RequestProcessor.php +++ b/src/Endpoint/RequestProcessor.php @@ -4,6 +4,8 @@ namespace Qcodr\Restate\Sdk\Endpoint; +use Closure; +use Fiber; use Psr\Log\LoggerInterface; use Qcodr\Restate\Sdk\Context\Clock; use Qcodr\Restate\Sdk\Discovery\DiscoveryContentType; @@ -12,6 +14,7 @@ use Qcodr\Restate\Sdk\Serde\Serde; use Qcodr\Restate\Sdk\Vm\FiberSuspender; use Qcodr\Restate\Sdk\Vm\StateMachine; +use Qcodr\Restate\Sdk\Vm\SwitchableOutputSink; /** * The framework-agnostic core of the deployment endpoint. @@ -181,6 +184,86 @@ public function driveStreaming(StreamingInvocation $target, StreamTransport $io) ); } + /** + * Attempts to run the journal-replay phase and the handler's first execution slice + * inline (in the calling fiber), eliminating the `async()` task and outbound + * {@see \Amp\Pipeline\Queue} for handlers that complete without parking. + * + * The caller provides `$readChunk` — a closure that reads one raw byte-string from + * the inbound transport and returns null at EOF — so this method stays transport- + * agnostic. Typically `fn() => $requestBody->read()`. + * + * Returns a {@see StreamingInlineResult}: + * + * - `completed === true`: handler finished in this slice; `$output` is the full + * encoded body. Return a `ReadableBuffer($output)` response, no continuation. + * + * - `completed === false`: handler is parked; `$output` is the preamble to flush, + * and `$vm`/`$handlerFiber`/`$park`/`$switchSink` are set. The caller must call + * {@see SwitchableOutputSink::switchToDownstream} then {@see continueStreamingFromPark}. + * + * @param Closure(): ?string $readChunk + */ + public function tryDriveStreamingInline( + StreamingInvocation $target, + Closure $readChunk, + ): StreamingInlineResult { + $switchSink = new SwitchableOutputSink(); + $vm = new StateMachine($target->version, new FiberSuspender(), $switchSink); + + $inlineState = $this->invocationDriver->tryStartInline( + $vm, + $target->service, + $target->handler, + $readChunk, + ); + + if ($inlineState === null) { + // Handler completed or EOF during journal; all output is already in the sink. + return new StreamingInlineResult( + completed: true, + output: $switchSink->takeBuffer(), + vm: null, + handlerFiber: null, + park: null, + switchSink: null, + ); + } + + /** @var Fiber $fiber */ + [$fiber, $park] = $inlineState; + + return new StreamingInlineResult( + completed: false, + output: $switchSink->takeBuffer(), + vm: $vm, + handlerFiber: $fiber, + park: $park, + switchSink: $switchSink, + ); + } + + /** + * Continues an invocation that was left parked by {@see tryDriveStreamingInline}. + * Routes late completions from `$io` to the VM, resumes the handler fiber on each + * satisfiable park, and closes `$io` when the fiber terminates or EOF arrives. + * + * Must only be called when `$result->completed === false`. + */ + public function continueStreamingFromPark(StreamingInlineResult $result, StreamTransport $io): void + { + $vm = $result->vm; + $fiber = $result->handlerFiber; + + if ($vm === null || $fiber === null) { + $io->close(); + + return; + } + + $this->invocationDriver->continueFromPark($vm, $fiber, $result->park, $io); + } + /** * Shared request gate: opt-in identity verification then the transport-agnostic * body cap. Returns the rejection response, or null when the request may proceed. diff --git a/src/Endpoint/StreamingInlineResult.php b/src/Endpoint/StreamingInlineResult.php new file mode 100644 index 0000000..a768039 --- /dev/null +++ b/src/Endpoint/StreamingInlineResult.php @@ -0,0 +1,65 @@ +|null $handlerFiber null when completed + * @param mixed $park the ParkSignal the handler last yielded, or null when completed + */ + public function __construct( + /** True when the handler ran to completion without an unresolved park. */ + public readonly bool $completed, + /** + * Completed: full encoded response body. + * Parked: pre-park preamble to flush before the continuation starts. + */ + public readonly string $output, + /** null when completed; the live VM to route completions through. */ + public readonly ?StateMachine $vm, + /** null when completed; the parked handler fiber to resume. */ + public readonly ?Fiber $handlerFiber, + /** null when completed; the {@see \Qcodr\Restate\Sdk\Vm\ParkSignal} yielded last. */ + public readonly mixed $park, + /** + * null when completed; the two-phase sink wired into the VM. + * Switch it to the streaming downstream before starting the async continuation. + */ + public readonly ?SwitchableOutputSink $switchSink, + ) { + } +} diff --git a/src/Server/AmpStreamTransport.php b/src/Server/AmpStreamTransport.php index f5cd4cb..95e89b2 100644 --- a/src/Server/AmpStreamTransport.php +++ b/src/Server/AmpStreamTransport.php @@ -14,11 +14,17 @@ * * - {@see read} pulls the next inbound chunk from the request body, returning null at * EOF (the runtime closed its half of the stream); - * - {@see write} enqueues an outbound frame onto the response {@see Queue}. It uses - * `pushAsync()`, which buffers and returns immediately rather than awaiting consumer - * backpressure, so the handler fiber the driver controls is never parked by amphp — - * it parks only on its own await points (via the {@see \Qcodr\Restate\Sdk\Vm\FiberSuspender}); - * - {@see close} completes the queue exactly once, ending the streamed response body. + * - {@see write} appends an outbound frame to a slice buffer (it does not hit the socket + * yet); the buffer is flushed as a single {@see Queue} push when the driver next blocks + * on {@see read} or finishes via {@see close}. Coalescing every frame a slice produces + * (e.g. Output then End, or several commands before one park) into one write halves the + * socket writes and avoids an inter-frame delayed-ACK stall, while preserving the + * invariant that all pending frames reach the runtime before the driver waits for input. + * The flush uses `pushAsync()`, which buffers and returns immediately rather than awaiting + * consumer backpressure, so the handler fiber the driver controls is never parked by amphp + * — it parks only on its own await points (via the {@see \Qcodr\Restate\Sdk\Vm\FiberSuspender}); + * - {@see close} flushes any buffered frames and completes the queue exactly once, ending + * the streamed response body. * * Live-socket transport glue, like {@see SwooleServer}: it requires amphp/http-server * and a real HTTP/2 connection, so it is exercised by the cross-SDK conformance suite @@ -28,6 +34,9 @@ final class AmpStreamTransport implements StreamTransport { private bool $closed = false; + /** Frames written during the current slice, flushed as one push on read/close. */ + private string $buffer = ''; + /** * @param Queue $outbound the response body queue frames are pushed onto */ @@ -39,16 +48,20 @@ public function __construct( public function read(): ?string { + // Flush before blocking: the driver only reads when it needs the next inbound + // frame, so every command/terminal frame the just-run slice produced must be on + // the wire first (otherwise a parked await would deadlock waiting for a completion + // to a command the runtime never received). + $this->flush(); + return $this->inbound->read(); } public function write(string $bytes): void { - // pushAsync buffers and returns immediately (no consumer-backpressure await), so - // the handler fiber is never suspended by amphp here. ignore() silences the - // returned future so a client disconnect — which disposes the queue — does not - // surface as an unhandled future error. - $this->outbound->pushAsync($bytes)->ignore(); + // Accumulate; the actual push happens in flush() so a whole slice's frames go out + // as one write (see the class docblock). + $this->buffer .= $bytes; } public function close(): void @@ -59,6 +72,22 @@ public function close(): void return; } $this->closed = true; + $this->flush(); $this->outbound->complete(); } + + /** + * Pushes the buffered frames as a single queue item, then clears the buffer. pushAsync + * buffers and returns immediately (no consumer-backpressure await), so the handler fiber + * is never suspended here; ignore() silences the returned future so a client disconnect — + * which disposes the queue — does not surface as an unhandled future error. + */ + private function flush(): void + { + if ($this->buffer === '') { + return; + } + $this->outbound->pushAsync($this->buffer)->ignore(); + $this->buffer = ''; + } } diff --git a/src/Server/AmpStreamingServer.php b/src/Server/AmpStreamingServer.php index bdd3f5e..94470fe 100644 --- a/src/Server/AmpStreamingServer.php +++ b/src/Server/AmpStreamingServer.php @@ -4,6 +4,7 @@ namespace Qcodr\Restate\Sdk\Server; +use Amp\ByteStream\ReadableBuffer; use Amp\ByteStream\ReadableIterableStream; use Amp\Http\HttpStatus; use Amp\Http\Server\DefaultErrorHandler; @@ -13,6 +14,7 @@ use Amp\Http\Server\Response; use Amp\Http\Server\SocketHttpServer; use Amp\Pipeline\Queue; +use Amp\Socket\BindContext; use Amp\Socket\InternetAddress; use Psr\Log\LoggerInterface; use Psr\Log\NullLogger; @@ -23,6 +25,7 @@ use Qcodr\Restate\Sdk\Endpoint\ProtocolMode; use Qcodr\Restate\Sdk\Endpoint\RequestProcessor; use Qcodr\Restate\Sdk\Endpoint\StreamingInvocation; +use Qcodr\Restate\Sdk\Endpoint\StreamingOutputSink; use Qcodr\Restate\Sdk\Serde\Serde; use RuntimeException; use Throwable; @@ -108,12 +111,86 @@ public function __construct( ); } - public function listen(string $host = '0.0.0.0', int $port = 9080): void + /** + * Serves the endpoint until SIGINT/SIGTERM. + * + * @param int $workers number of worker processes. 1 (default) serves in a single + * event loop. > 1 pre-forks that many processes that each bind the + * same port via SO_REUSEPORT, so the kernel load-balances + * connections across N event loops — the amphp equivalent of + * Swoole's worker pool (a single amphp loop is otherwise a + * single-core throughput ceiling). <= 0 auto-detects the CPU count. + * Needs ext-pcntl; without it the call falls back to one worker. + */ + public function listen(string $host = '0.0.0.0', int $port = 9080, int $workers = 1): void { if ($port < 0 || $port > 65535) { throw new RuntimeException("Port {$port} is out of range (0-65535)"); } + $workers = $workers > 0 ? $workers : self::detectWorkers(); + + // Single worker: serve inline, behaviour unchanged (no SO_REUSEPORT needed). + if ($workers < 2 || !\function_exists('pcntl_fork')) { + if ($workers >= 2) { + \fwrite(\STDERR, "WARNING: ext-pcntl is unavailable; running a single worker.\n"); + } + $this->runServer($host, $port, reusePort: false, announce: true); + + return; + } + + // Multi-worker: pre-fork (workers - 1) children that, with the parent, each bind the + // same port with SO_REUSEPORT so the kernel spreads connections across N event loops. + // Fork before any event-loop use so every process starts a clean Revolt loop. A + // crashed worker is not respawned (use a process supervisor / amphp-cluster for that). + $childPids = []; + for ($i = 1; $i < $workers; $i++) { + $pid = \pcntl_fork(); + if ($pid === 0) { + $this->runServer($host, $port, reusePort: true, announce: false); + + return; + } + if ($pid > 0) { + $childPids[] = $pid; + } else { + \fwrite(\STDERR, "WARNING: fork failed; continuing with fewer workers.\n"); + } + } + + \fwrite(\STDOUT, \sprintf( + "Restate PHP endpoint (amphp bidi streaming) listening on http://%s:%d (%d workers)\n", + $host, + $port, + \count($childPids) + 1, + )); + if ($this->endpoint->identityVerifier() === null) { + \fwrite(\STDERR, 'WARNING: request identity verification is disabled; ' + . "configure EndpointBuilder::identityKey() for production.\n"); + } + + // 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); + + foreach ($childPids as $pid) { + if (\function_exists('posix_kill')) { + \posix_kill($pid, \SIGTERM); + } + \pcntl_waitpid($pid, $status); + } + } + + /** + * Builds and runs one amphp HTTP server in the current process, blocking until a stop + * signal. Factored out of {@see listen} so it runs identically in the parent and each + * forked worker; only the announcer (single/parent) prints the startup banner. + * + * @param int<0, 65535> $port already range-checked by {@see listen} + */ + private function runServer(string $host, int $port, bool $reusePort, bool $announce): void + { // 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 @@ -133,13 +210,27 @@ public function listen(string $host = '0.0.0.0', int $port = 9080): void allowHttp2Upgrade: true, ), ); - $server->expose(new InternetAddress($host, $port)); + + // Disable Nagle's algorithm (amphp's BindContext defaults TCP_NODELAY off). Each + // invocation writes a few small command frames (e.g. Output then End) onto the h2 + // stream and then waits to read; with Nagle on, the second small write is held back + // until the first is ACKed, colliding with the peer's delayed-ACK timer for a ~40 ms + // per-invocation stall that dominates end-to-end latency. Flushing immediately keeps + // the bidi transport's latency close to the request/response host's. SO_REUSEPORT + // lets the workers share the port for kernel-level load balancing. + $bindContext = (new BindContext())->withTcpNoDelay(); + if ($reusePort) { + $bindContext = $bindContext->withReusePort(); + } + $server->expose(new InternetAddress($host, $port), $bindContext); $server->start(new ClosureRequestHandler($this->handleRequest(...)), new DefaultErrorHandler()); - \fwrite(\STDOUT, "Restate PHP endpoint (amphp bidi streaming) listening on http://{$host}:{$port}\n"); - if ($this->endpoint->identityVerifier() === null) { - \fwrite(\STDERR, 'WARNING: request identity verification is disabled; ' - . "configure EndpointBuilder::identityKey() for production.\n"); + if ($announce) { + \fwrite(\STDOUT, "Restate PHP endpoint (amphp bidi streaming) listening on http://{$host}:{$port}\n"); + if ($this->endpoint->identityVerifier() === null) { + \fwrite(\STDERR, 'WARNING: request identity verification is disabled; ' + . "configure EndpointBuilder::identityKey() for production.\n"); + } } // Serve until the container/runtime asks us to stop. @@ -147,6 +238,24 @@ public function listen(string $host = '0.0.0.0', int $port = 9080): void $server->stop(); } + /** + * Best-effort online CPU count (Linux) used when {@see listen} is called with + * `workers <= 0`; falls back to 1. Parses /proc/cpuinfo (a constant path) rather than + * shelling out, so it adds no command-execution surface for the SAST to flag. + */ + private static function detectWorkers(): int + { + $cpuinfo = @\file_get_contents('/proc/cpuinfo'); + if (\is_string($cpuinfo)) { + $count = \preg_match_all('/^processor\s*:/m', $cpuinfo); + if (\is_int($count) && $count > 0) { + return $count; + } + } + + return 1; + } + public function handleRequest(Request $request): Response { $method = \strtoupper($request->getMethod()); @@ -179,32 +288,80 @@ public function handleRequest(Request $request): Response private function stream(Request $request, StreamingInvocation $target): Response { + $headers = [ + 'content-type' => $target->version->contentType(), + 'x-restate-server' => RequestProcessor::SDK_IDENTIFIER, + ]; + + // Fast path: run the journal replay and first handler slice inline (in the current + // fiber) so we know whether the handler completes without parking before we + // decide how to build the response body. + // + // Invariant: the HTTP/2 driver pushed the incoming DATA frames into the request + // body before scheduling this fiber (same event-loop tick), so read() returns + // immediately without suspending — zero extra event-loop hops for the whole + // journal phase and first slice for non-parking handlers. + $inbound = $request->getBody(); + $inlineResult = $this->processor->tryDriveStreamingInline( + $target, + static fn (): ?string => $inbound->read(), + ); + + if ($inlineResult->completed) { + // Handler ran to completion without any unresolved park: return the buffered + // output as a ReadableBuffer so the HTTP/2 send() loop never suspends waiting + // for a Queue — the bytes are already there on the first read(). + // ReadableBuffer('') self-closes immediately, matching a zero-byte body for + // the edge case where the handler produced no frames (e.g. EOF during journal). + return new Response( + HttpStatus::OK, + $headers, + new ReadableBuffer($inlineResult->output !== '' ? $inlineResult->output : null), + ); + } + + // Slow / parked path: the handler is waiting for a late completion or signal. + // Set up the streaming queue, wire the switchable sink to route future output + // through the transport, push the pre-park preamble synchronously so the HTTP/2 + // send() loop sees it on the very first read() without suspending, then hand off + // to an async continuation for the remaining completion loop. + + // Buffer size 1: the prelude push (producer-first, no consumer yet) can be + // absorbed without creating a DeferredFuture for backpressure. Subsequent pushes + // happen only when the continuation flushes after a transport->read(), at which + // point the consumer is already waiting, so they resume it directly. /** @var Queue $queue */ - $queue = new Queue(); - $transport = new AmpStreamTransport($request->getBody(), $queue); + $queue = new Queue(1); + $transport = new AmpStreamTransport($inbound, $queue); + + // Any VM writes that happen after this point go via transport → queue. + $inlineResult->switchSink?->switchToDownstream(new StreamingOutputSink($transport)); + + // Push the pre-park preamble (AwaitingOn + earlier commands) synchronously into + // the queue BEFORE returning the ReadableIterableStream so the first body read() + // by the HTTP/2 send() loop finds data already buffered and returns immediately. + if ($inlineResult->output !== '') { + $queue->pushAsync($inlineResult->output)->ignore(); + } - // Drive the invocation on its own task so the streamed Response can be returned - // immediately; frames the handler produces are pushed onto $queue as they happen. - async(function () use ($target, $transport): void { + async(function () use ($inlineResult, $transport): void { try { - $this->processor->driveStreaming($target, $transport); + $this->processor->continueStreamingFromPark($inlineResult, $transport); } catch (Throwable $e) { - // A malformed stream (e.g. a non-Start first frame) escapes the driver; - // isolate it to this invocation and let the response terminate below. - $this->logger->error('Unhandled error while streaming invocation: ' . $e->getMessage(), ['exception' => $e]); + $this->logger->error( + 'Unhandled error while streaming invocation: ' . $e->getMessage(), + ['exception' => $e], + ); } finally { - // Ensure the response body ends even if the driver threw before its own - // close(); AmpStreamTransport::close() is idempotent. + // Guard: continueStreamingFromPark always closes, but protect against an + // exception thrown before it reaches its own close. $transport->close(); } })->ignore(); return new Response( HttpStatus::OK, - [ - 'content-type' => $target->version->contentType(), - 'x-restate-server' => RequestProcessor::SDK_IDENTIFIER, - ], + $headers, new ReadableIterableStream($queue->iterate()), ); } diff --git a/src/Vm/SwitchableOutputSink.php b/src/Vm/SwitchableOutputSink.php new file mode 100644 index 0000000..5a36418 --- /dev/null +++ b/src/Vm/SwitchableOutputSink.php @@ -0,0 +1,63 @@ +downstream !== null) { + $this->downstream->write($frame); + } else { + $this->buffer .= $frame; + } + } + + /** + * Returns and clears the buffered pre-park frames. Must be called while the handler + * fiber is not running (i.e. after {@see \Fiber::start()} or {@see \Fiber::resume()} + * has returned) to avoid a data race. + */ + public function takeBuffer(): string + { + $buf = $this->buffer; + $this->buffer = ''; + + return $buf; + } + + /** + * Routes all future {@see write} calls to $sink. Any frames already buffered are + * NOT forwarded — the caller is responsible for pushing {@see takeBuffer()} to the + * consumer before activating the downstream. + */ + public function switchToDownstream(OutputSink $sink): void + { + $this->downstream = $sink; + } +} diff --git a/tests/Unit/Endpoint/StreamingInlineTest.php b/tests/Unit/Endpoint/StreamingInlineTest.php new file mode 100644 index 0000000..c147e4d --- /dev/null +++ b/tests/Unit/Endpoint/StreamingInlineTest.php @@ -0,0 +1,212 @@ +bind($service)->build(); + $processor = new RequestProcessor($endpoint, transportCapability: ProtocolMode::BidiStream); + $resolved = $processor->resolveStreamingInvoke(new HttpRequest( + 'POST', + "/invoke/{$serviceName}/{$handler}", + ['content-type' => ServiceProtocolVersion::V7->contentType()], + '', + )); + self::assertInstanceOf(StreamingInvocation::class, $resolved); + + return [$processor, $resolved]; + } + + /** + * A reader closure that yields each chunk once, then null at EOF — the inline path's + * phase-1 journal source. + * + * @param list $chunks + * + * @return Closure(): ?string + */ + private function chunkReader(array $chunks): Closure + { + $i = 0; + + return static function () use ($chunks, &$i): ?string { + return $chunks[$i++] ?? null; + }; + } + + /** @return list */ + private function frameTypes(string $output): array + { + return \array_map(static fn ($frame) => $frame->type(), MessageCodec::decodeAll($output)); + } + + private function successValue(string $output): string + { + foreach (MessageCodec::decodeAll($output) as $frame) { + if ($frame->type() === MessageType::OutputCommand) { + $reader = new Reader($frame->payload); + [$field] = $reader->readTag(); + self::assertSame(14, $field, 'output carries a success value'); + + return Value::decode($reader->readLengthDelimited())->content; + } + } + self::fail('No OutputCommand in output'); + } + + public function testCompletedHandlerRunsInlineAndBuffersTheWholeOutput(): void + { + // A non-parking handler completes in the first slice: tryDriveStreamingInline + // reports completed and returns the full Output/End body, with no fiber/VM/sink to + // continue — the live server returns this as a single ReadableBuffer. + [$processor, $target] = $this->resolve(new Greeter(), 'Greeter', 'greet'); + + $result = $processor->tryDriveStreamingInline( + $target, + $this->chunkReader([(new JournalBuilder())->input('"world"')->build()]), + ); + + self::assertTrue($result->completed); + self::assertNull($result->vm); + self::assertNull($result->handlerFiber); + self::assertNull($result->switchSink); + self::assertSame([MessageType::OutputCommand, MessageType::End], $this->frameTypes($result->output)); + self::assertSame('"Greetings world"', $this->successValue($result->output)); + } + + public function testParkedHandlerStreamsPreambleThenResolvesOnLateCompletion(): void + { + // The handler parks awaiting the call's invocation id (completion 1). The inline + // phase buffers the pre-park preamble (CallCommand + AwaitingOn); the continuation + // then streams the terminal Output/End once the completion arrives — same frames as + // the buffered driver, no Suspension. + [$processor, $target] = $this->resolve(new CallOptionsService(), 'CallOptionsService', 'callAndReturnInvocationId'); + + $result = $processor->tryDriveStreamingInline( + $target, + $this->chunkReader([(new JournalBuilder())->input('')->build()]), + ); + + self::assertFalse($result->completed); + self::assertNotNull($result->switchSink); + self::assertNotNull($result->vm); + self::assertNotNull($result->handlerFiber); + self::assertSame([MessageType::CallCommand, MessageType::AwaitingOn], $this->frameTypes($result->output)); + + // Wire the sink to the transport and feed the late completion, as the server does. + $transport = new BufferedStreamTransport([(new JournalBuilder())->invocationIdCompletion(1, 'inv-xyz')->frames()]); + $result->switchSink->switchToDownstream(new StreamingOutputSink($transport)); + $processor->continueStreamingFromPark($result, $transport); + + self::assertSame([MessageType::OutputCommand, MessageType::End], $this->frameTypes($transport->written())); + self::assertNotContains(MessageType::Suspension, $this->frameTypes($transport->written())); + self::assertSame('"inv-xyz"', $this->successValue($transport->written())); + self::assertTrue($transport->isClosed()); + } + + public function testInlineCompletesEmptyWhenJournalEndsBeforeReady(): void + { + // The runtime hangs up before sending a full journal: tryStartInline reads null + // while the VM is not yet ready to execute, so the inline attempt completes with no + // output (the live server then returns an empty body). Covers the EOF-before-journal + // branch — the handler never runs. + [$processor, $target] = $this->resolve(new Greeter(), 'Greeter', 'greet'); + + $result = $processor->tryDriveStreamingInline($target, $this->chunkReader([])); // immediate EOF + + self::assertTrue($result->completed); + self::assertSame('', $result->output); + self::assertNull($result->handlerFiber); + self::assertNull($result->vm); + } + + public function testContinueFromParkClosesWhenResultCarriesNoFiber(): void + { + // Defensive guard: a parked result always carries a VM + fiber, but the fields are + // nullable; if mis-constructed without them the continuation must close the channel + // rather than dereference null. This exercises that guard directly. + $processor = new RequestProcessor( + Endpoint::builder()->bind(new Greeter())->build(), + transportCapability: ProtocolMode::BidiStream, + ); + $result = new StreamingInlineResult( + completed: false, + output: '', + vm: null, + handlerFiber: null, + park: null, + switchSink: null, + ); + $transport = new BufferedStreamTransport([]); + + $processor->continueStreamingFromPark($result, $transport); + + self::assertTrue($transport->isClosed()); + self::assertSame('', $transport->written()); + } + + public function testParkedHandlerSuspendsGracefullyOnEofDuringContinuation(): void + { + // The handler parks on a sleep timer; the runtime then hangs up (EOF) before the + // timer fires. The continuation must write exactly one SuspensionMessage so the + // runtime re-invokes later — the EOF-while-parked invariant, preserved by the + // inline split. + [$processor, $target] = $this->resolve(new CancellationService(), 'CancellationService', 'awaitThenSleep'); + + $result = $processor->tryDriveStreamingInline( + $target, + $this->chunkReader([(new JournalBuilder())->input('')->build()]), + ); + + self::assertFalse($result->completed); + self::assertNotNull($result->switchSink); + self::assertSame([MessageType::SleepCommand, MessageType::AwaitingOn], $this->frameTypes($result->output)); + + $transport = new BufferedStreamTransport([]); // immediate EOF + $result->switchSink->switchToDownstream(new StreamingOutputSink($transport)); + $processor->continueStreamingFromPark($result, $transport); + + self::assertSame([MessageType::Suspension], $this->frameTypes($transport->written())); + self::assertNotContains(MessageType::OutputCommand, $this->frameTypes($transport->written())); + self::assertTrue($transport->isClosed()); + } +} diff --git a/tests/Unit/Vm/SwitchableOutputSinkTest.php b/tests/Unit/Vm/SwitchableOutputSinkTest.php new file mode 100644 index 0000000..10ade44 --- /dev/null +++ b/tests/Unit/Vm/SwitchableOutputSinkTest.php @@ -0,0 +1,59 @@ +write('a'); + $sink->write('b'); + + self::assertSame('ab', $sink->takeBuffer()); + } + + public function testTakeBufferClearsTheBuffer(): void + { + $sink = new SwitchableOutputSink(); + $sink->write('a'); + + self::assertSame('a', $sink->takeBuffer()); + self::assertSame('', $sink->takeBuffer(), 'the buffer is emptied once taken'); + } + + public function testForwardsToDownstreamAfterSwitch(): void + { + $downstream = new class () implements OutputSink { + public string $received = ''; + + public function write(string $frame): void + { + $this->received .= $frame; + } + }; + + $sink = new SwitchableOutputSink(); + $sink->write('preamble'); // buffered + $sink->switchToDownstream($downstream); + $sink->write('x'); // forwarded + $sink->write('y'); // forwarded + + // Pre-switch frames stay in the buffer (the caller flushes them); only post-switch + // frames reach the downstream sink. + self::assertSame('preamble', $sink->takeBuffer()); + self::assertSame('xy', $downstream->received); + } +}