[WIP] Shared core + Host memory extension

.github/workflows/dependency-submission.yml

@@ -18,5 +18,9 @@ jobs:
         with:
           distribution: 'temurin'
           java-version: 17
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
       - name: Generate and submit dependency graph
         uses: gradle/actions/dependency-submission@v4

.github/workflows/docker.yaml

@@ -32,6 +32,11 @@ jobs:
       - name: Setup Gradle
         uses: gradle/actions/setup-gradle@v4
 
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
+
       - name: Log into GitHub container registry
         uses: docker/login-action@v2
         with:

.github/workflows/integration.yaml

@@ -123,6 +123,12 @@ jobs:
         if: ${{ inputs.serviceImage == '' }}
         uses: gradle/actions/setup-gradle@v4
 
+      - name: Install Rust toolchain
+        if: ${{ inputs.serviceImage == '' }}
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
+
       - name: Build restatedev/test-services-java image
         if: ${{ inputs.serviceImage == '' }}
         run: ./gradlew -Djib.console=plain :test-services:jibDockerBuild
@@ -135,7 +141,7 @@ jobs:
 
       - name: Run test tool
         continue-on-error: ${{ inputs.continueOnError == 'true' }}
-        uses: restatedev/e2e/sdk-tests@v1.0
+        uses: restatedev/e2e/sdk-tests@v2.1
         with:
           envVars: ${{ inputs.envVars }}
           testArtifactOutput: ${{ inputs.testArtifactOutput != '' && inputs.testArtifactOutput || 'sdk-java-integration-test-report' }}

.github/workflows/release-docs.yml

@@ -31,6 +31,10 @@ jobs:
           java-version: '21'
       - name: Setup Gradle
         uses: gradle/actions/setup-gradle@v4
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
 
       - name: Build Javadocs
         run: gradle :sdk-aggregated-javadocs:javadoc

.github/workflows/release.yml

@@ -20,6 +20,11 @@ jobs:
       - name: Setup Gradle
         uses: gradle/actions/setup-gradle@v4
 
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
+
       # Retrieve the version of the SDK
       - name: Install dasel
         run: curl -sSLf "$(curl -sSLf https://api.github.com/repos/tomwright/dasel/releases/latest | grep browser_download_url | grep linux_amd64 | grep -v .gz | cut -d\" -f 4)" -L -o dasel && chmod +x dasel && mv ./dasel /usr/local/bin/dasel

.github/workflows/tests.yml

@@ -27,6 +27,11 @@ jobs:
       - name: Setup Gradle
         uses: gradle/actions/setup-gradle@v4
 
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
+
       - name: Pull Restate docker image
         run: docker pull ghcr.io/restatedev/restate:main
 
@@ -54,6 +59,10 @@ jobs:
           java-version: '21'
       - name: Setup Gradle
         uses: gradle/actions/setup-gradle@v4
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          target: wasm32-unknown-unknown
 
       - name: Build Javadocs
         run: gradle :sdk-aggregated-javadocs:javadoc

.gitignore

@@ -37,4 +37,5 @@ build
 kls_database.db
 .kotlin
 
-.restate
+.restate
+/sdk-core/src/main/rust/target/

common/src/main/java/dev/restate/common/Slice.java

@@ -28,6 +28,16 @@ public interface Slice {
 
   byte[] toByteArray();
 
+  /**
+   * Returns a Slice over the sub-range {@code [offset, offset + length)} of this Slice. Shares
+   * storage with this Slice — no copy.
+   */
+  default Slice slice(int offset, int length) {
+    ByteBuffer view = asReadOnlyByteBuffer();
+    view.position(offset).limit(offset + length);
+    return Slice.wrap(view.slice());
+  }
+
   /** Wrap a {@link ByteBuffer}. This will not copy the buffer. */
   static Slice wrap(ByteBuffer byteBuffer) {
     return new Slice() {
@@ -48,7 +58,8 @@ public void copyTo(byte[] target) {
 
       @Override
       public void copyTo(byte[] target, int targetOffset) {
-        byteBuffer.slice().get(target, targetOffset, target.length);
+        ByteBuffer view = byteBuffer.slice();
+        view.get(target, targetOffset, view.remaining());
       }
 
       @Override
@@ -61,9 +72,19 @@ public void copyTo(ByteBuffer buffer) {
         buffer.put(byteBuffer.slice());
       }
 
+      @Override
+      public Slice slice(int offset, int length) {
+        return wrap(byteBuffer.slice(offset, length));
+      }
+
       @Override
       public byte[] toByteArray() {
-        if (byteBuffer.hasArray()) {
+        // Only the no-offset / no-limit case can return the backing array
+        // directly — otherwise array() would expose bytes outside the view.
+        if (byteBuffer.hasArray()
+            && byteBuffer.arrayOffset() == 0
+            && byteBuffer.position() == 0
+            && byteBuffer.remaining() == byteBuffer.array().length) {
           return byteBuffer.array();
         }
 

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

@@ -0,0 +1,85 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package my.restate.sdk.examples;
+
+import dev.restate.sdk.DurableFuture;
+import dev.restate.sdk.Restate;
+import dev.restate.sdk.annotation.Handler;
+import dev.restate.sdk.annotation.Service;
+import dev.restate.sdk.endpoint.Endpoint;
+import dev.restate.sdk.http.vertx.RestateHttpServer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * Spawns N concurrent {@code ctx.run} steps, each producing a large random payload (100KB–2MB),
+ * with ~1-in-4 retryable failures sprinkled in. Returns the concatenation of all the payloads.
+ *
+ * <p>Useful to exercise cooperative suspension and AwaitingOnMessage with a non-trivial {@code
+ * AllSucceededOrFirstFailed} combinator that the runtime can observe while runs are in flight.
+ */
+@Service
+public class ConcurrentRuns {
+
+  private static final Logger LOG = LogManager.getLogger(ConcurrentRuns.class);
+
+  private static final int NUM_RUNS = 6;
+  private static final int MIN_PAYLOAD_BYTES = 100 * 1024;
+  private static final int MAX_PAYLOAD_BYTES = 2 * 1024 * 1024;
+  private static final int FAILURE_DENOMINATOR = 4;
+
+  private static final String ALPHABET =
+      "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
+
+  @Handler
+  public String run() {
+    List<DurableFuture<String>> futures = new ArrayList<>(NUM_RUNS);
+    for (int i = 0; i < NUM_RUNS; i++) {
+      final int idx = i;
+      futures.add(
+          Restate.runAsync(
+              "payload-" + idx,
+              String.class,
+              () -> {
+                if (ThreadLocalRandom.current().nextInt(FAILURE_DENOMINATOR) == 0) {
+                  LOG.info("Run {} simulating retryable failure", idx);
+                  throw new RuntimeException("simulated retryable failure on run " + idx);
+                }
+                int size =
+                    ThreadLocalRandom.current().nextInt(MIN_PAYLOAD_BYTES, MAX_PAYLOAD_BYTES + 1);
+                LOG.info("Run {} generating {} bytes", idx, size);
+                return randomString(size);
+              }));
+    }
+
+    DurableFuture.all((List) futures).await();
+
+    StringBuilder sb = new StringBuilder();
+    for (DurableFuture<String> f : futures) {
+      sb.append(f.await());
+    }
+    return sb.toString();
+  }
+
+  private static String randomString(int size) {
+    ThreadLocalRandom rnd = ThreadLocalRandom.current();
+    char[] buf = new char[size];
+    for (int i = 0; i < size; i++) {
+      buf[i] = ALPHABET.charAt(rnd.nextInt(ALPHABET.length()));
+    }
+    return new String(buf);
+  }
+
+  public static void main(String[] args) {
+    RestateHttpServer.listen(Endpoint.bind(new ConcurrentRuns()).build());
+  }
+}

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

@@ -8,25 +8,67 @@
 // https://github.com/restatedev/sdk-java/blob/main/LICENSE
 package my.restate.sdk.examples;
 
+import dev.restate.sdk.Restate;
 import dev.restate.sdk.annotation.Handler;
 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;
 
 @Service
 public class Greeter {
 
+  private static final Logger LOG = LogManager.getLogger(Greeter.class);
+
   public record Greeting(String name) {}
 
   public record GreetingResponse(String message) {}
 
   @Handler
   public GreetingResponse greet(Greeting req) {
+    Restate.sleep(Duration.ofSeconds(1));
+
     // Respond to caller
-    return new GreetingResponse("You said hi to " + req.name + "!");
+    return new GreetingResponse(
+        "You said hi to "
+            + req.name
+            + " for the "
+            + Restate.virtualObject(Counter.class, req.name).getAndAdd(1).newValue()
+            + "th time!");
   }
 
   public static void main(String[] args) {
-    RestateHttpServer.listen(Endpoint.bind(new Greeter()));
+    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);
+            }
+          });
+    }
   }
 }

examples/src/main/resources/log4j2.properties

@@ -15,12 +15,23 @@ appender.console.filter.replay.0.type = KeyValuePair
 appender.console.filter.replay.0.key = restateInvocationStatus
 appender.console.filter.replay.0.value = REPLAYING
 
+logger.example.name = my.restate
+logger.example.level = warn
+logger.example.additivity = false
+logger.example.appenderRef.console.ref = consoleLogger
+
 # Restate logs to info level
 logger.app.name = dev.restate
-logger.app.level = info
+logger.app.level = warn
 logger.app.additivity = false
 logger.app.appenderRef.console.ref = consoleLogger
 
+# Restate vm logs to trace level
+logger.core.name = dev.restate.sdk.core.sharedcore
+logger.core.level = warn
+logger.core.additivity = false
+logger.core.appenderRef.console.ref = consoleLogger
+
 # Root logger
 rootLogger.level = warn
 rootLogger.appenderRef.stdout.ref = consoleLogger