State machine followups

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<Void> 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()));
   }
 }

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<StateMachine.NotificationValue> 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();

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;
 
 /**
@@ -40,16 +41,22 @@
  * <p>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}.
  *
  * <p>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
  * out-param and inputs in a confined {@link Arena} for the duration of the call.
  */
 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;
@@ -186,12 +193,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 +206,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 +286,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 +422,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 +462,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 +585,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,25 +708,25 @@ 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);
   }
 
   /**
    * 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 +759,29 @@ private void verifyNotFreed() {
     }
   }
 
+  /**
+   * Mark the state machine as closed and throw AbortedExecutionException.
+   *
+   * <p>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();
+  }
+
   private static String formatThrowableMessage(Throwable throwable) {
     return throwable.toString();
   }

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