Test with many timers

README.md

@@ -56,6 +56,7 @@ Currently, we run tests in the following configurations:
 * `gradle :tests:test`: Default runtime configuration
 * `gradle :tests:testAlwaysSuspending`: Runtime setup to always suspend after replay, to mimic the behavior of RequestResponse stream type
 * `gradle :tests:testSingleThreadSinglePartition`: Runtime setup with a single thread and single partition
+* `gradle :tests:testPersistedTimers`: Runtime setup with timers in memory = 1, to trigger timer queue spilling to disk
 
 ### `VerificationTest` seed
 

build.gradle.kts

@@ -24,7 +24,10 @@ allprojects {
   version = restateVersion
 
   configure<com.diffplug.gradle.spotless.SpotlessExtension> {
-    kotlin { ktfmt() }
+    kotlin {
+      ktfmt()
+      targetExclude("build/generated/**/*.kt")
+    }
     kotlinGradle { ktfmt() }
     java {
       googleJavaFormat()

contracts/build.gradle.kts

@@ -3,14 +3,18 @@ import com.google.protobuf.gradle.protobuf
 
 plugins {
   java
+  kotlin("jvm")
   alias(libs.plugins.protobuf)
 }
 
 dependencies {
   compileOnly(libs.javax.annotation.api)
 
+  api(libs.protobuf.java)
+  api(libs.protobuf.kotlin)
   api(libs.grpc.stub)
   api(libs.grpc.protobuf)
+  api(libs.grpc.kotlin.stub) { exclude("javax.annotation", "javax.annotation-api") }
 
   protobuf(libs.restate.sdk.core)
 }
@@ -21,11 +25,21 @@ protobuf {
     artifact = "com.google.protobuf:protoc:${libs.versions.protobuf.get()}"
   }
 
-  plugins { id("grpc") { artifact = "io.grpc:protoc-gen-grpc-java:${libs.versions.grpc.get()}" } }
+  plugins {
+    id("grpc") { artifact = "io.grpc:protoc-gen-grpc-java:${libs.versions.grpc.get()}" }
+    id("grpckt") {
+      artifact = "io.grpc:protoc-gen-grpc-kotlin:${libs.versions.grpckt.get()}:jdk8@jar"
+    }
+  }
 
   generateProtoTasks {
     ofSourceSet("main").forEach {
-      it.plugins { id("grpc") }
+      it.plugins {
+        id("grpc")
+        id("grpckt")
+      }
+      it.builtins { id("kotlin") }
+
       // Generate descriptors
       it.generateDescriptorSet = true
       it.descriptorSetOptions.includeImports = true

contracts/src/main/proto/coordinator.proto

@@ -12,12 +12,17 @@ service Coordinator {
   option (dev.restate.ext.service_type) = UNKEYED;
 
   rpc Sleep (Duration) returns (google.protobuf.Empty);
+  rpc ManyTimers (ManyTimersRequest) returns (google.protobuf.Empty);
   rpc Proxy (google.protobuf.Empty) returns (ProxyResponse);
   rpc Complex (ComplexRequest) returns (ComplexResponse);
   rpc Timeout (Duration) returns (TimeoutResponse);
   rpc InvokeSequentially(InvokeSequentiallyRequest) returns (google.protobuf.Empty);
 }
 
+message ManyTimersRequest {
+  repeated Duration timer = 1;
+}
+
 message Duration {
   uint64 millis = 1;
 }

services/java-services/src/main/java/dev/restate/e2e/services/coordinator/CoordinatorService.java

@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -27,9 +28,17 @@ public class CoordinatorService extends CoordinatorGrpc.CoordinatorImplBase
 
   @Override
   public void sleep(Duration request, StreamObserver<Empty> responseObserver) {
-    LOG.info("Putting service to sleep for {} ms", request.getMillis());
+    manyTimers(ManyTimersRequest.newBuilder().addTimer(request).build(), responseObserver);
+  }
+
+  @Override
+  public void manyTimers(ManyTimersRequest request, StreamObserver<Empty> responseObserver) {
+    LOG.info("many timers {}", request.getTimerList());
 
-    restateContext().sleep(java.time.Duration.ofMillis(request.getMillis()));
+    awaitableAll(
+        request.getTimerList().stream()
+            .map(d -> restateContext().timer(java.time.Duration.ofMillis(d.getMillis())))
+            .collect(Collectors.toList()));
 
     responseObserver.onNext(Empty.newBuilder().build());
     responseObserver.onCompleted();
@@ -128,19 +137,23 @@ public void invokeSequentially(
       }
     }
 
-    if (collectedAwaitables.size() == 1) {
-      collectedAwaitables.get(0).await();
-    } else if (collectedAwaitables.size() == 2) {
-      Awaitable.all(collectedAwaitables.get(0), collectedAwaitables.get(1)).await();
-    } else if (collectedAwaitables.size() >= 2) {
-      Awaitable.all(
-              collectedAwaitables.get(0),
-              collectedAwaitables.get(1),
-              collectedAwaitables.subList(2, collectedAwaitables.size()).toArray(Awaitable[]::new))
-          .await();
-    }
+    awaitableAll(collectedAwaitables);
 
     responseObserver.onNext(Empty.getDefaultInstance());
     responseObserver.onCompleted();
   }
+
+  private static void awaitableAll(List<Awaitable<?>> awaitables) {
+    if (awaitables.size() == 1) {
+      awaitables.get(0).await();
+    } else if (awaitables.size() == 2) {
+      Awaitable.all(awaitables.get(0), awaitables.get(1)).await();
+    } else if (awaitables.size() >= 2) {
+      Awaitable.all(
+              awaitables.get(0),
+              awaitables.get(1),
+              awaitables.subList(2, awaitables.size()).toArray(Awaitable[]::new))
+          .await();
+    }
+  }
 }

services/node-services/src/coordinator.ts

@@ -5,6 +5,7 @@ import {
   ComplexResponse,
   Coordinator,
   Duration,
+  ManyTimersRequest,
   ProxyResponse,
   TimeoutResponse,
   protobufPackage,
@@ -17,10 +18,16 @@ export const CoordinatorServiceFQN = protobufPackage + ".Coordinator";
 
 export class CoordinatorService implements Coordinator {
   async sleep(request: Duration): Promise<Empty> {
-    console.log("sleep: " + JSON.stringify(request));
+    return this.manyTimers({ timer: [request] });
+  }
+
+  async manyTimers(request: ManyTimersRequest): Promise<Empty> {
+    console.log("many timers: " + JSON.stringify(request));
+
     const ctx = restate.useContext(this);
 
-    await ctx.sleep(request.millis);
+    // Promise.all is not deterministic wrt failures, but this is fine as sleep never fails
+    await Promise.all(request.timer.map((value) => ctx.sleep(value.millis)));
 
     return {};
   }

settings.gradle.kts

@@ -24,6 +24,7 @@ dependencyResolutionManagement {
     create("libs") {
       version("protobuf", "3.21.9")
       version("grpc", "1.50.2")
+      version("grpckt", "1.3.0")
 
       version("log4j", "2.19.0")
 
@@ -43,9 +44,12 @@ dependencyResolutionManagement {
       library("restate-sdk-vertx", "dev.restate.sdk", "sdk-vertx").versionRef("restate")
 
       library("protoc", "com.google.protobuf", "protoc").versionRef("protobuf")
+      library("protobuf-java", "com.google.protobuf", "protobuf-java").versionRef("protobuf")
+      library("protobuf-kotlin", "com.google.protobuf", "protobuf-kotlin").versionRef("protobuf")
       library("grpc-stub", "io.grpc", "grpc-stub").versionRef("grpc")
       library("grpc-protobuf", "io.grpc", "grpc-protobuf").versionRef("grpc")
       library("grpc-netty-shaded", "io.grpc", "grpc-netty-shaded").versionRef("grpc")
+      library("grpc-kotlin-stub", "io.grpc", "grpc-kotlin-stub").versionRef("grpckt")
 
       library("log4j-api", "org.apache.logging.log4j", "log4j-api").versionRef("log4j")
       library("log4j-core", "org.apache.logging.log4j", "log4j-core").versionRef("log4j")

tests/build.gradle.kts

@@ -10,6 +10,7 @@ dependencies {
 
   testImplementation(libs.junit.all)
   testImplementation(libs.assertj)
+  testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.7.1")
 
   testImplementation(libs.log4j.api)
   testRuntimeOnly(libs.log4j.core)
@@ -77,6 +78,20 @@ tasks {
     }
   }
 
+  register<Test>("testPersistedTimers") {
+    environment =
+        environment +
+            baseRestateEnvironment(name) +
+            mapOf("RESTATE_WORKER__TIMERS__NUM_TIMERS_IN_MEMORY_LIMIT" to "1")
+
+    useJUnitPlatform {
+      // Run all the tests with always-suspending or only-always-suspending tag
+      includeTags("timers")
+    }
+    // Increase a bit the default timeout
+    systemProperties["junit.jupiter.execution.timeout.testable.method.default"] = "20 s"
+  }
+
   withType<Test>().configureEach {
     dependsOn(":services:http-server:jibDockerBuild")
     dependsOn(":services:java-services:jibDockerBuild")
@@ -87,4 +102,5 @@ tasks {
 tasks.named("check") {
   dependsOn("testAlwaysSuspending")
   dependsOn("testSingleThreadSinglePartition")
+  dependsOn("testPersistedTimers")
 }

tests/src/test/kotlin/dev/restate/e2e/Containers.kt

@@ -30,7 +30,8 @@ object Containers {
 
   fun getRestateEnvironment(): Map<String, String> {
     return System.getenv().filterKeys {
-      (it.startsWith("RESTATE_") && it != "RESTATE_RUNTIME_CONTAINER") || it.startsWith("RUST_")
+      (it.uppercase().startsWith("RESTATE_") && it.uppercase() != "RESTATE_RUNTIME_CONTAINER") ||
+          it.uppercase().startsWith("RUST_")
     }
   }