From dd22d488484a1014569af8f2bc5dd9918303f4bb Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Tue, 7 Jul 2026 12:38:36 +0200 Subject: [PATCH 1/3] Avoid loading the last vm error when the state machine has been closed anyway. --- .../restate/sdk/core/HandlerContextImpl.java | 17 ++---- .../statemachine/ffm/FfmStateMachine.java | 57 +++++++++---------- 2 files changed, 32 insertions(+), 42 deletions(-) diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java b/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java index 4c9840a6..5cc47aa5 100644 --- a/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java +++ b/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java @@ -417,7 +417,7 @@ public void pollAsyncResult(AsyncResultInternal asyncResult) { try { this.pumpOutput(); this.pollAsyncResultInner(asyncResult); - } catch (Exception e) { + } catch (Throwable e) { this.failWithoutContextSwitch(e); } } @@ -428,10 +428,7 @@ private void pollAsyncResultInner(AsyncResultInternal asyncResult) { try { asyncResult.tryComplete(this::takeNotification); } catch (Throwable e) { - // This can happen if the state machine was closed in the meantime. - if (!ExceptionUtils.containsAbortedExecutionException(e)) { - this.failWithoutContextSwitch(e); - } + this.failWithoutContextSwitch(e); asyncResult.publicFuture().completeExceptionally(AbortedExecutionException.INSTANCE); return; } @@ -447,11 +444,7 @@ private void pollAsyncResultInner(AsyncResultInternal asyncResult) { try { response = this.stateMachine.doAwait(asyncResult); } catch (Throwable e) { - // doAwait sneaky-throws AbortedExecutionException on suspension. In this case, no need to - // fail twice. - if (!ExceptionUtils.containsAbortedExecutionException(e)) { - this.failWithoutContextSwitch(e); - } + this.failWithoutContextSwitch(e); asyncResult.publicFuture().completeExceptionally(AbortedExecutionException.INSTANCE); return; } @@ -480,7 +473,7 @@ Optional takeNotification(int handle) { public void proposeRunSuccess(int runHandle, Slice toWrite) { try { this.stateMachine.proposeRunCompletion(runHandle, toWrite); - } catch (Exception e) { + } catch (Throwable e) { this.failWithoutContextSwitch(e); } this.pumpOutput(); @@ -499,7 +492,7 @@ public void proposeRunFailure( } else { this.stateMachine.proposeRunCompletion(runHandle, throwable, attemptDuration, retryPolicy); } - } catch (Exception e) { + } catch (Throwable e) { this.failWithoutContextSwitch(e); } this.pumpOutput(); diff --git a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java index 9e220966..03b9df73 100644 --- a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java +++ b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java @@ -40,9 +40,10 @@ *

Each call is a direct FFM downcall. Results come back either register-packed into the scalar * return value (the invocation state, plus a handle for handle-returning calls) or, when too wide * to pack, written into a caller-provided out-param struct whose first field is the state. Owned - * {@link Slice}s are copied out and freed by the caller. On error the state field carries the - * {@link SharedCoreNative#ERROR_STATE()} sentinel; the detail is then fetched via {@code - * vm_take_last_vm_error} and thrown as a {@link ProtocolException}. + * {@link Slice}s are copied out and freed by the caller. On the {@link + * SharedCoreNative#ERROR_STATE()} sentinel the core has already written its terminal message and + * closed, so we sneaky-throw {@link AbortedExecutionException} ({@link #abortAfterCoreClosed()}); + * construction and {@code isReadyToExecute} errors surface as a {@link ProtocolException}. * *

An instance is driven by a single thread at a time (no reentrancy, per the contract); only * {@link #state()} — a volatile read — is safe from any thread. Each downcall allocates its @@ -186,12 +187,9 @@ public AwaitResult doAwait(UnresolvedFuture future) { packed = SharedCoreNative.vm_do_await(vmHandle, futureSeg); } - // Register-packed result (a bare u64, no out-param struct): the low byte is the variant tag, - // the - // high 32 bits carry the ExecuteRun run handle. On error the core stashed the detail and we - // fetch - // it via vm_take_last_vm_error. Suspension isn't a variant here — the driver surfaces it by - // sneaky-throwing AbortedExecutionException per the StateMachine contract. + // Register-packed u64: low byte = variant tag, high 32 bits = ExecuteRun handle. The ERROR + // variant also covers suspension (the core folds it in); either way the core already wrote its + // terminal message and closed, so we abort cleanly rather than fetch the discarded detail. int variant = (int) packed; if (variant == SharedCoreNative.AWAIT_VARIANT_ANY_COMPLETED()) { return AwaitResult.ANY_COMPLETED; @@ -202,22 +200,11 @@ public AwaitResult doAwait(UnresolvedFuture future) { } else if (variant == SharedCoreNative.AWAIT_VARIANT_CANCEL_SIGNAL_RECEIVED()) { return AwaitResult.CANCEL_SIGNAL_RECEIVED; } else if (variant == SharedCoreNative.AWAIT_VARIANT_ERROR()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } throw new IllegalStateException("Unknown do_await variant: " + variant); } - /** - * Fetches + maps the error the core stashed on the last failing packed-result call (cold path). - */ - private ProtocolException takeLastVmError() { - try (Arena arena = Arena.ofConfined()) { - MemorySegment err = VmError.allocate(arena); - SharedCoreNative.vm_take_last_vm_error(vmHandle, err); - return closeVmAndMapError(err); - } - } - @Override public @Nullable NotificationValue takeNotification(int handle) { verifyNotFreed(); @@ -293,7 +280,7 @@ public Input input() { SharedCoreNative.vm_sys_input(vmHandle, out); int state = InputResult.state(out); if (state == SharedCoreNative.ERROR_STATE()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } updateState(state); @@ -429,7 +416,7 @@ public CallHandle call( SharedCoreNative.vm_sys_call(vmHandle, args, out); int state = CallResult.state(out); if (state == SharedCoreNative.ERROR_STATE()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } updateState(state); return new CallHandle(CallResult.invocation_id_handle(out), CallResult.result_handle(out)); @@ -469,7 +456,7 @@ public Awakeable awakeable() { SharedCoreNative.vm_sys_awakeable(vmHandle, out); int state = AwakeableResult.state(out); if (state == SharedCoreNative.ERROR_STATE()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } updateState(state); return new Awakeable( @@ -592,7 +579,7 @@ public RunResultHandle run(String name) { SharedCoreNative.vm_sys_run(vmHandle, FfmEncoding.foreignUtf8(arena, name), out); int state = RunResult.state(out); if (state == SharedCoreNative.ERROR_STATE()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } updateState(state); return new RunResultHandle(RunResult.replayed(out) != 0, RunResult.handle(out)); @@ -715,12 +702,12 @@ public void end() { /** * Decodes a register-packed handle result ({@code u64}): low 32 bits = state (or the {@link * SharedCoreNative#ERROR_STATE()} sentinel), high 32 bits = handle. On the error sentinel we - * fetch + throw the stashed error; otherwise update the cached state and return the handle. + * {@link #abortAfterCoreClosed()}; otherwise update the cached state and return the handle. */ private int parseHandleResult(long packed) { int state = (int) packed; if (state == SharedCoreNative.ERROR_STATE()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } updateState(state); return (int) (packed >>> 32); @@ -728,12 +715,12 @@ private int parseHandleResult(long packed) { /** * Applies a register-packed empty result (a bare {@code u32}): the new state, or the {@link - * SharedCoreNative#ERROR_STATE()} sentinel. On the error sentinel we fetch + throw the stashed - * error; otherwise update the cached state. + * SharedCoreNative#ERROR_STATE()} sentinel. On the error sentinel we {@link + * #abortAfterCoreClosed()}; otherwise update the cached state. */ private void consumeEmptyResult(int packed) { if (packed == SharedCoreNative.ERROR_STATE()) { - throw takeLastVmError(); + abortAfterCoreClosed(); } updateState(packed); } @@ -766,6 +753,16 @@ private void verifyNotFreed() { } } + /** + * Mark the state machine as closed and throw AbortedExecutionException. + * + *

A subsequent notifyError will no-op. + */ + private void abortAfterCoreClosed() { + cachedState = InvocationState.CLOSED; + AbortedExecutionException.sneakyThrow(); + } + private static String formatThrowableMessage(Throwable throwable) { return throwable.toString(); } From 261843d7f7e1d6909585e773b0b15fd102be04d3 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Tue, 7 Jul 2026 12:40:32 +0200 Subject: [PATCH 2/3] Revert example changes --- .../java/my/restate/sdk/examples/Greeter.java | 31 +------------------ 1 file changed, 1 insertion(+), 30 deletions(-) diff --git a/examples/src/main/java/my/restate/sdk/examples/Greeter.java b/examples/src/main/java/my/restate/sdk/examples/Greeter.java index 7aae6d22..d807f4db 100644 --- a/examples/src/main/java/my/restate/sdk/examples/Greeter.java +++ b/examples/src/main/java/my/restate/sdk/examples/Greeter.java @@ -13,12 +13,6 @@ import dev.restate.sdk.annotation.Service; import dev.restate.sdk.endpoint.Endpoint; import dev.restate.sdk.http.vertx.RestateHttpServer; -import io.vertx.core.AbstractVerticle; -import io.vertx.core.Promise; -import io.vertx.core.Vertx; -import io.vertx.core.VertxOptions; -import io.vertx.core.http.Http2Settings; -import io.vertx.core.http.HttpServerOptions; import java.time.Duration; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -46,29 +40,6 @@ public GreetingResponse greet(Greeting req) { } public static void main(String[] args) { - var vertxOptions = new VertxOptions(); - var eventLoopPoolSize = vertxOptions.getEventLoopPoolSize(); - var vertx = Vertx.vertx(new VertxOptions()); - var httpServerOptions = - new HttpServerOptions().setInitialSettings(new Http2Settings().setMaxConcurrentStreams(10)); - - var endpoint = Endpoint.bind(new Greeter()).bind(new Counter()).build(); - - for (int i = 0; i < eventLoopPoolSize; i++) { - vertx.deployVerticle( - new AbstractVerticle() { - @Override - public void start(Promise startPromise) { - RestateHttpServer.fromEndpoint(vertx, endpoint, httpServerOptions) - .listen(9080) - .map( - server -> { - LOG.info("Server started on port {}", server.actualPort()); - return (Void) null; - }) - .andThen(startPromise); - } - }); - } + RestateHttpServer.listen(Endpoint.bind(new Greeter()).bind(new Counter())); } } From fbb3436bfc8043d7c85890fb2d99f17d46124c0f Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Tue, 7 Jul 2026 14:41:36 +0200 Subject: [PATCH 3/3] Make sure we log state machine errors --- .../statemachine/ffm/FfmStateMachine.java | 21 ++++++++++++++++++- .../core/statemachine/ffm/NativeLogging.java | 4 ++-- 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java index 03b9df73..aca3a40c 100644 --- a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java +++ b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java @@ -30,6 +30,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import org.apache.logging.log4j.Logger; import org.jspecify.annotations.Nullable; /** @@ -51,6 +52,11 @@ */ public final class FfmStateMachine implements StateMachine { + // Same logger as the native tracing bridge, so SDK- and core-emitted lines interleave. + private static final Logger LOG = NativeLogging.LOG; + + private static final int SUSPENDED_ERROR_CODE = 599; + private final MemorySegment vmHandle; private final String responseContentType; private boolean freed = false; @@ -756,10 +762,23 @@ private void verifyNotFreed() { /** * Mark the state machine as closed and throw AbortedExecutionException. * - *

A subsequent notifyError will no-op. + *

A subsequent notifyError will no-op. Before aborting we surface the failure the core stashed + * (skipping the fetch when WARN is off, and skipping suspensions, which ride the same channel). */ private void abortAfterCoreClosed() { cachedState = InvocationState.CLOSED; + if (LOG.isWarnEnabled()) { + try (Arena arena = Arena.ofConfined()) { + MemorySegment err = VmError.allocate(arena); + SharedCoreNative.vm_take_last_vm_error(vmHandle, err); + int code = VmError.code(err); + // takeSliceString also frees the owned message Slice, so decode it even when not logging. + String message = FfmEncoding.takeSliceString(VmError.message(err)); + if (code != SUSPENDED_ERROR_CODE) { + LOG.warn("Invocation failed: {}", message); + } + } + } AbortedExecutionException.sneakyThrow(); } diff --git a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/NativeLogging.java b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/NativeLogging.java index 6b481d30..3b353000 100644 --- a/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/NativeLogging.java +++ b/sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/NativeLogging.java @@ -52,9 +52,9 @@ final class NativeLogging { private NativeLogging() {} /** Dedicated logger all native (shared-core) events are routed through. */ - static final String LOGGER_NAME = "dev.restate.sdk.core.StateMachine"; + private static final String LOGGER_NAME = "dev.restate.sdk.core.StateMachine"; - private static final Logger LOG = LogManager.getLogger(LOGGER_NAME); + static final Logger LOG = LogManager.getLogger(LOGGER_NAME); /** * Optional override for the native max level, bypassing the Log4j2-derived value. Accepts {@code