Make take output zero copy

Author: slinkydeveloper

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

@@ -138,15 +138,9 @@ public Slice takeOutput() {
       return Slice.EMPTY;
     }
     try (Arena arena = Arena.ofConfined()) {
-      // take_output never fails: it writes a bare Slice (not a BufferResult). Copy the buffer out
-      // into a heap byte[] and free the native buffer immediately (takeSliceBytes). This is the
-      // safe
-      // default while we chase a GC-tied-lifetime flake in the bidi path: the zero-copy
-      // wrapOwnedSlice variant handed the buffer to the async output stream and could be collected
-      // before the stream consumed it, intermittently dropping a flushed chunk.
       MemorySegment out = FfmEncoding.allocateSliceStruct(arena);
       SharedCoreNative.vm_take_output(vmHandle, out);
-      return Slice.wrap(FfmEncoding.takeSliceBytes(out));
+      return FfmEncoding.wrapOwnedSlice(out);
     }
   }
 

sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpResponseFlowAdapter.java

@@ -13,6 +13,7 @@
 import io.netty.buffer.Unpooled;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpServerResponse;
+import java.lang.ref.Reference;
 import java.util.concurrent.Flow;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -44,9 +45,16 @@ public void onNext(Slice slice) {
       return;
     }
 
-    // If HTTP HEADERS frame have not been sent, Vert.x will send them
-    this.httpServerResponse.write(
-        Buffer.buffer(Unpooled.wrappedBuffer(slice.asReadOnlyByteBuffer())));
+    // If HTTP HEADERS frame have not been sent, Vert.x will send them.
+    //
+    // The slice may be backed by native memory with a GC-tied free (FFM wrapOwnedSlice), and this
+    // write is zero-copy + asynchronous — Vert.x flushes the wrapped buffer later. Keep the slice
+    // reachable until the write completes so the cleanup can't free the native buffer mid-flush:
+    // the completion callback captures `slice` (reachabilityFence is an un-elidable use), and the
+    // future holds that callback until the flush is done.
+    this.httpServerResponse
+        .write(Buffer.buffer(Unpooled.wrappedBuffer(slice.asReadOnlyByteBuffer())))
+        .onComplete(ignored -> Reference.reachabilityFence(slice));
   }
 
   @Override