slop slop

Author: slinkydeveloper

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

@@ -68,7 +68,7 @@ class HandlerContextImpl implements HandlerContextInternal {
         new HandlerRequest(
             new InvocationIdImpl(input.invocationId(), input.randomSeed()),
             otelContext,
-            Slice.wrap(input.input()),
+            input.input(),
             input.headersAsMap(),
             serviceName,
             handlerName);
@@ -456,7 +456,11 @@ private void pollAsyncResultInner(AsyncResultInternal<?> asyncResult) {
       try {
         response = this.stateMachine.doAwait(future);
       } catch (Throwable e) {
-        this.failWithoutContextSwitch(e);
+        // doAwait sneaky-throws AbortedExecutionException on suspension. In this case, no need to
+        // fail twice.
+        if (!ExceptionUtils.containsAbortedExecutionException(e)) {
+          this.failWithoutContextSwitch(e);
+        }
         asyncResult.publicFuture().completeExceptionally(AbortedExecutionException.INSTANCE);
         return;
       }

sdk-core/src/main/java/dev/restate/sdk/core/statemachine/JavaStateMachine.java

@@ -17,7 +17,6 @@
 import dev.restate.common.Target;
 import dev.restate.sdk.common.RetryPolicy;
 import dev.restate.sdk.common.TerminalException;
-import dev.restate.sdk.core.ExceptionUtils;
 import dev.restate.sdk.core.ProtocolException;
 import dev.restate.sdk.core.generated.protocol.Protocol;
 import dev.restate.sdk.endpoint.HeadersAccessor;
@@ -183,7 +182,8 @@ public AwaitResult doAwait(UnresolvedFuture future) {
       }
       AwaitResult resolve = doProgress(List.of(tracked.handle));
       if (!(resolve instanceof AwaitResult.AnyCompleted)) {
-        // Can't resolve the invocation id yet (e.g. suspended); propagate.
+        // Can't resolve the invocation id yet (still waiting on external progress); propagate.
+        // (Suspension would have already thrown AbortedExecutionException out of doProgress.)
         return resolve;
       }
       NotificationValue value = takeNotification(tracked.handle);
@@ -209,21 +209,8 @@ public AwaitResult doAwait(UnresolvedFuture future) {
    * Single step of {@code do_progress}, translating the legacy {@link State.DoProgressResponse}.
    */
   private AwaitResult doProgress(List<Integer> anyHandle) {
-    State.DoProgressResponse response;
-    try {
-      response = this.stateContext.getCurrentState().doProgress(anyHandle, this.stateContext);
-    } catch (Throwable t) {
-      // The legacy state machine signalled both suspension (after writing the
-      // SuspensionMessage) and replay journal mismatches (after writing the ErrorMessage)
-      // by sneaky-throwing AbortedExecutionException out of doProgress. In both cases the
-      // state has already transitioned to ClosedState and the relevant message was written
-      // out, so we surface it as SUSPENDED: the driving handler context will observe the
-      // CLOSED state on the next loop iteration and abort the user code.
-      if (ExceptionUtils.containsAbortedExecutionException(t)) {
-        return AwaitResult.SUSPENDED;
-      }
-      throw t;
-    }
+    State.DoProgressResponse response =
+        this.stateContext.getCurrentState().doProgress(anyHandle, this.stateContext);
 
     if (response instanceof State.DoProgressResponse.AnyCompleted) {
       return AwaitResult.ANY_COMPLETED;

sdk-core/src/main/java/dev/restate/sdk/core/statemachine/ReplayingState.java

@@ -12,6 +12,7 @@
 
 import com.google.protobuf.ByteString;
 import com.google.protobuf.MessageLite;
+import dev.restate.common.Slice;
 import dev.restate.sdk.common.AbortedExecutionException;
 import dev.restate.sdk.core.ExceptionUtils;
 import dev.restate.sdk.core.ProtocolException;
@@ -195,7 +196,7 @@ public StateMachine.Input processInputCommand(StateContext stateContext) {
         startInfo.debugId(),
         startInfo.objectKey(),
         headers,
-        inputCommandMessage.getValue().getContent().toByteArray(),
+        Slice.wrap(inputCommandMessage.getValue().getContent().toByteArray()),
         Util.randomSeed(startInfo.debugId(), startInfo.randomSeed()),
         // scope/limitKey/idempotencyKey are V7-only; the legacy (V6) state machine doesn't carry
         // them.

sdk-core/src/main/java/dev/restate/sdk/core/statemachine/StateMachine.java

@@ -66,6 +66,14 @@ public interface StateMachine extends AutoCloseable {
 
   // --- Async results
 
+  /**
+   * Make progress on the given await tree.
+   *
+   * <p>On suspension this does not return a value: it sneaky-throws {@code
+   * AbortedExecutionException} (the VM is now closed and its output already pumped). The driving
+   * handler context treats that sentinel as a clean abort of the user code, not a failure to
+   * re-report.
+   */
   AwaitResult doAwait(UnresolvedFuture future);
 
   @Nullable NotificationValue takeNotification(int handle);
@@ -76,7 +84,7 @@ record Input(
       String invocationId,
       String key,
       List<String[]> headers,
-      byte[] input,
+      Slice input,
       long randomSeed,
       // V7 optionals, propagated from the native core; null when absent. Surfaced to the public API
       // separately at a later point.
@@ -181,12 +189,15 @@ void proposeRunCompletion(
   // Value types
   // =========================================================================
 
-  /** Outcome of {@link #doAwait(UnresolvedFuture)}. */
+  /**
+   * Outcome of {@link #doAwait(UnresolvedFuture)}. Suspension is not represented here: on
+   * suspension {@code doAwait} sneaky-throws {@code AbortedExecutionException} instead of
+   * returning.
+   */
   sealed interface AwaitResult {
     AwaitResult ANY_COMPLETED = new AnyCompleted();
     AwaitResult WAIT_EXTERNAL_PROGRESS = new WaitExternalProgress();
     AwaitResult CANCEL_SIGNAL_RECEIVED = new CancelSignalReceived();
-    AwaitResult SUSPENDED = new Suspended();
 
     record AnyCompleted() implements AwaitResult {}
 
@@ -195,8 +206,6 @@ record WaitExternalProgress() implements AwaitResult {}
     record ExecuteRun(int handle) implements AwaitResult {}
 
     record CancelSignalReceived() implements AwaitResult {}
-
-    record Suspended() implements AwaitResult {}
   }
 
   /**

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

@@ -12,9 +12,10 @@
 import dev.restate.sdk.common.RetryPolicy;
 import dev.restate.sdk.common.TerminalException;
 import dev.restate.sdk.core.statemachine.StateMachine;
+import dev.restate.sdk.core.statemachine.ffm.generated.CallArguments;
+import dev.restate.sdk.core.statemachine.ffm.generated.ForeignSlice;
 import dev.restate.sdk.core.statemachine.ffm.generated.SharedCoreNative;
 import dev.restate.sdk.core.statemachine.ffm.generated.Slice;
-import dev.restate.sdk.core.statemachine.ffm.generated.TargetAbi;
 import java.io.ByteArrayOutputStream;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
@@ -57,39 +58,111 @@ static MemorySegment allocateBytes(SegmentAllocator alloc, byte @Nullable [] byt
   }
 
   /**
-   * Allocate a native segment holding the bytes of {@code slice}, copied straight from the slice's
-   * buffer (no intermediate heap {@code byte[]}), or {@link MemorySegment#NULL} when empty.
+   * Allocate a native segment holding the UTF-8 bytes of {@code s}, or {@link MemorySegment#NULL}
+   * when {@code s} is null. An empty (non-null) string yields a zero-length, non-null segment.
    */
-  static MemorySegment allocateSlice(SegmentAllocator alloc, dev.restate.common.Slice slice) {
+  static MemorySegment allocateUtf8(SegmentAllocator alloc, @Nullable String s) {
+    if (s == null) {
+      return MemorySegment.NULL;
+    }
+    return allocateBytes(alloc, s.getBytes(StandardCharsets.UTF_8));
+  }
+
+  /**
+   * Allocate a Rust-owned native buffer (via {@code alloc_buffer}) and copy {@code slice} into it,
+   * transferring ownership to the native callee. The returned segment is tied to no {@link Arena},
+   * so Java never frees it: the native side takes ownership on the call and frees it when the core
+   * is done with it (see {@code MEMORY_MODEL.md}). Used for opaque payloads only. Returns {@link
+   * MemorySegment#NULL} for an empty slice.
+   */
+  static MemorySegment transferSlice(dev.restate.common.Slice slice) {
     int len = slice.readableBytes();
     if (len == 0) {
       return MemorySegment.NULL;
     }
-    MemorySegment seg = alloc.allocate(len);
-    slice.copyTo(seg.asByteBuffer());
-    return seg;
+    MemorySegment buf = SharedCoreNative.alloc_buffer(len).reinterpret(len);
+    slice.copyTo(buf.asByteBuffer());
+    return buf;
   }
 
   /**
-   * Allocate a native segment holding the UTF-8 bytes of {@code s}, or {@link MemorySegment#NULL}
-   * when {@code s} is null. An empty (non-null) string yields a zero-length, non-null segment.
+   * Like {@link #transferSlice} but for the UTF-8 bytes of {@code s}: allocates a Rust-owned buffer
+   * via {@code alloc_buffer}, copies the encoded bytes in, and transfers ownership to the native
+   * callee (which re-owns it as a {@code String} zero-copy, see {@code take_string}). Used for the
+   * {@code notify_error} message/stacktrace. Returns {@link MemorySegment#NULL} for a null/empty
+   * string (the Rust side maps that to the empty string / absent stacktrace).
    */
-  static MemorySegment allocateUtf8(SegmentAllocator alloc, @Nullable String s) {
+  static MemorySegment transferUtf8(@Nullable String s) {
     if (s == null) {
       return MemorySegment.NULL;
     }
-    return allocateBytes(alloc, s.getBytes(StandardCharsets.UTF_8));
-  }
-
-  static long len(MemorySegment seg) {
-    return seg.byteSize();
+    byte[] bytes = s.getBytes(StandardCharsets.UTF_8);
+    if (bytes.length == 0) {
+      return MemorySegment.NULL;
+    }
+    MemorySegment buf = SharedCoreNative.alloc_buffer(bytes.length).reinterpret(bytes.length);
+    MemorySegment.copy(bytes, 0, buf, ValueLayout.JAVA_BYTE, 0, bytes.length);
+    return buf;
   }
 
   /** Allocates a result {@link Slice} struct out-param (for calls that write a bare Slice). */
   static MemorySegment allocateSliceStruct(SegmentAllocator alloc) {
     return Slice.allocate(alloc);
   }
 
+  // -------------------------------------------------------------------------
+  // By-value boundary slices. A ForeignSlice borrows Java-arena memory (Rust reads it in-call and
+  // never frees it); a Slice is alloc_buffer-backed memory whose ownership transfers into the
+  // native callee. Both cross by value as a {ptr,len} struct; we keep the construction here so the
+  // state machine never touches the (name-colliding) generated Slice type directly.
+  // -------------------------------------------------------------------------
+
+  /** Writes {@code buf}'s ptr+len into an already-allocated ForeignSlice field segment. */
+  private static void setForeign(MemorySegment foreignSliceField, MemorySegment buf) {
+    ForeignSlice.ptr(foreignSliceField, buf);
+    ForeignSlice.len(foreignSliceField, buf.byteSize());
+  }
+
+  /** Writes {@code buf}'s ptr+len into an already-allocated (ownership-transfer) Slice field. */
+  private static void setOwned(MemorySegment sliceField, MemorySegment buf) {
+    Slice.ptr(sliceField, buf);
+    Slice.len(sliceField, buf.byteSize());
+  }
+
+  /** A by-value ForeignSlice borrowing a fresh UTF-8 copy of {@code s} (null/empty → null ptr). */
+  static MemorySegment foreignUtf8(SegmentAllocator alloc, @Nullable String s) {
+    MemorySegment fs = ForeignSlice.allocate(alloc);
+    setForeign(fs, allocateUtf8(alloc, s));
+    return fs;
+  }
+
+  /** A by-value ForeignSlice borrowing a fresh copy of {@code bytes} (null/empty → null ptr). */
+  static MemorySegment foreignBytes(SegmentAllocator alloc, byte @Nullable [] bytes) {
+    MemorySegment fs = ForeignSlice.allocate(alloc);
+    setForeign(fs, allocateBytes(alloc, bytes));
+    return fs;
+  }
+
+  /**
+   * A by-value Slice transferring {@code slice}'s bytes (alloc_buffer buffer the callee re-owns).
+   */
+  static MemorySegment transferToSlice(SegmentAllocator alloc, dev.restate.common.Slice slice) {
+    MemorySegment buf = transferSlice(slice);
+    MemorySegment s = Slice.allocate(alloc);
+    Slice.ptr(s, buf);
+    Slice.len(s, buf.byteSize());
+    return s;
+  }
+
+  /** A by-value Slice transferring the UTF-8 bytes of {@code s} (callee re-owns it as a String). */
+  static MemorySegment transferUtf8ToSlice(SegmentAllocator alloc, @Nullable String s) {
+    MemorySegment buf = transferUtf8(s);
+    MemorySegment slice = Slice.allocate(alloc);
+    Slice.ptr(slice, buf);
+    Slice.len(slice, buf.byteSize());
+    return slice;
+  }
+
   // -------------------------------------------------------------------------
   // Result Slice reading (outputs: owned by Rust, must be freed)
   // -------------------------------------------------------------------------
@@ -121,12 +194,6 @@ static String takeSliceString(MemorySegment sliceStruct) {
     return new String(bytes, StandardCharsets.UTF_8);
   }
 
-  /** Like {@link #takeSliceString} but returns {@code null} for the empty/absent slice. */
-  static @Nullable String takeSliceStringOrNull(MemorySegment sliceStruct) {
-    byte[] bytes = takeSliceBytes(sliceStruct);
-    return bytes.length == 0 ? null : new String(bytes, StandardCharsets.UTF_8);
-  }
-
   /**
    * Wraps an owned result {@link Slice} (Rust-allocated; must be released via {@code free_buffer})
    * in a {@link MemorySegmentSlice} read zero-copy — no copy-out into a heap {@code byte[]}. The
@@ -157,44 +224,26 @@ static dev.restate.common.Slice wrapOwnedSlice(MemorySegment sliceStruct) {
    * (ptr,len)}; a null ptr means the optional field is absent. {@code headers} is the encoded
    * header-list blob.
    */
-  static MemorySegment buildTarget(
+  static MemorySegment buildCallArguments(
       SegmentAllocator alloc,
       Target target,
+      dev.restate.common.Slice payload,
       @Nullable String idempotencyKey,
       @Nullable String scope,
       @Nullable String limitKey,
       @Nullable Collection<Map.Entry<String, String>> headers) {
-    MemorySegment t = TargetAbi.allocate(alloc);
-
-    MemorySegment service = allocateUtf8(alloc, target.getService());
-    TargetAbi.service_ptr(t, service);
-    TargetAbi.service_len(t, len(service));
-
-    MemorySegment handler = allocateUtf8(alloc, target.getHandler());
-    TargetAbi.handler_ptr(t, handler);
-    TargetAbi.handler_len(t, len(handler));
-
-    MemorySegment key = allocateUtf8(alloc, target.getKey());
-    TargetAbi.key_ptr(t, key);
-    TargetAbi.key_len(t, len(key));
-
-    MemorySegment idem = allocateUtf8(alloc, idempotencyKey);
-    TargetAbi.idempotency_key_ptr(t, idem);
-    TargetAbi.idempotency_key_len(t, len(idem));
-
-    // V7 optionals: a null/empty segment maps to `None` on the Rust side.
-    MemorySegment scopeSeg = allocateUtf8(alloc, scope);
-    TargetAbi.scope_ptr(t, scopeSeg);
-    TargetAbi.scope_len(t, len(scopeSeg));
-
-    MemorySegment limitKeySeg = allocateUtf8(alloc, limitKey);
-    TargetAbi.limit_key_ptr(t, limitKeySeg);
-    TargetAbi.limit_key_len(t, len(limitKeySeg));
-
-    MemorySegment headersBlob = allocateBytes(alloc, encodeHeaderList(headers));
-    TargetAbi.headers_ptr(t, headersBlob);
-    TargetAbi.headers_len(t, len(headersBlob));
-
+    MemorySegment t = CallArguments.allocate(alloc);
+    // Target fields are nested ForeignSlices borrowing a fresh copy in `alloc`; a null/empty buffer
+    // maps to `None` (V7 optionals) / the empty slice on the Rust side.
+    setForeign(CallArguments.service(t), allocateUtf8(alloc, target.getService()));
+    setForeign(CallArguments.handler(t), allocateUtf8(alloc, target.getHandler()));
+    setForeign(CallArguments.key(t), allocateUtf8(alloc, target.getKey()));
+    setForeign(CallArguments.idempotency_key(t), allocateUtf8(alloc, idempotencyKey));
+    setForeign(CallArguments.scope(t), allocateUtf8(alloc, scope));
+    setForeign(CallArguments.limit_key(t), allocateUtf8(alloc, limitKey));
+    setForeign(CallArguments.headers(t), allocateBytes(alloc, encodeHeaderList(headers)));
+    // input is a Slice: ownership of the alloc_buffer-backed payload transfers into the core.
+    setOwned(CallArguments.input(t), transferSlice(payload));
     return t;
   }