From 491f0630b09d4faa964c5efa4a23e3a7a3139f41 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Wed, 12 Aug 2026 09:43:21 +0200 Subject: [PATCH] feat(relay-tunnel): embed the relay receiver as a native library MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Let a Restate Java service register with the relay (M:N HTTP/2 reverse-tunnel broker) with no service-code change and no separate ops artifact, by embedding the relay receiver as native code inside the JVM. The Rust engine does all relay-facing work on its own tokio runtime and bridges each forwarded request to the SDK's own HTTP/2 server over a loopback socket; the FFI boundary is control-plane only (start / status / stop). Rust (sdk-core/src/main/rust — one cdylib, alongside vm_*): * Depend on restate-sdk-shared-core at the `relay` branch with the `tunnel` feature (the engine — relay stack + loopback — now lives in shared-core). * src/relay_tunnel.rs: the relay_tunnel_{start,status,stop} C ABI over the engine's relay::{Config, Engine, Handle}. Reuses the existing FFI conventions — ForeignSlice in, a VmNewResult-style tagged-union result, an owned Slice out freed via free_buffer — rather than a bespoke free_string. Java: * dev.restate.sdk.core.RelayTunnel (+ RelayTunnelException): transport-neutral control class in the Java-17 base source set (so sdk-http-vertx can see it); builds the JSON config, parses status, and reflectively loads the FFM impl (mirrors StateMachineFactory). No pure-Java fallback — the tunnel requires JDK 23+ (FFM). * dev.restate.sdk.core.statemachine.ffm.FfmRelayTunnel (java23 overlay): the FFM downcalls. Loads the native lib but deliberately does NOT call SharedCoreNative.init(...) — that once-only tracing install is FfmStateMachine's, and a tunnel may start before any state machine exists. * dev.restate.sdk.http.vertx.RestateRelayServer.listen(endpoint, config): binds the existing Vert.x h2c server on 127.0.0.1:0 and points the tunnel at it; a shutdown hook drains the tunnel then closes the server. Swap RestateHttpServer.listen(endpoint) for this; dispatch is untouched (the forwarded :path is the SDK-relative tail, so request-identity holds). Verified: the cdylib builds, cbindgen emits the relay_tunnel_* symbols + tagged-union, jextract binds them, and the base + java23 + sdk-http-vertx sources compile clean. The loopback round-trip is proven in shared-core's tests/relay_loopback.rs. Design/rationale in development/relay-receiver-plan.md. Co-Authored-By: Claude Opus 4.8 (1M context) --- development/relay-receiver-plan.md | 255 ++++++++++++ .../dev/restate/sdk/core/RelayTunnel.java | 179 +++++++++ .../sdk/core/RelayTunnelException.java | 21 + .../core/statemachine/ffm/FfmRelayTunnel.java | 87 ++++ sdk-core/src/main/rust/Cargo.lock | 375 +++++++++++++++++- sdk-core/src/main/rust/Cargo.toml | 6 +- sdk-core/src/main/rust/src/lib.rs | 5 + sdk-core/src/main/rust/src/relay_tunnel.rs | 80 ++++ .../sdk/http/vertx/RestateRelayServer.java | 103 +++++ 9 files changed, 1091 insertions(+), 20 deletions(-) create mode 100644 development/relay-receiver-plan.md create mode 100644 sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnel.java create mode 100644 sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnelException.java create mode 100644 sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmRelayTunnel.java create mode 100644 sdk-core/src/main/rust/src/relay_tunnel.rs create mode 100644 sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateRelayServer.java diff --git a/development/relay-receiver-plan.md b/development/relay-receiver-plan.md new file mode 100644 index 000000000..ec44ff179 --- /dev/null +++ b/development/relay-receiver-plan.md @@ -0,0 +1,255 @@ +# Embedded native relay receiver — implementation plan + +**Status:** **landed** (Phases 0–3 of the Java side). **Date:** 2026-08-11. +**Scope:** the Rust engine now lives in **`restatedev/sdk-shared-core`** (branch +`relay`, behind the `tunnel` feature — see that repo's +`docs/relay-receiver-in-shared-core.md`), and this repo holds the Java binding + +Vert.x wiring. A sibling TS/napi track reuses the same engine. + +> **Implemented.** The `sdk-core` native cdylib crate +> (`sdk-core/src/main/rust`) now depends on `restate-sdk-shared-core` at +> `{ git = restatedev/sdk-shared-core, branch = "relay", features = ["tunnel"] }` +> and adds the `relay_tunnel_*` C ABI (`src/relay_tunnel.rs`) over the engine's +> `relay::{Config, Engine, Handle}` API. On the Java side: `RelayTunnel` + +> `RelayTunnelException` (base `dev.restate.sdk.core`, transport-neutral, +> reflective FFM load), `FfmRelayTunnel` (java23 FFM overlay), and +> `RestateRelayServer.listen(endpoint, config)` (sdk-http-vertx). Verified: +> the cdylib builds + cbindgen emits the symbols; jextract binds them; the base, +> java23, and sdk-http-vertx sources all compile. Remaining is operator-time — +> a live run against a real relay (Phase 0's round-trip is proven in shared-core's +> `tests/relay_loopback.rs`). + +## Goal + +Let a Restate Java service register with the **relay** (an M:N HTTP/2 +reverse-tunnel rendezvous broker) with **no service code changes** and **no +separate ops artifact**, by embedding the Rust `relay-receiver` runtime as +native code inside the JVM. The Rust core does all relay-facing work on its own +tokio runtime and bridges each forwarded request to the SDK's **own** HTTP/2 +server over a loopback socket. The FFI boundary is a control plane only +(`start`/`stop`/`status`) — no request data crosses it. + +Background on the relay protocol/roles lives in the relay repo +(`design.md`, `docs/reconnection-contract.md`, `docs/native-js-receiver.md`). +The service SDK plays the **receiver** role: it dials *out* to the relay `:8080`, +the relay drives HTTP/2 as the *client* back over that socket (the "role-flip"), +and the SDK serves the forwarded requests. + +## Architecture + +``` + relay :8080 ONE JVM process +┌──────────┐ h2 (role-flip) ┌───────────────────────────────────────────────┐ +│ relay │◄──────────────────►│ tunnel code inside librestate_sdk_core │ +└──────────┘ │ (relay-receiver + LoopbackHandler) │ + │ owns its OWN tokio runtime + threads │ + │ │ h2c dial per stream │ + │ ▼ │ + │ Vert.x HttpServer on 127.0.0.1: │ + │ → HttpEndpointRequestHandler (UNCHANGED) │ + └───────────────────────────────────────────────┘ + FFM = start / stop / status downcalls only. Data rides the loopback socket. +``` + +**Invariant:** each runtime stays sovereign on its own side of a local socket — +the Rust core never calls into the JVM per request; the JVM never touches Rust +threads. The loopback connection is indistinguishable from a normal Restate→SDK +h2c bidi connection, so SDK dispatch is untouched. + +**The contract (sdk-core ↔ transport):** *"give me a localhost h2c bidi +server's port and I'll drive it."* That is the entire coupling — `sdk-core` +knows nothing about Vert.x. + +## Key decision: one native lib, not two (measured) + +Fold the tunnel into the **existing** `librestate_sdk_core` cdylib — it exports +`vm_*` (today) plus new `relay_tunnel_*` symbols. Justified by a real build of +the tunnel cdylib (relay workspace `lto=fat`, stripped, `ring` crypto backend, +**TLS confirmed linked** via `nm`): + +| | size | +|---|---| +| VM lib today (baseline, sans-IO) | 1.60 MB | +| tunnel code, stripped | ~1.5–1.8 MB | +| tunnel code, gzipped (jar-stored) | ~0.7–0.85 MB | + +At ~1.5–2 MB the tunnel is an order of magnitude under the ~15–20 MB that would +justify a separate lib. Merging buys dedup (one copy of the +`bytes`/`http`/`tracing`/`serde` overlap) + one artifact, one loader, one +jextract header. Cost to the Lambda/serverless path is ~0.7 MB gzipped of dead +bytes it never loads — tokio only spawns on `relay_tunnel_start`, which those +deployments never call. No `NativeLibraryLoader` basename change needed. + +> If a size-sensitive Lambda build ever objects, that's when to revisit the +> split — not preemptively. + +## Component layout + +| Where | What | Vert.x dep? | +|---|---|---| +| **sdk-shared-core** `src/relay/{protocol,bridge,receiver}` (`tunnel` feature) | forked receiver runtime + protocol + bridge: dial-out, role-flip, `/whoami`, liveness, redial, multi-home (R4/R5) | — | +| **sdk-shared-core** `src/relay/loopback.rs` (`tunnel` feature) | binding-agnostic **engine**: `LoopbackHandler` (bridges each `Invocation` → h2c dial to `127.0.0.1:localPort`, **tail-only `:path`**, in-crate bridge pump) + `Engine::start` → `Handle::{status,stop}` + serde JSON `Config` | — | +| **this repo** `sdk-core/src/main/rust` (`src/relay_tunnel.rs`, extends the existing cdylib) | `relay_tunnel_*` C ABI over the engine → same `librestate_sdk_core`; depends on the shared-core `relay` branch w/ `tunnel` | — | +| **this repo** `sdk-core` (java) | jextract bindings + `FfmRelayTunnel` (java23 source set) + **`RelayTunnel`** / `RelayTunnelException` neutral control classes | **no** | +| **this repo** `sdk-http-vertx` (java) | **`RestateRelayServer.listen(endpoint, config)`** — the only Vert.x-coupled piece | yes (already) | + +The same shared-core engine is later wrapped by a **napi** shim for the TS SDK +— the engine (~80% of the work) is shared; only the binding (C-ABI vs napi) and +the SDK-side wrapper (Vert.x vs node-http2) differ. + +## FFM surface (control plane only) + +C ABI (via `cbindgen`, added to the existing header). It **reuses the existing +`vm_*` conventions** — `ForeignSlice` in, `Slice` out (freed via `free_buffer`), +a `#[repr(C, u32)]` tagged-union result — rather than a bespoke +`const char*`/`free_string` pair: + +```c +typedef struct RelayTunnelHandle RelayTunnelHandle; // opaque (like VmHandle) + +// Ok { handle } | Err { error: Slice } — mirrors VmNewResult. +typedef struct RelayTunnelStartResult { RelayTunnelStartResult_Tag tag; union { … }; } RelayTunnelStartResult; + +void relay_tunnel_start(struct ForeignSlice config_json, struct RelayTunnelStartResult *out); +void relay_tunnel_status(const struct RelayTunnelHandle *handle, struct Slice *out); // owned JSON; free_buffer +void relay_tunnel_stop(struct RelayTunnelHandle *handle); // signal shutdown, join runtime, free +``` + +Config passed as a borrowed UTF-8 JSON `ForeignSlice` (version-tolerant; no +struct-layout coupling), built in a confined `Arena`. **Zero upcalls.** +`status()` reads an `Arc` shared with the runtime task and returns owned +JSON (`{"running":bool,"last_error":…}`) the caller copies out + frees via the +existing `free_buffer`. `FfmRelayTunnel` mirrors `FfmStateMachine`'s static-init +(load the lib before the generated bindings class is initialized, per the +`loaderLookup()` ordering rule) — but deliberately does **not** call +`SharedCoreNative.init(...)`: that once-only tracing-subscriber install is +`FfmStateMachine`'s job, and a tunnel may start before any state machine exists. + +## sdk-core: neutral control class + +```java +package dev.restate.sdk.core.relay; // transport-neutral, no Vert.x + +public final class RelayTunnel implements AutoCloseable { + public record Config( + List relay, // "host:port" per node, or one name w/ many A records + String env, String tunnel, String apiKey, + int localPort, // the h2c server to dial; filled by the transport wrapper + boolean tls, // to the relay (false = h2c) + Integer connections, // R4; null = all resolved nodes + String instanceId, // R5; generated by the caller + H2Tuning h2) { /* toJson() */ } + + public record Status(int connections, String lastError) {} + + public static RelayTunnel start(Config cfg); // FfmRelayTunnel under the hood + public Status status(); + public void stop(); + @Override public void close() { stop(); } +} +``` + +## sdk-http-vertx: the Vert.x wiring + +```java +public final class RestateRelayServer { + public static RelayTunnel listen(Endpoint endpoint, RelayConfig relayConfig) { + Vertx vertx = Vertx.vertx(); + HttpServer server = RestateHttpServer.fromEndpoint(vertx, endpoint); // existing, unchanged + int port = server.listen(0, "127.0.0.1") + .toCompletionStage().toCompletableFuture().join().actualPort(); + RelayTunnel tunnel = RelayTunnel.start(relayConfig.toCoreConfig(port)); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + tunnel.stop(); server.close(); vertx.close(); + })); + return tunnel; + } +} +``` + +The service author swaps `RestateHttpServer.listen(endpoint)` → +`RestateRelayServer.listen(endpoint, relayConfig)`. Nothing else changes: +`HttpEndpointRequestHandler` reads `uri.getPath()`, which is the tail (the Rust +`LoopbackHandler` sends a tail-only `:path`), so dispatch + request-identity are +intact. + +## Packaging / CI / Lambda + +- Extend `sdk-core/src/main/rust` so the one cdylib exports both symbol sets; + extend the jextract task to bind `relay_tunnel_*`. `NativeLibraryLoader` + unchanged (same lib). +- The `native.yaml` CI matrix already cross-compiles `librestate_sdk_core` for + all release classifiers — the tunnel rides along; the VM lib grows to + ~3.2–3.4 MB/platform. +- **Lambda:** `sdk-lambda` never calls `RelayTunnel` → tokio never spawns; the + only cost is ~0.7 MB gzipped/platform of dead bytes. `sdk-core` gains **no + Vert.x dependency** (the wiring lives in `sdk-http-vertx`). + +## Phased milestones (with acceptance criteria) + +- **Phase 0 — spike / retire the make-or-break risks.** Add a minimal + `relay_tunnel_start`/`stop` + `LoopbackHandler` to `sdk-core/src/main/rust`; + extend jextract; a Java 23 harness that binds `RestateHttpServer.fromEndpoint` + on `127.0.0.1:0` and calls `start`. **Accept:** against `cargo run --bin relay + examples/relay.toml` (whoami mode), a sender request on `:9080` round-trips + relay → Rust → loopback → Vert.x → back. Proves (a) FFM start/stop + + encapsulated tokio runtime, (b) **prior-knowledge h2c bidi into the Vert.x + server** (risk #1). +- **Phase 1 — real dispatch.** Bind a real `Endpoint`; run a signed discovery + + invocation end to end. **Accept:** signed invocation → 200, request-identity + intact (tail-only path). +- **Phase 2 — lifecycle.** `RelayTunnel` + `RestateRelayServer`, `status()`, + shutdown hook, auth failure surfaced via `status().lastError`. **Accept:** + SIGTERM drains cleanly; a bad `apiKey` surfaces a readable error, no hang. +- **Phase 3 — HA parity.** Wire `connections` (R4) + `instanceId` (R5) into the + config JSON. **Accept:** multi-home + affinity match the pure-JS `connect()` + client against the relay's cluster e2e. +- **Phase 4 — packaging.** Fold `relay_tunnel_*` into the released cdylib across + the CI matrix; verify the lazy behaviour on the Lambda path; smoke-install on + linux/mac. + +## Risks + +1. **Prior-knowledge h2c bidi into Vert.x** (#1). The Rust h2 client must send + the h2 preface (not an h1→h2 upgrade), and the Vert.x server must set + `request.version() == HTTP_2` so bidi engages. Strongly expected (Restate + already talks h2c bidi to this exact server on default `HttpServerOptions`), + but verify in Phase 0 — it is make-or-break. +2. **Java 23 floor** for the FFM path. Fallback: the same cdylib built as a + binary, spawned as a subprocess over the identical loopback socket — build + only if pre-23 support is required. +3. **One-lib discipline.** The tunnel symbols must be genuinely inert unless + `relay_tunnel_start` is called (no global constructors spinning up tokio), so + the Lambda path stays load-only. + +## Scope cuts for v1 + +- **`/whoami` mode** targeting the relay (what `relay-receiver` speaks today). + `/_/start-tunnel` (real Restate Cloud tunnel server) is a later extension to + `relay-receiver` itself. +- **Loopback TCP**, not UDS. +- **No drain** capability advertised. +- **FFM only** (Java 23+); subprocess fallback deferred. + +## Open items — resolved / remaining + +Resolved (superseded by the shared-core decision): + +1. ~~`relay-receiver` must be consumable as a crate dependency~~ → the stack was + **copied into `sdk-shared-core`** (forked), so there is no `publish = false` + dependency to unblock and no cross-repo coupling. The sdk-java cdylib depends + on the shared-core `relay` branch with `features = ["tunnel"]`. +2. ~~Confirm the engine home in the relay repo~~ → the engine is + `sdk-shared-core/src/relay/loopback.rs`; both the C-ABI and (future) napi + shims depend on shared-core's `relay::{Config, Engine, Handle}`. + +Remaining (operator-time): + +3. A live end-to-end run against a real relay (`cargo run --bin relay + examples/relay.toml`, whoami mode): sender request on `:9080` → relay → tunnel + → loopback → Vert.x → back. The mechanism is proven in shared-core's + `tests/relay_loopback.rs`; this is the on-a-real-relay confirmation. +4. Swap the sdk-java cdylib's git dependency back to a crates.io + `restate-sdk-shared-core = "…"` once the `tunnel` feature ships in a release. + +*(R5 `instance_id` support — previously open — is present in `ReceiverConfig`.)* diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnel.java b/sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnel.java new file mode 100644 index 000000000..ded146221 --- /dev/null +++ b/sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnel.java @@ -0,0 +1,179 @@ +// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH +// +// This file is part of the Restate Java SDK, +// which is released under the MIT license. +// +// You can find a copy of the license in file LICENSE in the root +// directory of this repository or package, or at +// https://github.com/restatedev/sdk-java/blob/main/LICENSE +package dev.restate.sdk.core; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import org.jspecify.annotations.Nullable; + +/** + * Transport-neutral control handle for an embedded relay tunnel: the SDK registers as a + * relay receiver and the native engine bridges each forwarded request to a local HTTP/2 + * server over a loopback socket (see {@code restate_sdk_shared_core::relay}). The engine owns its + * own tokio runtime; this class only drives the control plane — {@link #start(Config)}, {@link + * #status()}, {@link #stop()}. + * + *

This class carries no transport dependency: it just needs the port of a local h2c + * server to bridge to. {@code sdk-http-vertx}'s {@code RestateRelayServer} wires a Vert.x server to + * it. + * + *

Requires JDK 23+ (the FFM native path); there is no pure-Java fallback for the tunnel. + * The native implementation is resolved reflectively (mirroring {@code StateMachineFactory}) so + * this base-level class never references the java23 FFM overlay directly. + */ +public final class RelayTunnel implements AutoCloseable { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final int FFM_MIN_JAVA_FEATURE = 23; + + /** + * The native delegate, implemented by the java23 FFM class and resolved reflectively by {@link + * #start(Config)}. Public only because the implementation lives in another package (the FFM + * overlay); not intended for direct use. + */ + public interface Native { + /** The engine status as JSON: {@code {"running": bool, "last_error": string|null}}. */ + String statusJson(); + + /** Signal a graceful shutdown, join the runtime, and free the native handle. Idempotent. */ + void stop(); + } + + private final Native delegate; + + private RelayTunnel(Native delegate) { + this.delegate = delegate; + } + + /** Start a tunnel from the given configuration. Returns once the engine's runtime is spawned. */ + public static RelayTunnel start(Config config) { + return new RelayTunnel(loadNative(config.toJson())); + } + + /** Current engine status. */ + public Status status() { + String json = delegate.statusJson(); + try { + JsonNode n = MAPPER.readTree(json); + boolean running = n.path("running").asBoolean(false); + String lastError = n.hasNonNull("last_error") ? n.get("last_error").asText() : null; + return new Status(running, lastError); + } catch (Exception e) { + throw new RelayTunnelException("could not parse tunnel status: " + json, e); + } + } + + /** Signal a graceful shutdown and join the engine's runtime. Idempotent. */ + public void stop() { + delegate.stop(); + } + + @Override + public void close() { + stop(); + } + + // ------------------------------------------------------------------------- + // Reflective FFM loader (mirrors StateMachineFactory.Loader; no legacy fallback — the tunnel + // hard-requires the native FFM path). + // ------------------------------------------------------------------------- + + private static Native loadNative(String configJson) { + if (Runtime.version().feature() < FFM_MIN_JAVA_FEATURE) { + throw new RelayTunnelException( + "The relay tunnel requires Java " + + FFM_MIN_JAVA_FEATURE + + "+ (Foreign Function & Memory API)."); + } + try { + // Load the native library first; a linkage failure here means this platform isn't supported. + Class.forName("dev.restate.sdk.core.statemachine.ffm.NativeLibraryLoader") + .getMethod("ensureLoaded") + .invoke(null); + Method start = + Class.forName("dev.restate.sdk.core.statemachine.ffm.FfmRelayTunnel") + .getMethod("start", String.class); + return (Native) start.invoke(null, configJson); + } catch (InvocationTargetException e) { + // Unwrap so a start failure (bad config, engine error) propagates as-is. + Throwable cause = e.getCause() != null ? e.getCause() : e; + if (cause instanceof RuntimeException re) throw re; + if (cause instanceof Error er) throw er; + throw new RelayTunnelException("relay tunnel start failed", cause); + } catch (ReflectiveOperationException e) { + throw new RelayTunnelException( + "native relay tunnel unavailable on this platform (" + + System.getProperty("os.name") + + " " + + System.getProperty("os.arch") + + ")", + e); + } + } + + // ------------------------------------------------------------------------- + // Config & status + // ------------------------------------------------------------------------- + + /** + * Tunnel configuration. {@code localPort} is the local h2c server to bridge forwarded requests + * to; a transport wrapper (e.g. {@code RestateRelayServer}) fills it after binding via {@link + * #withLocalPort(int)}. {@code connections} (R4 multi-homing) and {@code instanceId} (R5 + * affinity) are optional. The field names below are the JSON contract with the native engine. + */ + public record Config( + String relayAddress, + String env, + String tunnel, + String apiKey, + int localPort, + @Nullable Integer connections, + @Nullable String instanceId) { + + /** The relay coordinates, with {@code localPort} unset (0) and no optional knobs. */ + public Config(String relayAddress, String env, String tunnel, String apiKey) { + this(relayAddress, env, tunnel, apiKey, 0, null, null); + } + + public Config withLocalPort(int localPort) { + return new Config(relayAddress, env, tunnel, apiKey, localPort, connections, instanceId); + } + + public Config withConnections(int connections) { + return new Config(relayAddress, env, tunnel, apiKey, localPort, connections, instanceId); + } + + public Config withInstanceId(String instanceId) { + return new Config(relayAddress, env, tunnel, apiKey, localPort, connections, instanceId); + } + + /** Serialise to the JSON wire shape the native engine's {@code Config} deserialises. */ + String toJson() { + ObjectNode n = MAPPER.createObjectNode(); + n.put("relay_addr", relayAddress); + n.put("env", env); + n.put("tunnel", tunnel); + n.put("api_key", apiKey); + n.put("local_port", localPort); + if (connections != null) { + n.put("connections", connections); + } + if (instanceId != null) { + n.put("instance_id", instanceId); + } + return n.toString(); + } + } + + /** A snapshot of the tunnel's runtime status. */ + public record Status(boolean running, @Nullable String lastError) {} +} diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnelException.java b/sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnelException.java new file mode 100644 index 000000000..7ab4058dc --- /dev/null +++ b/sdk-core/src/main/java/dev/restate/sdk/core/RelayTunnelException.java @@ -0,0 +1,21 @@ +// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH +// +// This file is part of the Restate Java SDK, +// which is released under the MIT license. +// +// You can find a copy of the license in file LICENSE in the root +// directory of this repository or package, or at +// https://github.com/restatedev/sdk-java/blob/main/LICENSE +package dev.restate.sdk.core; + +/** Thrown when starting or driving a {@link RelayTunnel} fails. */ +public class RelayTunnelException extends RuntimeException { + + public RelayTunnelException(String message) { + super(message); + } + + public RelayTunnelException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmRelayTunnel.java b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmRelayTunnel.java new file mode 100644 index 000000000..077ad6b99 --- /dev/null +++ b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmRelayTunnel.java @@ -0,0 +1,87 @@ +// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH +// +// This file is part of the Restate Java SDK, +// which is released under the MIT license. +// +// You can find a copy of the license in file LICENSE in the root +// directory of this repository or package, or at +// https://github.com/restatedev/sdk-java/blob/main/LICENSE +package dev.restate.sdk.core.statemachine.ffm; + +import dev.restate.sdk.core.RelayTunnel; +import dev.restate.sdk.core.RelayTunnelException; +import dev.restate.sdk.core.statemachine.ffm.generated.*; +import java.lang.foreign.Arena; +import java.lang.foreign.MemorySegment; + +/** + * Panama/FFM (JDK 23+) implementation of {@link RelayTunnel.Native}, driving the {@code + * relay_tunnel_*} C ABI in the native {@code librestate_sdk_core} library (see {@code + * sdk-core/src/main/rust/src/relay_tunnel.rs}). The boundary is control-plane only; every forwarded + * request rides a loopback socket inside the engine's own tokio runtime. + * + *

Unlike {@link FfmStateMachine}, the static initializer only loads the native library — it does + * not call {@code SharedCoreNative.init(...)}. That installs the process-global tracing + * subscriber and must run exactly once; {@link FfmStateMachine} owns that call (a tunnel can start + * before any state machine exists, so calling it here would risk a double-init panic when the first + * invocation later initializes {@code FfmStateMachine}). + * + *

Resolved reflectively by {@link RelayTunnel}, so it is never referenced by the Java-17 base + * source set. Control calls ({@link #statusJson()} / {@link #stop()}) are serialized on the + * instance so a concurrent {@code stop} can never free the handle out from under a {@code status}. + */ +public final class FfmRelayTunnel implements RelayTunnel.Native { + + static { + // Load the native library BEFORE SharedCoreNative is class-initialized so its + // SymbolLookup.loaderLookup() resolves the relay_tunnel_* symbols. (No init() here — see the + // class doc.) + NativeLibraryLoader.ensureLoaded(); + } + + private final MemorySegment handle; + private boolean stopped = false; + + private FfmRelayTunnel(MemorySegment handle) { + this.handle = handle; + } + + /** Parse the JSON config natively and start the engine, or throw on failure. */ + public static FfmRelayTunnel start(String configJson) { + try (Arena arena = Arena.ofConfined()) { + MemorySegment out = RelayTunnelStartResult.allocate(arena); + SharedCoreNative.relay_tunnel_start(FfmEncoding.foreignUtf8(arena, configJson), out); + if (RelayTunnelStartResult.tag(out) == SharedCoreNative.RelayTunnelStartResult_Err()) { + String err = + FfmEncoding.takeSliceString( + RelayTunnelStartResult_Err_Body.error(RelayTunnelStartResult.err(out))); + throw new RelayTunnelException(err.isEmpty() ? "relay tunnel start failed" : err); + } + MemorySegment handle = RelayTunnelStartResult_Ok_Body.handle(RelayTunnelStartResult.ok(out)); + return new FfmRelayTunnel(handle); + } + } + + @Override + public synchronized String statusJson() { + if (stopped) { + return "{\"running\":false,\"last_error\":null}"; + } + try (Arena arena = Arena.ofConfined()) { + MemorySegment out = FfmEncoding.allocateSliceStruct(arena); + SharedCoreNative.relay_tunnel_status(handle, out); + return FfmEncoding.takeSliceString(out); + } + } + + @Override + public synchronized void stop() { + if (stopped) { + return; + } + // Set before the call: relay_tunnel_stop frees the handle (Box::from_raw), so it must never run + // twice, even if the native call itself throws. + stopped = true; + SharedCoreNative.relay_tunnel_stop(handle); + } +} diff --git a/sdk-core/src/main/rust/Cargo.lock b/sdk-core/src/main/rust/Cargo.lock index 30836a3d3..f4e0279a1 100644 --- a/sdk-core/src/main/rust/Cargo.lock +++ b/sdk-core/src/main/rust/Cargo.lock @@ -47,7 +47,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -58,7 +58,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -67,6 +67,23 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "base64" version = "0.22.1" @@ -109,11 +126,21 @@ dependencies = [ "quote", "serde", "serde_json", - "syn", + "syn 2.0.117", "tempfile", "toml", ] +[[package]] +name = "cc" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -172,7 +199,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -181,12 +208,47 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "find-msvc-tools" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + [[package]] name = "getrandom" version = "0.4.2" @@ -200,6 +262,25 @@ dependencies = [ "wasip3", ] +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -227,6 +308,16 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + [[package]] name = "id-arena" version = "2.3.0" @@ -311,13 +402,24 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -351,7 +453,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.117", ] [[package]] @@ -383,7 +485,7 @@ dependencies = [ "itertools", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -420,18 +522,24 @@ checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" [[package]] name = "restate-sdk-shared-core" -version = "7.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8bd1f1d1961ac32aea52923f089e668a07630ca9c4edcadf18d38bc24e5f14c0" +version = "7.0.2" +source = "git+https://github.com/restatedev/sdk-shared-core.git?branch=relay#7789348906c6ae7d2ed6719043c4a2ff4b99b9d0" dependencies = [ + "async-trait", "base64", "bytes", "bytes-utils", + "h2", + "http", "pastey", "prost", + "rustls-pki-types", "serde", + "serde_json", "strum", "thiserror", + "tokio", + "tokio-rustls", "tracing", "tracing-subscriber", ] @@ -443,10 +551,25 @@ dependencies = [ "bytes", "cbindgen", "restate-sdk-shared-core", + "serde_json", "tracing", "tracing-subscriber", ] +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rustix" version = "1.1.4" @@ -457,7 +580,41 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", ] [[package]] @@ -493,7 +650,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -527,12 +684,34 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "strsim" version = "0.11.1" @@ -557,9 +736,15 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.117" @@ -571,6 +756,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "tempfile" version = "3.27.0" @@ -578,10 +774,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom", + "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -601,7 +797,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -613,6 +809,56 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "0.8.23" @@ -673,7 +919,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -727,6 +973,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "utf8parse" version = "0.2.2" @@ -739,6 +991,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasip2" version = "1.0.3+wasi-0.2.9" @@ -797,6 +1055,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -806,6 +1073,70 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + [[package]] name = "winnow" version = "0.7.15" @@ -851,7 +1182,7 @@ dependencies = [ "heck 0.5.0", "indexmap", "prettyplease", - "syn", + "syn 2.0.117", "wasm-metadata", "wit-bindgen-core", "wit-component", @@ -867,7 +1198,7 @@ dependencies = [ "prettyplease", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wit-bindgen-core", "wit-bindgen-rust", ] @@ -909,6 +1240,12 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zmij" version = "1.0.21" diff --git a/sdk-core/src/main/rust/Cargo.toml b/sdk-core/src/main/rust/Cargo.toml index 1ea2859f5..32fd171e3 100644 --- a/sdk-core/src/main/rust/Cargo.toml +++ b/sdk-core/src/main/rust/Cargo.toml @@ -9,10 +9,14 @@ name = "restate_sdk_core" crate-type = ["cdylib"] [dependencies] -restate-sdk-shared-core = { version = "7.0.1", features = ["tracing_pretty"] } +# Pinned to the `relay` branch, which adds the embedded relay-tunnel receiver +# behind the `tunnel` feature (the `relay_tunnel_*` C ABI below wraps it). Swap +# back to a crates.io `version = "…"` once that lands in a release. +restate-sdk-shared-core = { git = "https://github.com/restatedev/sdk-shared-core.git", branch = "relay", features = ["tracing_pretty", "tunnel"] } bytes = "1.9" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] } +serde_json = "1" [build-dependencies] cbindgen = "0.27" diff --git a/sdk-core/src/main/rust/src/lib.rs b/sdk-core/src/main/rust/src/lib.rs index 14629dfe6..f4604036d 100644 --- a/sdk-core/src/main/rust/src/lib.rs +++ b/sdk-core/src/main/rust/src/lib.rs @@ -20,6 +20,11 @@ mod logging; mod mem; +/// The relay-tunnel C ABI (`relay_tunnel_*`). `pub` so its `#[repr(C)]` result +/// type is part of the crate's reachable surface (matching the `vm_*` types in +/// this file) — otherwise its FFI-only, write-then-read-by-Java fields trip the +/// dead-code lint. +pub mod relay_tunnel; pub use logging::AbiLogLevel; pub use mem::{ForeignSlice, Slice}; diff --git a/sdk-core/src/main/rust/src/relay_tunnel.rs b/sdk-core/src/main/rust/src/relay_tunnel.rs new file mode 100644 index 000000000..53776cb52 --- /dev/null +++ b/sdk-core/src/main/rust/src/relay_tunnel.rs @@ -0,0 +1,80 @@ +//! C-ABI wrapper around the embedded relay-tunnel engine +//! (`restate_sdk_shared_core::relay`), compiled into the same `librestate_sdk_core` +//! cdylib as the `vm_*` surface and called from Java via Panama FFM. +//! +//! The boundary is control-plane only — start / status / stop. Every forwarded +//! request rides a loopback socket inside the engine's own tokio runtime, never +//! this FFI boundary. Conventions mirror the `vm_*` surface: +//! +//! - The running tunnel is an opaque `*mut RelayTunnelHandle` from +//! `relay_tunnel_start`, released by `relay_tunnel_stop` (a plain `Box`). +//! - Config crosses **in** as a borrowed [`ForeignSlice`] of UTF-8 JSON (Java +//! owns it); `relay_tunnel_start` writes a `#[repr(C, u32)]` tagged-union +//! `RelayTunnelStartResult` (`Ok { handle }` / `Err { error }`). +//! - `relay_tunnel_status` returns Rust-owned JSON as a [`Slice`] the caller +//! frees with the existing `free_buffer` (no bespoke free fn needed). +//! +//! Java serializes access per handle (one thread at a time, no reentrancy), so +//! no internal locking is required here. + +use crate::{assert_not_null, write_out, ForeignSlice, Slice}; +use restate_sdk_shared_core::relay::{Config, Engine, Handle}; + +/// Opaque handle to a running tunnel. cbindgen emits it as an opaque struct +/// (like `VmHandle`) since it is only ever referenced behind a pointer. +pub struct RelayTunnelHandle { + handle: Handle, +} + +/// Result of `relay_tunnel_start`: the opaque handle on success, or an owned +/// UTF-8 error message (`Slice`, freed via `free_buffer`) on failure. +#[repr(C, u32)] +pub enum RelayTunnelStartResult { + Ok { handle: *mut RelayTunnelHandle }, + Err { error: Slice }, +} + +/// Parse the JSON config and start the engine. `config_json` is borrowed Java +/// memory (read in-call, never freed here). On success the caller owns the +/// returned handle and must release it with `relay_tunnel_stop`. +#[export_name = "relay_tunnel_start"] +pub unsafe extern "C" fn _relay_tunnel_start( + config_json: ForeignSlice, + out: *mut RelayTunnelStartResult, +) { + let result = match relay_tunnel_start(config_json.as_slice()) { + Ok(handle) => RelayTunnelStartResult::Ok { + handle: Box::into_raw(Box::new(RelayTunnelHandle { handle })), + }, + Err(message) => RelayTunnelStartResult::Err { + error: Slice::from_string(message), + }, + }; + write_out(out, result); +} + +#[inline] +fn relay_tunnel_start(config_json: &[u8]) -> Result { + let config: Config = serde_json::from_slice(config_json) + .map_err(|e| format!("invalid tunnel config JSON: {e}"))?; + Engine::start(config).map_err(|e| e.to_string()) +} + +/// Read the tunnel status as owned UTF-8 JSON (`{"running":bool,"last_error":…}`). +/// The returned `Slice` is freed by the caller via `free_buffer`. +#[export_name = "relay_tunnel_status"] +pub unsafe extern "C" fn _relay_tunnel_status(handle: *const RelayTunnelHandle, out: *mut Slice) { + assert_not_null(handle); + let json = (*handle).handle.status_json(); + write_out(out, Slice::from_string(json)); +} + +/// Signal a graceful shutdown, join the engine's runtime, and free the handle. +/// After this call the handle pointer is dangling and must not be reused. +#[export_name = "relay_tunnel_stop"] +pub unsafe extern "C" fn _relay_tunnel_stop(handle: *mut RelayTunnelHandle) { + assert_not_null(handle); + let mut boxed = Box::from_raw(handle); + boxed.handle.stop(); + drop(boxed); +} diff --git a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateRelayServer.java b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateRelayServer.java new file mode 100644 index 000000000..c7a8b0fec --- /dev/null +++ b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateRelayServer.java @@ -0,0 +1,103 @@ +// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH +// +// This file is part of the Restate Java SDK, +// which is released under the MIT license. +// +// You can find a copy of the license in file LICENSE in the root +// directory of this repository or package, or at +// https://github.com/restatedev/sdk-java/blob/main/LICENSE +package dev.restate.sdk.http.vertx; + +import dev.restate.sdk.core.RelayTunnel; +import dev.restate.sdk.endpoint.Endpoint; +import io.vertx.core.Vertx; +import io.vertx.core.http.HttpServer; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +/** + * Serve a Restate {@link Endpoint} through an embedded relay tunnel instead of a public + * port. + * + *

Boots a normal Vert.x HTTP/2 (h2c) server bound to {@code 127.0.0.1} on a random port — the + * same server {@link RestateHttpServer} builds, so request dispatch is untouched — then starts a + * {@link RelayTunnel} pointed at that port. The tunnel registers as a relay receiver and bridges + * each forwarded request over the loopback socket into this server. The service author swaps {@code + * RestateHttpServer.listen(endpoint)} for {@code RestateRelayServer.listen(endpoint, config)}; the + * service code and {@code HttpEndpointRequestHandler} are unchanged (the forwarded {@code :path} is + * the SDK-relative tail, so paths and request-identity are intact). + * + *

Requires JDK 23+ (the native FFM tunnel; see {@link RelayTunnel}). + * + * 

+ * RelayTunnel tunnel = RestateRelayServer.listen(
+ *     endpoint,
+ *     new RelayTunnel.Config("relay.example:8080", "myenv", "mytunnel", "api-key"));
+ *
+ */ +public final class RestateRelayServer { + + private static final Logger LOG = LogManager.getLogger(RestateRelayServer.class); + + private RestateRelayServer() {} + + /** + * Bind the local Vert.x server, start the tunnel against it, and register a JVM shutdown hook + * that stops the tunnel and closes the server. Blocks until the local server is bound. Returns + * the live {@link RelayTunnel} (also {@link AutoCloseable}) for explicit lifecycle control. + */ + public static RelayTunnel listen(Endpoint endpoint, RelayTunnel.Config config) { + Vertx vertx = Vertx.vertx(); + HttpServer server = RestateHttpServer.fromEndpoint(vertx, endpoint); + + int port; + try { + port = + server + .listen(0, "127.0.0.1") + .toCompletionStage() + .toCompletableFuture() + .join() + .actualPort(); + } catch (RuntimeException e) { + vertx.close(); + throw e; + } + LOG.info("Relay-tunnel local HTTP/2 server listening on 127.0.0.1:{}", port); + + RelayTunnel tunnel; + try { + tunnel = RelayTunnel.start(config.withLocalPort(port)); + } catch (RuntimeException e) { + server.close(); + vertx.close(); + throw e; + } + + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + try { + tunnel.stop(); + } catch (Throwable t) { + LOG.warn("Error stopping relay tunnel on shutdown", t); + } + server.close(); + vertx.close(); + }, + "restate-relay-shutdown")); + + LOG.info( + "Relay tunnel started for env='{}' tunnel='{}' against relay '{}'", + config.env(), + config.tunnel(), + config.relayAddress()); + return tunnel; + } + + /** Like {@link #listen(Endpoint, RelayTunnel.Config)}. */ + public static RelayTunnel listen(Endpoint.Builder endpointBuilder, RelayTunnel.Config config) { + return listen(endpointBuilder.build(), config); + } +}