Skip to content

Commit 370b3a3

Browse files
State machine followups (#631)
* Avoid loading the last vm error when the state machine has been closed anyway. * Revert example changes * Make sure we log state machine errors
1 parent 152b712 commit 370b3a3

4 files changed

Lines changed: 54 additions & 74 deletions

File tree

examples/src/main/java/my/restate/sdk/examples/Greeter.java

Lines changed: 1 addition & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -13,12 +13,6 @@
1313
import dev.restate.sdk.annotation.Service;
1414
import dev.restate.sdk.endpoint.Endpoint;
1515
import dev.restate.sdk.http.vertx.RestateHttpServer;
16-
import io.vertx.core.AbstractVerticle;
17-
import io.vertx.core.Promise;
18-
import io.vertx.core.Vertx;
19-
import io.vertx.core.VertxOptions;
20-
import io.vertx.core.http.Http2Settings;
21-
import io.vertx.core.http.HttpServerOptions;
2216
import java.time.Duration;
2317
import org.apache.logging.log4j.LogManager;
2418
import org.apache.logging.log4j.Logger;
@@ -46,29 +40,6 @@ public GreetingResponse greet(Greeting req) {
4640
}
4741

4842
public static void main(String[] args) {
49-
var vertxOptions = new VertxOptions();
50-
var eventLoopPoolSize = vertxOptions.getEventLoopPoolSize();
51-
var vertx = Vertx.vertx(new VertxOptions());
52-
var httpServerOptions =
53-
new HttpServerOptions().setInitialSettings(new Http2Settings().setMaxConcurrentStreams(10));
54-
55-
var endpoint = Endpoint.bind(new Greeter()).bind(new Counter()).build();
56-
57-
for (int i = 0; i < eventLoopPoolSize; i++) {
58-
vertx.deployVerticle(
59-
new AbstractVerticle() {
60-
@Override
61-
public void start(Promise<Void> startPromise) {
62-
RestateHttpServer.fromEndpoint(vertx, endpoint, httpServerOptions)
63-
.listen(9080)
64-
.map(
65-
server -> {
66-
LOG.info("Server started on port {}", server.actualPort());
67-
return (Void) null;
68-
})
69-
.andThen(startPromise);
70-
}
71-
});
72-
}
43+
RestateHttpServer.listen(Endpoint.bind(new Greeter()).bind(new Counter()));
7344
}
7445
}

sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java

Lines changed: 5 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -417,7 +417,7 @@ public void pollAsyncResult(AsyncResultInternal<?> asyncResult) {
417417
try {
418418
this.pumpOutput();
419419
this.pollAsyncResultInner(asyncResult);
420-
} catch (Exception e) {
420+
} catch (Throwable e) {
421421
this.failWithoutContextSwitch(e);
422422
}
423423
}
@@ -428,10 +428,7 @@ private void pollAsyncResultInner(AsyncResultInternal<?> asyncResult) {
428428
try {
429429
asyncResult.tryComplete(this::takeNotification);
430430
} catch (Throwable e) {
431-
// This can happen if the state machine was closed in the meantime.
432-
if (!ExceptionUtils.containsAbortedExecutionException(e)) {
433-
this.failWithoutContextSwitch(e);
434-
}
431+
this.failWithoutContextSwitch(e);
435432
asyncResult.publicFuture().completeExceptionally(AbortedExecutionException.INSTANCE);
436433
return;
437434
}
@@ -447,11 +444,7 @@ private void pollAsyncResultInner(AsyncResultInternal<?> asyncResult) {
447444
try {
448445
response = this.stateMachine.doAwait(asyncResult);
449446
} catch (Throwable e) {
450-
// doAwait sneaky-throws AbortedExecutionException on suspension. In this case, no need to
451-
// fail twice.
452-
if (!ExceptionUtils.containsAbortedExecutionException(e)) {
453-
this.failWithoutContextSwitch(e);
454-
}
447+
this.failWithoutContextSwitch(e);
455448
asyncResult.publicFuture().completeExceptionally(AbortedExecutionException.INSTANCE);
456449
return;
457450
}
@@ -480,7 +473,7 @@ Optional<StateMachine.NotificationValue> takeNotification(int handle) {
480473
public void proposeRunSuccess(int runHandle, Slice toWrite) {
481474
try {
482475
this.stateMachine.proposeRunCompletion(runHandle, toWrite);
483-
} catch (Exception e) {
476+
} catch (Throwable e) {
484477
this.failWithoutContextSwitch(e);
485478
}
486479
this.pumpOutput();
@@ -499,7 +492,7 @@ public void proposeRunFailure(
499492
} else {
500493
this.stateMachine.proposeRunCompletion(runHandle, throwable, attemptDuration, retryPolicy);
501494
}
502-
} catch (Exception e) {
495+
} catch (Throwable e) {
503496
this.failWithoutContextSwitch(e);
504497
}
505498
this.pumpOutput();

sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java

Lines changed: 46 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import java.util.LinkedHashMap;
3131
import java.util.List;
3232
import java.util.Map;
33+
import org.apache.logging.log4j.Logger;
3334
import org.jspecify.annotations.Nullable;
3435

3536
/**
@@ -40,16 +41,22 @@
4041
* <p>Each call is a direct FFM downcall. Results come back either register-packed into the scalar
4142
* return value (the invocation state, plus a handle for handle-returning calls) or, when too wide
4243
* to pack, written into a caller-provided out-param struct whose first field is the state. Owned
43-
* {@link Slice}s are copied out and freed by the caller. On error the state field carries the
44-
* {@link SharedCoreNative#ERROR_STATE()} sentinel; the detail is then fetched via {@code
45-
* vm_take_last_vm_error} and thrown as a {@link ProtocolException}.
44+
* {@link Slice}s are copied out and freed by the caller. On the {@link
45+
* SharedCoreNative#ERROR_STATE()} sentinel the core has already written its terminal message and
46+
* closed, so we sneaky-throw {@link AbortedExecutionException} ({@link #abortAfterCoreClosed()});
47+
* construction and {@code isReadyToExecute} errors surface as a {@link ProtocolException}.
4648
*
4749
* <p>An instance is driven by a single thread at a time (no reentrancy, per the contract); only
4850
* {@link #state()} — a volatile read — is safe from any thread. Each downcall allocates its
4951
* out-param and inputs in a confined {@link Arena} for the duration of the call.
5052
*/
5153
public final class FfmStateMachine implements StateMachine {
5254

55+
// Same logger as the native tracing bridge, so SDK- and core-emitted lines interleave.
56+
private static final Logger LOG = NativeLogging.LOG;
57+
58+
private static final int SUSPENDED_ERROR_CODE = 599;
59+
5360
private final MemorySegment vmHandle;
5461
private final String responseContentType;
5562
private boolean freed = false;
@@ -186,12 +193,9 @@ public AwaitResult doAwait(UnresolvedFuture future) {
186193
packed = SharedCoreNative.vm_do_await(vmHandle, futureSeg);
187194
}
188195

189-
// Register-packed result (a bare u64, no out-param struct): the low byte is the variant tag,
190-
// the
191-
// high 32 bits carry the ExecuteRun run handle. On error the core stashed the detail and we
192-
// fetch
193-
// it via vm_take_last_vm_error. Suspension isn't a variant here — the driver surfaces it by
194-
// sneaky-throwing AbortedExecutionException per the StateMachine contract.
196+
// Register-packed u64: low byte = variant tag, high 32 bits = ExecuteRun handle. The ERROR
197+
// variant also covers suspension (the core folds it in); either way the core already wrote its
198+
// terminal message and closed, so we abort cleanly rather than fetch the discarded detail.
195199
int variant = (int) packed;
196200
if (variant == SharedCoreNative.AWAIT_VARIANT_ANY_COMPLETED()) {
197201
return AwaitResult.ANY_COMPLETED;
@@ -202,22 +206,11 @@ public AwaitResult doAwait(UnresolvedFuture future) {
202206
} else if (variant == SharedCoreNative.AWAIT_VARIANT_CANCEL_SIGNAL_RECEIVED()) {
203207
return AwaitResult.CANCEL_SIGNAL_RECEIVED;
204208
} else if (variant == SharedCoreNative.AWAIT_VARIANT_ERROR()) {
205-
throw takeLastVmError();
209+
abortAfterCoreClosed();
206210
}
207211
throw new IllegalStateException("Unknown do_await variant: " + variant);
208212
}
209213

210-
/**
211-
* Fetches + maps the error the core stashed on the last failing packed-result call (cold path).
212-
*/
213-
private ProtocolException takeLastVmError() {
214-
try (Arena arena = Arena.ofConfined()) {
215-
MemorySegment err = VmError.allocate(arena);
216-
SharedCoreNative.vm_take_last_vm_error(vmHandle, err);
217-
return closeVmAndMapError(err);
218-
}
219-
}
220-
221214
@Override
222215
public @Nullable NotificationValue takeNotification(int handle) {
223216
verifyNotFreed();
@@ -293,7 +286,7 @@ public Input input() {
293286
SharedCoreNative.vm_sys_input(vmHandle, out);
294287
int state = InputResult.state(out);
295288
if (state == SharedCoreNative.ERROR_STATE()) {
296-
throw takeLastVmError();
289+
abortAfterCoreClosed();
297290
}
298291
updateState(state);
299292

@@ -429,7 +422,7 @@ public CallHandle call(
429422
SharedCoreNative.vm_sys_call(vmHandle, args, out);
430423
int state = CallResult.state(out);
431424
if (state == SharedCoreNative.ERROR_STATE()) {
432-
throw takeLastVmError();
425+
abortAfterCoreClosed();
433426
}
434427
updateState(state);
435428
return new CallHandle(CallResult.invocation_id_handle(out), CallResult.result_handle(out));
@@ -469,7 +462,7 @@ public Awakeable awakeable() {
469462
SharedCoreNative.vm_sys_awakeable(vmHandle, out);
470463
int state = AwakeableResult.state(out);
471464
if (state == SharedCoreNative.ERROR_STATE()) {
472-
throw takeLastVmError();
465+
abortAfterCoreClosed();
473466
}
474467
updateState(state);
475468
return new Awakeable(
@@ -592,7 +585,7 @@ public RunResultHandle run(String name) {
592585
SharedCoreNative.vm_sys_run(vmHandle, FfmEncoding.foreignUtf8(arena, name), out);
593586
int state = RunResult.state(out);
594587
if (state == SharedCoreNative.ERROR_STATE()) {
595-
throw takeLastVmError();
588+
abortAfterCoreClosed();
596589
}
597590
updateState(state);
598591
return new RunResultHandle(RunResult.replayed(out) != 0, RunResult.handle(out));
@@ -715,25 +708,25 @@ public void end() {
715708
/**
716709
* Decodes a register-packed handle result ({@code u64}): low 32 bits = state (or the {@link
717710
* SharedCoreNative#ERROR_STATE()} sentinel), high 32 bits = handle. On the error sentinel we
718-
* fetch + throw the stashed error; otherwise update the cached state and return the handle.
711+
* {@link #abortAfterCoreClosed()}; otherwise update the cached state and return the handle.
719712
*/
720713
private int parseHandleResult(long packed) {
721714
int state = (int) packed;
722715
if (state == SharedCoreNative.ERROR_STATE()) {
723-
throw takeLastVmError();
716+
abortAfterCoreClosed();
724717
}
725718
updateState(state);
726719
return (int) (packed >>> 32);
727720
}
728721

729722
/**
730723
* Applies a register-packed empty result (a bare {@code u32}): the new state, or the {@link
731-
* SharedCoreNative#ERROR_STATE()} sentinel. On the error sentinel we fetch + throw the stashed
732-
* error; otherwise update the cached state.
724+
* SharedCoreNative#ERROR_STATE()} sentinel. On the error sentinel we {@link
725+
* #abortAfterCoreClosed()}; otherwise update the cached state.
733726
*/
734727
private void consumeEmptyResult(int packed) {
735728
if (packed == SharedCoreNative.ERROR_STATE()) {
736-
throw takeLastVmError();
729+
abortAfterCoreClosed();
737730
}
738731
updateState(packed);
739732
}
@@ -766,6 +759,29 @@ private void verifyNotFreed() {
766759
}
767760
}
768761

762+
/**
763+
* Mark the state machine as closed and throw AbortedExecutionException.
764+
*
765+
* <p>A subsequent notifyError will no-op. Before aborting we surface the failure the core stashed
766+
* (skipping the fetch when WARN is off, and skipping suspensions, which ride the same channel).
767+
*/
768+
private void abortAfterCoreClosed() {
769+
cachedState = InvocationState.CLOSED;
770+
if (LOG.isWarnEnabled()) {
771+
try (Arena arena = Arena.ofConfined()) {
772+
MemorySegment err = VmError.allocate(arena);
773+
SharedCoreNative.vm_take_last_vm_error(vmHandle, err);
774+
int code = VmError.code(err);
775+
// takeSliceString also frees the owned message Slice, so decode it even when not logging.
776+
String message = FfmEncoding.takeSliceString(VmError.message(err));
777+
if (code != SUSPENDED_ERROR_CODE) {
778+
LOG.warn("Invocation failed: {}", message);
779+
}
780+
}
781+
}
782+
AbortedExecutionException.sneakyThrow();
783+
}
784+
769785
private static String formatThrowableMessage(Throwable throwable) {
770786
return throwable.toString();
771787
}

sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/NativeLogging.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,9 +52,9 @@ final class NativeLogging {
5252
private NativeLogging() {}
5353

5454
/** Dedicated logger all native (shared-core) events are routed through. */
55-
static final String LOGGER_NAME = "dev.restate.sdk.core.StateMachine";
55+
private static final String LOGGER_NAME = "dev.restate.sdk.core.StateMachine";
5656

57-
private static final Logger LOG = LogManager.getLogger(LOGGER_NAME);
57+
static final Logger LOG = LogManager.getLogger(LOGGER_NAME);
5858

5959
/**
6060
* Optional override for the native max level, bypassing the Log4j2-derived value. Accepts {@code

0 commit comments

Comments
 (0)