From d14145a98dee9336180d8b216d195bf9c06fb5df Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Wed, 1 Jul 2026 16:10:57 +0200 Subject: [PATCH 1/2] Scope and Limit Key API --- .../dev/restate/client/kotlin/ingress.kt | 127 ++++++++++++++- .../main/java/dev/restate/client/Client.java | 56 +++++++ .../client/ClientServiceHandleImpl.java | 15 ++ .../java/dev/restate/client/ScopedClient.java | 152 ++++++++++++++++++ .../dev/restate/client/base/BaseClient.java | 24 ++- .../reflection/kotlin/RequestCaptureProxy.kt | 8 +- .../common/reflection/kotlin/reflections.kt | 8 +- common/build.gradle.kts | 1 + .../dev/restate/common/InvocationOptions.java | 79 +++++++-- .../main/java/dev/restate/common/Request.java | 9 ++ .../dev/restate/common/RequestBuilder.java | 18 +++ .../java/dev/restate/common/RequestImpl.java | 35 +++- .../main/java/dev/restate/common/Target.java | 90 +++++++++-- .../common/reflections/RestateUtils.java | 19 ++- .../sdk/examples/ConcurrencyLimitExample.java | 89 ++++++++++ .../dev/restate/sdk/kotlin/ContextImpl.kt | 2 + .../main/kotlin/dev/restate/sdk/kotlin/api.kt | 101 +++++++++++- .../java/dev/restate/sdk/ContextImpl.java | 2 + .../main/java/dev/restate/sdk/Restate.java | 55 +++++++ .../src/main/java/dev/restate/sdk/Scope.java | 132 +++++++++++++++ .../dev/restate/sdk/ServiceHandleImpl.java | 14 ++ sdk-common/build.gradle.kts | 1 + .../restate/sdk/common/HandlerRequest.java | 103 +++++++++--- .../endpoint/definition/HandlerContext.java | 21 +++ .../ExecutorSwitchingHandlerContextImpl.java | 7 +- .../restate/sdk/core/HandlerContextImpl.java | 13 +- .../restate/sdk/core/ProtocolException.java | 8 + .../sdk/core/legacy/LegacyStateMachine.java | 6 + .../restate/sdk/fake/FakeHandlerContext.java | 10 +- 29 files changed, 1124 insertions(+), 81 deletions(-) create mode 100644 client/src/main/java/dev/restate/client/ScopedClient.java create mode 100644 examples/src/main/java/my/restate/sdk/examples/ConcurrencyLimitExample.java create mode 100644 sdk-api/src/main/java/dev/restate/sdk/Scope.java diff --git a/client-kotlin/src/main/kotlin/dev/restate/client/kotlin/ingress.kt b/client-kotlin/src/main/kotlin/dev/restate/client/kotlin/ingress.kt index b21f4ec3d..9179b2d3f 100644 --- a/client-kotlin/src/main/kotlin/dev/restate/client/kotlin/ingress.kt +++ b/client-kotlin/src/main/kotlin/dev/restate/client/kotlin/ingress.kt @@ -271,6 +271,108 @@ val Response.response: Res val SendResponse.sendStatus: SendResponse.SendStatus get() = this.sendStatus() +/** + * **PREVIEW:** Returns a [ScopedKotlinClient] that routes all outgoing calls within the given + * scope. + * + * **NOTE:** This API is in preview and is not enabled by default. To use it in restate-server 1.7, + * enable the flow control and protocol v7 experimental features, via + * `RESTATE_EXPERIMENTAL_ENABLE_PROTOCOL_V7=true` and `RESTATE_EXPERIMENTAL_ENABLE_VQUEUES=true`. + * These can be enabled only on **new clusters**, for more info check out + * https://docs.restate.dev/services/flow-control#enabling-flow-control. If these experimental + * features aren't enabled, the call fails with a retryable error and keeps retrying until they are. + * + * A scope is a sub-grouping of resources (invocations, virtual object instances, workflow + * instances, concurrency limits) within the Restate cluster. It becomes part of the target identity + * tuple: + * - `scope, service, handler, idempotencyKey?` + * - `scope, virtualObject, objectKey, handler, idempotencyKey?` + * - `scope, workflow, workflowKey, handler` + * + * Under the hood, the scope contributes to the partition key, so all resources in a scope get + * co-located by the restate-server. + * + * Omitting the scope (i.e. using the regular `service` / `workflow` methods) is equivalent to + * calling with no scope, which is the existing behavior. + * + * The scope key must consist only of `[a-zA-Z0-9_.-]` characters, with `1 <= length <= 36` chars. + * + * Example usage: + * ```kotlin + * // Route a call into a named scope + * client.scope("tenant-123").service().process(payload) + * ``` + * + * @param scopeKey the scope identifier + * @see https://docs.restate.dev/services/flow-control + */ +@org.jetbrains.annotations.ApiStatus.Experimental +fun Client.scope(scopeKey: String): ScopedKotlinClient = ScopedKotlinClient(this, scopeKey) + +/** + * **PREVIEW:** A client for making RPC calls within a specific scope. + * + * Obtain an instance via [Client.scope]. + * + * @see Client.scope + */ +@org.jetbrains.annotations.ApiStatus.Experimental +class ScopedKotlinClient +@PublishedApi +internal constructor( + @PublishedApi internal val client: Client, + @PublishedApi internal val scopeKey: String, +) { + /** @see Client.service */ + @org.jetbrains.annotations.ApiStatus.Experimental + inline fun service(): SVC { + return service(client, SVC::class.java, scopeKey) + } + + /** @see Client.virtualObject */ + @org.jetbrains.annotations.ApiStatus.Experimental + inline fun virtualObject(key: String): SVC { + return virtualObject(client, SVC::class.java, key, scopeKey) + } + + /** @see Client.workflow */ + @org.jetbrains.annotations.ApiStatus.Experimental + inline fun workflow(key: String): SVC { + return workflow(client, SVC::class.java, key, scopeKey) + } + + /** @see Client.toService */ + @org.jetbrains.annotations.ApiStatus.Experimental + inline fun toService(): KClientRequestBuilder { + ReflectionUtils.mustHaveServiceAnnotation(SVC::class.java) + require(ReflectionUtils.isKotlinClass(SVC::class.java)) { + "Using Java classes with Kotlin's API is not supported" + } + return KClientRequestBuilder(client, SVC::class.java, null, scopeKey) + } + + /** @see Client.toVirtualObject */ + @org.jetbrains.annotations.ApiStatus.Experimental + inline fun toVirtualObject(key: String): KClientRequestBuilder { + ReflectionUtils.mustHaveVirtualObjectAnnotation(SVC::class.java) + require(ReflectionUtils.isKotlinClass(SVC::class.java)) { + "Using Java classes with Kotlin's API is not supported" + } + return KClientRequestBuilder(client, SVC::class.java, key, scopeKey) + } + + /** @see Client.toWorkflow */ + @org.jetbrains.annotations.ApiStatus.Experimental + inline fun toWorkflow(key: String): KClientRequestBuilder { + ReflectionUtils.mustHaveWorkflowAnnotation(SVC::class.java) + require(ReflectionUtils.isKotlinClass(SVC::class.java)) { + "Using Java classes with Kotlin's API is not supported" + } + return KClientRequestBuilder(client, SVC::class.java, key, scopeKey) + } +} + /** * Create a proxy client for a Restate service. * @@ -329,7 +431,7 @@ inline fun Client.workflow(key: String): SVC { * @return a proxy that intercepts method calls and executes them via the client */ @PublishedApi -internal fun service(client: Client, clazz: Class): SVC { +internal fun service(client: Client, clazz: Class, scope: String? = null): SVC { ReflectionUtils.mustHaveServiceAnnotation(clazz) require(ReflectionUtils.isKotlinClass(clazz)) { "Using Java classes with Kotlin's API is not supported" @@ -337,7 +439,7 @@ internal fun service(client: Client, clazz: Class): SVC { val serviceName = ReflectionUtils.extractServiceName(clazz) return ProxySupport.createProxy(clazz) { invocation -> - val request = invocation.captureInvocation(serviceName, null).toRequest() + val request = invocation.captureInvocation(serviceName, null, scope).toRequest() @Suppress("UNCHECKED_CAST") val continuation = invocation.arguments.last() as Continuation // Start a coroutine that calls the client and resumes the continuation @@ -356,7 +458,12 @@ internal fun service(client: Client, clazz: Class): SVC { * @return a proxy that intercepts method calls and executes them via the client */ @PublishedApi -internal fun virtualObject(client: Client, clazz: Class, key: String): SVC { +internal fun virtualObject( + client: Client, + clazz: Class, + key: String, + scope: String? = null, +): SVC { ReflectionUtils.mustHaveVirtualObjectAnnotation(clazz) require(ReflectionUtils.isKotlinClass(clazz)) { "Using Java classes with Kotlin's API is not supported" @@ -364,7 +471,7 @@ internal fun virtualObject(client: Client, clazz: Class, key: S val serviceName = ReflectionUtils.extractServiceName(clazz) return ProxySupport.createProxy(clazz) { invocation -> - val request = invocation.captureInvocation(serviceName, key).toRequest() + val request = invocation.captureInvocation(serviceName, key, scope).toRequest() @Suppress("UNCHECKED_CAST") val continuation = invocation.arguments.last() as Continuation // Start a coroutine that calls the client and resumes the continuation @@ -383,7 +490,12 @@ internal fun virtualObject(client: Client, clazz: Class, key: S * @return a proxy that intercepts method calls and executes them via the client */ @PublishedApi -internal fun workflow(client: Client, clazz: Class, key: String): SVC { +internal fun workflow( + client: Client, + clazz: Class, + key: String, + scope: String? = null, +): SVC { ReflectionUtils.mustHaveWorkflowAnnotation(clazz) require(ReflectionUtils.isKotlinClass(clazz)) { "Using Java classes with Kotlin's API is not supported" @@ -391,7 +503,7 @@ internal fun workflow(client: Client, clazz: Class, key: String val serviceName = ReflectionUtils.extractServiceName(clazz) return ProxySupport.createProxy(clazz) { invocation -> - val request = invocation.captureInvocation(serviceName, key).toRequest() + val request = invocation.captureInvocation(serviceName, key, scope).toRequest() @Suppress("UNCHECKED_CAST") val continuation = invocation.arguments.last() as Continuation // Start a coroutine that calls the client and resumes the continuation @@ -414,6 +526,7 @@ internal constructor( private val client: Client, private val clazz: Class, private val key: String?, + private val scope: String? = null, ) { /** * Create a request by invoking a method on the target. @@ -428,7 +541,7 @@ internal constructor( suspend fun request(block: suspend SVC.() -> Res): KClientRequest { return KClientRequestImpl( client, - RequestCaptureProxy(clazz, key).capture(block as suspend SVC.() -> Any?).toRequest(), + RequestCaptureProxy(clazz, key, scope).capture(block as suspend SVC.() -> Any?).toRequest(), ) as KClientRequest } diff --git a/client/src/main/java/dev/restate/client/Client.java b/client/src/main/java/dev/restate/client/Client.java index 287097e8d..d52e8025d 100644 --- a/client/src/main/java/dev/restate/client/Client.java +++ b/client/src/main/java/dev/restate/client/Client.java @@ -531,6 +531,62 @@ default Response> getOutput() throws IngressException { } } + /** + * PREVIEW: Returns a {@link ScopedClient} that routes all outgoing calls within the given + * scope. + * + *

NOTE: This API is in preview and is not enabled by default. To use it in + * restate-server 1.7, enable the flow control and protocol v7 experimental features, via {@code + * RESTATE_EXPERIMENTAL_ENABLE_PROTOCOL_V7=true} and {@code + * RESTATE_EXPERIMENTAL_ENABLE_VQUEUES=true}. These can be enabled only on new clusters, + * for more info check out https://docs.restate.dev/services/flow-control#enabling-flow-control. + * If these experimental features aren't enabled, the call fails with a retryable error and keeps + * retrying until they are. + * + *

A scope is a sub-grouping of resources (invocations, virtual object instances, workflow + * instances, concurrency limits) within the Restate cluster. It becomes part of the target + * identity tuple: + * + *

    + *
  • {@code scope, service, handler, idempotencyKey?} + *
  • {@code scope, virtualObject, objectKey, handler, idempotencyKey?} + *
  • {@code scope, workflow, workflowKey, handler} + *
+ * + *

Under the hood, the scope contributes to the partition key, so all resources in a scope get + * co-located by the restate-server. + * + *

Omitting the scope (i.e. using the regular {@link #service(Class)} / {@link #workflow(Class, + * String)} methods) is equivalent to calling with no scope, which is the existing behavior. + * + *

The scope key must consist only of {@code [a-zA-Z0-9_.-]} characters, with {@code 1 <= + * length <= 36} chars. + * + * 

{@code
+   * Client client = Client.connect("http://localhost:8080");
+   *
+   * // Route a call into a named scope
+   * client.scope("tenant-123").service(MyService.class).process(payload);
+   *
+   * // Idempotency keys are scoped — "req-1" in "tenant-123" is distinct from "req-1" in "tenant-456"
+   * client.scope("tenant-123").serviceHandle(MyService.class)
+   *     .call(MyService::process, payload, InvocationOptions.idempotencyKey("req-1").build());
+   *
+   * // Combine with a limit key to enforce per-scope concurrency limits
+   * client.scope("tenant-123").workflowHandle(MyWorkflow.class, "wf-key")
+   *     .call(MyWorkflow::run, input, InvocationOptions.limitKey("api-key/user42").build());
+   * }
+ * + * @param scopeKey the scope identifier + * @return a {@link ScopedClient} + * @see https://docs.restate.dev/services/flow-control + */ + @org.jetbrains.annotations.ApiStatus.Experimental + default ScopedClient scope(String scopeKey) { + return new ScopedClient(this, scopeKey); + } + /** * Simple API to invoke a Restate service from the ingress. * diff --git a/client/src/main/java/dev/restate/client/ClientServiceHandleImpl.java b/client/src/main/java/dev/restate/client/ClientServiceHandleImpl.java index 6fc122376..1a39274a8 100644 --- a/client/src/main/java/dev/restate/client/ClientServiceHandleImpl.java +++ b/client/src/main/java/dev/restate/client/ClientServiceHandleImpl.java @@ -29,14 +29,21 @@ final class ClientServiceHandleImpl implements ClientServiceHandle { private final Class clazz; private final String serviceName; private final @Nullable String key; + private final @Nullable String scope; private MethodInfoCollector methodInfoCollector; ClientServiceHandleImpl(Client innerClient, Class clazz, @Nullable String key) { + this(innerClient, clazz, key, null); + } + + ClientServiceHandleImpl( + Client innerClient, Class clazz, @Nullable String key, @Nullable String scope) { this.innerClient = innerClient; this.clazz = clazz; this.serviceName = ReflectionUtils.extractServiceName(clazz); this.key = key; + this.scope = scope; } @SuppressWarnings("unchecked") @@ -46,6 +53,7 @@ public CompletableFuture> callAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s, input); return innerClient.callAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -62,6 +70,7 @@ public CompletableFuture> callAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s, input); return innerClient.callAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -78,6 +87,7 @@ public CompletableFuture> callAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s); return innerClient.callAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -93,6 +103,7 @@ public CompletableFuture> callAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s); return innerClient.callAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -109,6 +120,7 @@ public CompletableFuture> sendAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s, input); return innerClient.sendAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -126,6 +138,7 @@ public CompletableFuture> sendAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s, input); return innerClient.sendAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -143,6 +156,7 @@ public CompletableFuture> sendAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s); return innerClient.sendAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -159,6 +173,7 @@ public CompletableFuture> sendAsync( MethodInfo methodInfo = getMethodInfoCollector().resolve(s); return innerClient.sendAsync( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), diff --git a/client/src/main/java/dev/restate/client/ScopedClient.java b/client/src/main/java/dev/restate/client/ScopedClient.java new file mode 100644 index 000000000..974b73e76 --- /dev/null +++ b/client/src/main/java/dev/restate/client/ScopedClient.java @@ -0,0 +1,152 @@ +// 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 dev.restate.client; + +import dev.restate.common.Request; +import dev.restate.common.Target; +import dev.restate.common.reflections.MethodInfo; +import dev.restate.common.reflections.ProxySupport; +import dev.restate.common.reflections.ReflectionUtils; +import dev.restate.serde.TypeTag; + +/** + * PREVIEW: A client for making RPC calls within a specific scope. + * + *

Obtain an instance via {@link Client#scope(String)}. + * + * @see Client#scope(String) + */ +@org.jetbrains.annotations.ApiStatus.Experimental +public final class ScopedClient { + + private final Client client; + private final String scope; + + ScopedClient(Client client, String scope) { + this.client = client; + this.scope = scope; + } + + /** + * @see Client#service(Class) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public SVC service(Class clazz) { + ReflectionUtils.mustHaveServiceAnnotation(clazz); + if (ReflectionUtils.isKotlinClass(clazz)) { + throw new IllegalArgumentException("Using Kotlin classes with Java's API is not supported"); + } + var serviceName = ReflectionUtils.extractServiceName(clazz); + return ProxySupport.createProxy( + clazz, + invocation -> { + var methodInfo = MethodInfo.fromMethod(invocation.getMethod()); + + //noinspection unchecked + return client + .call( + Request.of( + Target.virtualObject(scope, serviceName, null, methodInfo.getHandlerName()), + (TypeTag) methodInfo.getInputType(), + (TypeTag) methodInfo.getOutputType(), + invocation.getArguments().length == 0 ? null : invocation.getArguments()[0])) + .response(); + }); + } + + /** + * @see Client#serviceHandle(Class) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public ClientServiceHandle serviceHandle(Class clazz) { + ReflectionUtils.mustHaveServiceAnnotation(clazz); + if (ReflectionUtils.isKotlinClass(clazz)) { + throw new IllegalArgumentException("Using Kotlin classes with Java's API is not supported"); + } + return new ClientServiceHandleImpl<>(client, clazz, null, scope); + } + + /** + * @see Client#virtualObject(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public SVC virtualObject(Class clazz, String key) { + ReflectionUtils.mustHaveVirtualObjectAnnotation(clazz); + if (ReflectionUtils.isKotlinClass(clazz)) { + throw new IllegalArgumentException("Using Kotlin classes with Java's API is not supported"); + } + var serviceName = ReflectionUtils.extractServiceName(clazz); + return ProxySupport.createProxy( + clazz, + invocation -> { + var methodInfo = MethodInfo.fromMethod(invocation.getMethod()); + + //noinspection unchecked + return client + .call( + Request.of( + Target.virtualObject(scope, serviceName, key, methodInfo.getHandlerName()), + (TypeTag) methodInfo.getInputType(), + (TypeTag) methodInfo.getOutputType(), + invocation.getArguments().length == 0 ? null : invocation.getArguments()[0])) + .response(); + }); + } + + /** + * @see Client#virtualObjectHandle(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public ClientServiceHandle virtualObjectHandle(Class clazz, String key) { + ReflectionUtils.mustHaveVirtualObjectAnnotation(clazz); + if (ReflectionUtils.isKotlinClass(clazz)) { + throw new IllegalArgumentException("Using Kotlin classes with Java's API is not supported"); + } + return new ClientServiceHandleImpl<>(client, clazz, key, scope); + } + + /** + * @see Client#workflow(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public SVC workflow(Class clazz, String key) { + ReflectionUtils.mustHaveWorkflowAnnotation(clazz); + if (ReflectionUtils.isKotlinClass(clazz)) { + throw new IllegalArgumentException("Using Kotlin classes with Java's API is not supported"); + } + var serviceName = ReflectionUtils.extractServiceName(clazz); + return ProxySupport.createProxy( + clazz, + invocation -> { + var methodInfo = MethodInfo.fromMethod(invocation.getMethod()); + + //noinspection unchecked + return client + .call( + Request.of( + Target.virtualObject(scope, serviceName, key, methodInfo.getHandlerName()), + (TypeTag) methodInfo.getInputType(), + (TypeTag) methodInfo.getOutputType(), + invocation.getArguments().length == 0 ? null : invocation.getArguments()[0])) + .response(); + }); + } + + /** + * @see Client#workflowHandle(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public ClientServiceHandle workflowHandle(Class clazz, String key) { + ReflectionUtils.mustHaveWorkflowAnnotation(clazz); + if (ReflectionUtils.isKotlinClass(clazz)) { + throw new IllegalArgumentException("Using Kotlin classes with Java's API is not supported"); + } + return new ClientServiceHandleImpl<>(client, clazz, key, scope); + } +} diff --git a/client/src/main/java/dev/restate/client/base/BaseClient.java b/client/src/main/java/dev/restate/client/base/BaseClient.java index d9742a450..a3fdcffcc 100644 --- a/client/src/main/java/dev/restate/client/base/BaseClient.java +++ b/client/src/main/java/dev/restate/client/base/BaseClient.java @@ -64,7 +64,7 @@ public CompletableFuture> callAsync(Request r Serde reqSerde = this.serdeFactory.create(request.getRequestTypeTag()); Serde resSerde = this.serdeFactory.create(request.getResponseTypeTag()); - URI requestUri = toRequestURI(request.getTarget(), false, null); + URI requestUri = toRequestURI(request.getTarget(), false, null, request.getLimitKey()); Stream> headersStream = Stream.concat( baseOptions.headers().entrySet().stream(), @@ -92,7 +92,7 @@ public CompletableFuture> sendAsync( Request request, @Nullable Duration delay) { Serde reqSerde = this.serdeFactory.create(request.getRequestTypeTag()); - URI requestUri = toRequestURI(request.getTarget(), true, delay); + URI requestUri = toRequestURI(request.getTarget(), true, delay, request.getLimitKey()); Stream> headersStream = Stream.concat( baseOptions.headers().entrySet().stream(), @@ -435,13 +435,29 @@ private String targetToURI(Target target) { return builder.toString(); } - private URI toRequestURI(Target target, boolean isSend, @Nullable Duration delay) { + private URI toRequestURI( + Target target, boolean isSend, @Nullable Duration delay, @Nullable String limitKey) { StringBuilder builder = new StringBuilder(targetToURI(target)); if (isSend) { builder.append("/send"); } + String separator = "?"; + if (target.getScope() != null) { + builder + .append(separator) + .append("scope=") + .append(URLEncoder.encode(target.getScope(), StandardCharsets.UTF_8)); + separator = "&"; + } + if (limitKey != null) { + builder + .append(separator) + .append("limitKey=") + .append(URLEncoder.encode(limitKey, StandardCharsets.UTF_8)); + separator = "&"; + } if (delay != null && !delay.isZero() && !delay.isNegative()) { - builder.append("?delay=").append(delay); + builder.append(separator).append("delay=").append(delay); } return this.baseUri.resolve(builder.toString()); diff --git a/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/RequestCaptureProxy.kt b/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/RequestCaptureProxy.kt index c50ad3d10..dee7a53e7 100644 --- a/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/RequestCaptureProxy.kt +++ b/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/RequestCaptureProxy.kt @@ -23,7 +23,11 @@ import dev.restate.common.reflections.ReflectionUtils * @property serviceName the resolved service name * @property key the virtual object/workflow key (null for stateless services) */ -class RequestCaptureProxy(private val clazz: Class, private val key: String?) { +class RequestCaptureProxy( + private val clazz: Class, + private val key: String?, + private val scope: String? = null, +) { private val serviceName: String = ReflectionUtils.extractServiceName(clazz) @@ -36,7 +40,7 @@ class RequestCaptureProxy(private val clazz: Class, private val suspend fun capture(block: suspend SVC.() -> Any?): CapturedInvocation { val proxy = ProxySupport.createProxy(clazz) { invocation -> - throw invocation.captureInvocation(serviceName, key) + throw invocation.captureInvocation(serviceName, key, scope) } try { diff --git a/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/reflections.kt b/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/reflections.kt index 0b30577d6..291217a0e 100644 --- a/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/reflections.kt +++ b/common-kotlin/src/main/kotlin/dev/restate/common/reflection/kotlin/reflections.kt @@ -46,6 +46,7 @@ data class CapturedInvocation( fun ProxyFactory.MethodInvocation.captureInvocation( serviceName: String, key: String?, + scope: String? = null, ): CapturedInvocation { val handlerInfo = ReflectionUtils.mustHaveHandlerAnnotation(method) val handlerName = handlerInfo.name @@ -73,12 +74,7 @@ fun ProxyFactory.MethodInvocation.captureInvocation( kFunction.findAnnotation(), ) - val target = - if (key != null) { - Target.virtualObject(serviceName, key, handlerName) - } else { - Target.service(serviceName, handlerName) - } + val target = Target.virtualObject(scope, serviceName, key, handlerName) // For suspend functions, arguments are: [input?, continuation] // Extract the input (first argument, excluding continuation) diff --git a/common/build.gradle.kts b/common/build.gradle.kts index 27cfb5f99..b9245b262 100644 --- a/common/build.gradle.kts +++ b/common/build.gradle.kts @@ -9,6 +9,7 @@ description = "Common types used by different Restate Java modules" dependencies { compileOnly(libs.jspecify) + compileOnly(libs.jetbrains.annotations) implementation(libs.log4j.api) diff --git a/common/src/main/java/dev/restate/common/InvocationOptions.java b/common/src/main/java/dev/restate/common/InvocationOptions.java index ab4d58094..6d31069da 100644 --- a/common/src/main/java/dev/restate/common/InvocationOptions.java +++ b/common/src/main/java/dev/restate/common/InvocationOptions.java @@ -15,14 +15,18 @@ public class InvocationOptions { - public static final InvocationOptions DEFAULT = new InvocationOptions(null, null); + public static final InvocationOptions DEFAULT = new InvocationOptions(null, null, null); private final @Nullable String idempotencyKey; + private final @Nullable String limitKey; private final @Nullable LinkedHashMap headers; InvocationOptions( - @Nullable String idempotencyKey, @Nullable LinkedHashMap headers) { + @Nullable String idempotencyKey, + @Nullable String limitKey, + @Nullable LinkedHashMap headers) { this.idempotencyKey = idempotencyKey; + this.limitKey = limitKey; this.headers = headers; } @@ -30,6 +34,15 @@ public class InvocationOptions { return idempotencyKey; } + /** + * PREVIEW: Limit key to use within the scope. Requires {@code scope} to be set (see {@link + * Target#scoped(String)}). + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public @Nullable String getLimitKey() { + return limitKey; + } + public @Nullable Map getHeaders() { return headers; } @@ -38,12 +51,13 @@ public class InvocationOptions { public boolean equals(Object o) { if (!(o instanceof InvocationOptions that)) return false; return Objects.equals(getIdempotencyKey(), that.getIdempotencyKey()) + && Objects.equals(getLimitKey(), that.getLimitKey()) && Objects.equals(getHeaders(), that.getHeaders()); } @Override public int hashCode() { - return Objects.hash(getIdempotencyKey(), getHeaders()); + return Objects.hash(getIdempotencyKey(), getLimitKey(), getHeaders()); } @Override @@ -52,34 +66,50 @@ public String toString() { + "idempotencyKey='" + idempotencyKey + '\'' + + ", limitKey='" + + limitKey + + '\'' + ", headers=" + headers + '}'; } public static Builder builder() { - return new Builder(null, null); + return new Builder(null, null, null); } public static Builder idempotencyKey(String idempotencyKey) { - return new Builder(null, null).idempotencyKey(idempotencyKey); + return new Builder(null, null, null).idempotencyKey(idempotencyKey); + } + + /** + * PREVIEW: Create a builder with the given limit key. Limit key to use within the scope. + * Requires {@code scope} to be set (see {@link Target#scoped(String)}). + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public static Builder limitKey(String limitKey) { + return new Builder(null, null, null).limitKey(limitKey); } public static Builder header(String key, String value) { - return new Builder(null, null).header(key, value); + return new Builder(null, null, null).header(key, value); } public static Builder headers(Map newHeaders) { - return new Builder(null, null).headers(newHeaders); + return new Builder(null, null, null).headers(newHeaders); } public static final class Builder { @Nullable private String idempotencyKey; + @Nullable private String limitKey; @Nullable private LinkedHashMap headers; private Builder( - @Nullable String idempotencyKey, @Nullable LinkedHashMap headers) { + @Nullable String idempotencyKey, + @Nullable String limitKey, + @Nullable LinkedHashMap headers) { this.idempotencyKey = idempotencyKey; + this.limitKey = limitKey; this.headers = headers; } @@ -92,6 +122,19 @@ public Builder idempotencyKey(String idempotencyKey) { return this; } + /** + * PREVIEW: Limit key to use within the scope. Requires {@code scope} to be set (see + * {@link Target#scoped(String)}). + * + * @param limitKey Limit key to attach in the request. + * @return this instance, so the builder can be used fluently. + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public Builder limitKey(String limitKey) { + this.limitKey = limitKey; + return this; + } + /** * Append this header to the list of configured headers. * @@ -132,6 +175,22 @@ public void setIdempotencyKey(@Nullable String idempotencyKey) { idempotencyKey(idempotencyKey); } + /** PREVIEW. */ + @org.jetbrains.annotations.ApiStatus.Experimental + public @Nullable String getLimitKey() { + return limitKey; + } + + /** + * PREVIEW. + * + * @param limitKey Limit key to attach in the request. + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public void setLimitKey(@Nullable String limitKey) { + this.limitKey = limitKey; + } + public @Nullable Map getHeaders() { return headers; } @@ -148,11 +207,11 @@ public void setHeaders(@Nullable Map headers) { * @return build the request */ public InvocationOptions build() { - return new InvocationOptions(this.idempotencyKey, this.headers); + return new InvocationOptions(this.idempotencyKey, this.limitKey, this.headers); } } public Builder toBuilder() { - return new Builder(this.idempotencyKey, this.headers); + return new Builder(this.idempotencyKey, this.limitKey, this.headers); } } diff --git a/common/src/main/java/dev/restate/common/Request.java b/common/src/main/java/dev/restate/common/Request.java index 4a503e8bf..e47ac7e2f 100644 --- a/common/src/main/java/dev/restate/common/Request.java +++ b/common/src/main/java/dev/restate/common/Request.java @@ -73,6 +73,15 @@ static RequestBuilder of(Target target, byte[] request) { */ @Nullable String getIdempotencyKey(); + /** + * PREVIEW: Limit key to use within the scope. Requires {@code scope} to be set. + * + * @return the limit key + * @see Target#scoped(String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + @Nullable String getLimitKey(); + /** * @return the request headers */ diff --git a/common/src/main/java/dev/restate/common/RequestBuilder.java b/common/src/main/java/dev/restate/common/RequestBuilder.java index ae9643d41..ded593ff9 100644 --- a/common/src/main/java/dev/restate/common/RequestBuilder.java +++ b/common/src/main/java/dev/restate/common/RequestBuilder.java @@ -25,6 +25,24 @@ public interface RequestBuilder extends Request { */ RequestBuilder setIdempotencyKey(@Nullable String idempotencyKey); + /** + * PREVIEW: Limit key to use within the scope. Requires {@code scope} to be set (see {@link + * Target#scoped(String)}). + * + * @param limitKey Limit key to attach in the request. + * @return this instance, so the builder can be used fluently. + */ + @org.jetbrains.annotations.ApiStatus.Experimental + RequestBuilder limitKey(@Nullable String limitKey); + + /** + * PREVIEW. + * + * @param limitKey Limit key to attach in the request. + */ + @org.jetbrains.annotations.ApiStatus.Experimental + RequestBuilder setLimitKey(@Nullable String limitKey); + /** * Append this header to the list of configured headers. * diff --git a/common/src/main/java/dev/restate/common/RequestImpl.java b/common/src/main/java/dev/restate/common/RequestImpl.java index 41faf3b4a..80fb7decb 100644 --- a/common/src/main/java/dev/restate/common/RequestImpl.java +++ b/common/src/main/java/dev/restate/common/RequestImpl.java @@ -21,6 +21,7 @@ final class RequestImpl implements WorkflowRequest { private final TypeTag resTypeTag; private final Req request; @Nullable private final String idempotencyKey; + @Nullable private final String limitKey; @Nullable private final LinkedHashMap headers; RequestImpl( @@ -29,12 +30,14 @@ final class RequestImpl implements WorkflowRequest { TypeTag resTypeTag, Req request, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable LinkedHashMap headers) { this.target = target; this.reqTypeTag = reqTypeTag; this.resTypeTag = resTypeTag; this.request = request; this.idempotencyKey = idempotencyKey; + this.limitKey = limitKey; this.headers = headers; } @@ -63,6 +66,11 @@ public Req getRequest() { return idempotencyKey; } + @Override + public @Nullable String getLimitKey() { + return limitKey; + } + @Override public Map getHeaders() { if (this.headers == null) { @@ -77,6 +85,7 @@ static final class Builder implements WorkflowRequestBuilder private final TypeTag resTypeTag; private final Req request; @Nullable private String idempotencyKey; + @Nullable private String limitKey; @Nullable private LinkedHashMap headers; Builder( @@ -85,12 +94,14 @@ static final class Builder implements WorkflowRequestBuilder TypeTag resTypeTag, Req request, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable LinkedHashMap headers) { this.target = target; this.reqTypeTag = reqTypeTag; this.resTypeTag = resTypeTag; this.request = request; this.idempotencyKey = idempotencyKey; + this.limitKey = limitKey; this.headers = headers; } @@ -177,6 +188,22 @@ public Builder setIdempotencyKey(@Nullable String idempotencyKey) { return idempotencyKey(idempotencyKey); } + @Override + public Builder limitKey(@Nullable String limitKey) { + this.limitKey = limitKey; + return this; + } + + @Override + public @Nullable String getLimitKey() { + return limitKey; + } + + @Override + public Builder setLimitKey(@Nullable String limitKey) { + return limitKey(limitKey); + } + @Override public @Nullable Map getHeaders() { return headers; @@ -206,6 +233,7 @@ public RequestImpl build() { this.resTypeTag, this.request, this.idempotencyKey, + this.limitKey, this.headers); } } @@ -218,6 +246,7 @@ public Builder toBuilder() { this.resTypeTag, this.request, this.idempotencyKey, + this.limitKey, this.headers); } @@ -229,12 +258,13 @@ public boolean equals(Object o) { && Objects.equals(resTypeTag, that.getResponseTypeTag()) && Objects.equals(request, that.getRequest()) && Objects.equals(idempotencyKey, that.getIdempotencyKey()) + && Objects.equals(limitKey, that.getLimitKey()) && Objects.equals(headers, that.getHeaders()); } @Override public int hashCode() { - return Objects.hash(target, reqTypeTag, resTypeTag, request, idempotencyKey, headers); + return Objects.hash(target, reqTypeTag, resTypeTag, request, idempotencyKey, limitKey, headers); } @Override @@ -251,6 +281,9 @@ public String toString() { + ", idempotencyKey='" + idempotencyKey + '\'' + + ", limitKey='" + + limitKey + + '\'' + ", headers=" + headers + '}'; diff --git a/common/src/main/java/dev/restate/common/Target.java b/common/src/main/java/dev/restate/common/Target.java index cdbd70e3a..8da76e89c 100644 --- a/common/src/main/java/dev/restate/common/Target.java +++ b/common/src/main/java/dev/restate/common/Target.java @@ -14,26 +14,89 @@ /** Represents an invocation target. */ public final class Target { + private final @Nullable String scope; private final String service; private final String handler; - private final String key; + private final @Nullable String key; - private Target(String service, String handler, String key) { + private Target(@Nullable String scope, String service, String handler, @Nullable String key) { + this.scope = scope; this.service = service; this.handler = handler; this.key = key; } public static Target virtualObject(String name, String key, String handler) { - return new Target(name, handler, key); + return new Target(null, name, handler, key); + } + + /** + * PREVIEW. + * + * @see #scoped(String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public static Target virtualObject(String scope, String name, String key, String handler) { + return new Target(scope, name, handler, key); } public static Target workflow(String name, String key, String handler) { - return new Target(name, handler, key); + return new Target(null, name, handler, key); + } + + /** + * PREVIEW. + * + * @see #scoped(String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public static Target workflow(String scope, String name, String key, String handler) { + return new Target(scope, name, handler, key); } public static Target service(String name, String handler) { - return new Target(name, handler, null); + return new Target(null, name, handler, null); + } + + /** + * PREVIEW. + * + * @see #scoped(String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public static Target service(String scope, String name, String handler) { + return new Target(scope, name, handler, null); + } + + /** + * PREVIEW: Returns a new {@link Target} routed within the given scope. + * + *

A scope is a sub-grouping of resources (invocations, virtual object instances, workflow + * instances, concurrency limits) within the Restate cluster. It becomes part of the target + * identity and contributes to the partition key, so all resources sharing a scope get co-located. + * The scope key must consist only of {@code [a-zA-Z0-9_.-]} characters, with {@code 1 <= length + * <= 36}. + * + *

Requires Restate >= 1.7. See flow + * control. + * + * @param scope the scope key to route this target within + * @return a new {@link Target} with the given scope + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public Target scoped(String scope) { + return new Target(scope, this.service, this.handler, this.key); + } + + /** + * PREVIEW: Scope of the request target. + * + * @return the scope. Null if no scope is set. + * @see #scoped(String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public @Nullable String getScope() { + return scope; } public String getService() { @@ -56,21 +119,28 @@ public boolean equals(Object object) { if (this == object) return true; if (object == null || getClass() != object.getClass()) return false; Target target = (Target) object; - return Objects.equals(service, target.service) + return Objects.equals(scope, target.scope) + && Objects.equals(service, target.service) && Objects.equals(handler, target.handler) && Objects.equals(key, target.key); } @Override public int hashCode() { - return Objects.hash(service, handler, key); + return Objects.hash(scope, service, handler, key); } @Override public String toString() { - if (key == null) { - return service + "/" + handler; + StringBuilder sb = new StringBuilder(); + if (scope != null) { + sb.append(scope).append("/"); + } + sb.append(service).append("/"); + if (key != null) { + sb.append(key).append("/"); } - return service + "/" + key + "/" + handler; + sb.append(handler); + return sb.toString(); } } diff --git a/common/src/main/java/dev/restate/common/reflections/RestateUtils.java b/common/src/main/java/dev/restate/common/reflections/RestateUtils.java index 4f5786389..e03d9f1aa 100644 --- a/common/src/main/java/dev/restate/common/reflections/RestateUtils.java +++ b/common/src/main/java/dev/restate/common/reflections/RestateUtils.java @@ -19,6 +19,7 @@ public final class RestateUtils { + @Deprecated public static Request toRequest( String serviceName, @Nullable String key, @@ -27,11 +28,23 @@ public static Request toRequest( TypeTag resTypeTag, Req request, @Nullable InvocationOptions options) { - var builder = - Request.of( - Target.virtualObject(serviceName, key, handlerName), reqTypeTag, resTypeTag, request); + return toRequest(null, serviceName, key, handlerName, reqTypeTag, resTypeTag, request, options); + } + + public static Request toRequest( + @Nullable String scope, + String serviceName, + @Nullable String key, + String handlerName, + TypeTag reqTypeTag, + TypeTag resTypeTag, + Req request, + @Nullable InvocationOptions options) { + Target target = Target.virtualObject(scope, serviceName, key, handlerName); + var builder = Request.of(target, reqTypeTag, resTypeTag, request); if (options != null) { builder.setIdempotencyKey(options.getIdempotencyKey()); + builder.setLimitKey(options.getLimitKey()); if (options.getHeaders() != null) { builder.setHeaders(options.getHeaders()); } diff --git a/examples/src/main/java/my/restate/sdk/examples/ConcurrencyLimitExample.java b/examples/src/main/java/my/restate/sdk/examples/ConcurrencyLimitExample.java new file mode 100644 index 000000000..b37505a32 --- /dev/null +++ b/examples/src/main/java/my/restate/sdk/examples/ConcurrencyLimitExample.java @@ -0,0 +1,89 @@ +// 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.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; + +public class ConcurrencyLimitExample { + + // --- Amazon Merchant Service: a third-party API wrapper --- + + @Service + public static class AmazonMerchantService { + + public record CheckoutRequest(String orderId, String productId, int quantity) {} + + public record CheckoutResponse(String confirmationId) {} + + @Handler + public CheckoutResponse checkout(CheckoutRequest req) { + // In a real app, this would call the Amazon Merchant API + // using the user-provided API key from the scope. + return new CheckoutResponse("conf-" + req.orderId); + } + } + + // --- Order Processor: uses scoped calls to rate-limit per API key --- + + @Service + public static class OrderProcessor { + + public record ProcessOrderRequest(String orderId, String amazonApiKey) {} + + /** + * Process an order by calling AmazonMerchantService within a scope keyed by the user's Amazon + * API key. + * + *

The scope + configured rate limit rules ensure that calls sharing the same API key are + * rate-limited (e.g. 10 requests every 2 hours), preventing us from exceeding the third-party + * API quota. + * + *

Rate limit rules are configured externally in Restate: + * + * 

{@code
+     * // Default rule: on any scope, rate limit AmazonMerchantService to 10 req / 2h
+     * {
+     *   "scope": { "any": true },
+     *   "match": { "service": "AmazonMerchantService" },
+     *   "limit": {
+     *     "rateLimit": { "count": 10, "interval": { "hours": 2 } }
+     *   }
+     * }
+     *
+     * // Override for a specific API key: increase limit to 100 req / 2h
+     * {
+     *   "scope": { "equals": "amz-api-key-123" },
+     *   "match": { "service": "AmazonMerchantService" },
+     *   "limit": {
+     *     "rateLimit": { "count": 100, "interval": { "hours": 2 } }
+     *   }
+     * }
+     * }
+ */ + @Handler + public String processOrder(ProcessOrderRequest req) { + // Scope the call by the user's Amazon API key. + // Restate enforces the rate limit rules configured above. + var response = + Restate.scope(req.amazonApiKey) + .service(AmazonMerchantService.class) + .checkout(new AmazonMerchantService.CheckoutRequest(req.orderId, "product-42", 1)); + + return "Order " + req.orderId + " confirmed: " + response.confirmationId(); + } + } + + public static void main(String[] args) { + RestateHttpServer.listen(Endpoint.bind(new AmazonMerchantService()).bind(new OrderProcessor())); + } +} diff --git a/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/ContextImpl.kt b/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/ContextImpl.kt index 1806fce2e..60dfee7fb 100644 --- a/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/ContextImpl.kt +++ b/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/ContextImpl.kt @@ -96,6 +96,7 @@ internal constructor( request.getTarget(), resolveAndSerialize(request.getRequestTypeTag(), request.getRequest()), request.getIdempotencyKey(), + request.getLimitKey(), request.getHeaders()?.entries, ) .await() @@ -121,6 +122,7 @@ internal constructor( request.getTarget(), resolveAndSerialize(request.getRequestTypeTag(), request.getRequest()), request.getIdempotencyKey(), + request.getLimitKey(), request.getHeaders()?.entries, delay?.toJavaDuration(), ) diff --git a/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/api.kt b/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/api.kt index dd7e04dfc..f42b3ec3c 100644 --- a/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/api.kt +++ b/sdk-api-kotlin/src/main/kotlin/dev/restate/sdk/kotlin/api.kt @@ -844,6 +844,20 @@ val HandlerRequest.bodyAsByteBuffer: ByteBuffer get() = this.bodyAsBodyBuffer() val HandlerRequest.headers: Map get() = this.headers() +/** **PREVIEW:** The scope key with which this invocation was submitted, if any. */ +@get:org.jetbrains.annotations.ApiStatus.Experimental +val HandlerRequest.scope: String? + get() = this.scope() + +/** **PREVIEW:** The limit key with which this invocation was submitted, if any. */ +@get:org.jetbrains.annotations.ApiStatus.Experimental +val HandlerRequest.limitKey: String? + get() = this.limitKey() + +/** The idempotency key with which this invocation was submitted, if any. */ +@get:org.jetbrains.annotations.ApiStatus.Experimental +val HandlerRequest.idempotencyKey: String? + get() = this.idempotencyKey() // ============================================================================= // Free-floating API functions for the reflection-based API @@ -1466,6 +1480,81 @@ private class KRequestImpl(private val request: Request) : } } +/** + * **PREVIEW:** Returns a [KScope] that routes all outgoing calls within the given scope. + * + * **NOTE:** This API is in preview and is not enabled by default. To use it in restate-server 1.7, + * enable the flow control and protocol v7 experimental features, via + * `RESTATE_EXPERIMENTAL_ENABLE_PROTOCOL_V7=true` and `RESTATE_EXPERIMENTAL_ENABLE_VQUEUES=true`. + * These can be enabled only on **new clusters**, for more info check out + * https://docs.restate.dev/services/flow-control#enabling-flow-control. If these experimental + * features aren't enabled, the call fails with a retryable error and keeps retrying until they are. + * + * A scope is a sub-grouping of resources (invocations, virtual object instances, workflow + * instances, concurrency limits) within the Restate cluster. It becomes part of the target identity + * tuple: + * - `scope, service, handler, idempotencyKey?` + * - `scope, virtualObject, objectKey, handler, idempotencyKey?` + * - `scope, workflow, workflowKey, handler` + * + * Under the hood, the scope contributes to the partition key, so all resources in a scope get + * co-located by the restate-server. + * + * Omitting the scope (i.e. using the regular `service` / `workflow` methods) is equivalent to + * calling with no scope, which is the existing behavior. + * + * The scope key must consist only of `[a-zA-Z0-9_.-]` characters, with `1 <= length <= 36` chars. + * + * Example usage: + * ```kotlin + * @Handler + * suspend fun myHandler(): String { + * // Route a call into a named scope + * val greeter = scope("tenant-123").service() + * val response = greeter.greet("Alice") + * return "Got: $response" + * } + * ``` + * + * @param scopeKey the scope identifier + * @see https://docs.restate.dev/services/flow-control + */ +@org.jetbrains.annotations.ApiStatus.Experimental +fun scope(scopeKey: String): KScope = KScope(scopeKey) + +/** + * **PREVIEW:** A context for making RPC calls within a specific scope. + * + * Obtain an instance via [scope]. + * + * @see scope + */ +@org.jetbrains.annotations.ApiStatus.Experimental +class KScope +@PublishedApi +internal constructor( + @PublishedApi internal val scopeKey: String, +) { + /** @see service */ + @org.jetbrains.annotations.ApiStatus.Experimental + suspend inline fun service(): SVC { + return service(SVC::class.java, scopeKey) + } + + /** @see virtualObject */ + @org.jetbrains.annotations.ApiStatus.Experimental + suspend inline fun virtualObject(key: String): SVC { + return virtualObject(SVC::class.java, key, scopeKey) + } + + /** @see workflow */ + @org.jetbrains.annotations.ApiStatus.Experimental + suspend inline fun workflow(key: String): SVC { + return workflow(SVC::class.java, key, scopeKey) + } +} + /** * Create a proxy client for a Restate service. * @@ -1521,7 +1610,7 @@ suspend inline fun workflow(key: String): SVC { } @PublishedApi -internal fun service(clazz: Class): SVC { +internal fun service(clazz: Class, scope: String? = null): SVC { ReflectionUtils.mustHaveServiceAnnotation(clazz) require(ReflectionUtils.isKotlinClass(clazz)) { "Using Java classes with Kotlin's API is not supported" @@ -1529,7 +1618,7 @@ internal fun service(clazz: Class): SVC { val serviceName = ReflectionUtils.extractServiceName(clazz) return ProxySupport.createProxy(clazz) { invocation -> - val request = invocation.captureInvocation(serviceName, null).toRequest() + val request = invocation.captureInvocation(serviceName, null, scope).toRequest() // Last argument is the continuation for suspend functions @Suppress("UNCHECKED_CAST") val continuation = invocation.arguments.last() as Continuation @@ -1542,7 +1631,7 @@ internal fun service(clazz: Class): SVC { } @PublishedApi -internal fun virtualObject(clazz: Class, key: String): SVC { +internal fun virtualObject(clazz: Class, key: String, scope: String? = null): SVC { ReflectionUtils.mustHaveVirtualObjectAnnotation(clazz) require(ReflectionUtils.isKotlinClass(clazz)) { "Using Java classes with Kotlin's API is not supported" @@ -1550,7 +1639,7 @@ internal fun virtualObject(clazz: Class, key: String): SVC { val serviceName = ReflectionUtils.extractServiceName(clazz) return ProxySupport.createProxy(clazz) { invocation -> - val request = invocation.captureInvocation(serviceName, key).toRequest() + val request = invocation.captureInvocation(serviceName, key, scope).toRequest() // Last argument is the continuation for suspend functions @Suppress("UNCHECKED_CAST") val continuation = invocation.arguments.last() as Continuation @@ -1563,7 +1652,7 @@ internal fun virtualObject(clazz: Class, key: String): SVC { } @PublishedApi -internal fun workflow(clazz: Class, key: String): SVC { +internal fun workflow(clazz: Class, key: String, scope: String? = null): SVC { ReflectionUtils.mustHaveWorkflowAnnotation(clazz) require(ReflectionUtils.isKotlinClass(clazz)) { "Using Java classes with Kotlin's API is not supported" @@ -1571,7 +1660,7 @@ internal fun workflow(clazz: Class, key: String): SVC { val serviceName = ReflectionUtils.extractServiceName(clazz) return ProxySupport.createProxy(clazz) { invocation -> - val request = invocation.captureInvocation(serviceName, key).toRequest() + val request = invocation.captureInvocation(serviceName, key, scope).toRequest() // Last argument is the continuation for suspend functions @Suppress("UNCHECKED_CAST") val continuation = invocation.arguments.last() as Continuation diff --git a/sdk-api/src/main/java/dev/restate/sdk/ContextImpl.java b/sdk-api/src/main/java/dev/restate/sdk/ContextImpl.java index e02c26dc3..5190bca48 100644 --- a/sdk-api/src/main/java/dev/restate/sdk/ContextImpl.java +++ b/sdk-api/src/main/java/dev/restate/sdk/ContextImpl.java @@ -137,6 +137,7 @@ public CallDurableFuture call(Request request) { request.getTarget(), input, request.getIdempotencyKey(), + request.getLimitKey(), request.getHeaders() == null ? Collections.emptyList() : request.getHeaders().entrySet())); @@ -169,6 +170,7 @@ public InvocationHandle send( request.getTarget(), input, request.getIdempotencyKey(), + request.getLimitKey(), request.getHeaders() == null ? Collections.emptyList() : request.getHeaders().entrySet(), diff --git a/sdk-api/src/main/java/dev/restate/sdk/Restate.java b/sdk-api/src/main/java/dev/restate/sdk/Restate.java index a3d26c172..8ccbd4f41 100644 --- a/sdk-api/src/main/java/dev/restate/sdk/Restate.java +++ b/sdk-api/src/main/java/dev/restate/sdk/Restate.java @@ -688,6 +688,61 @@ public static ServiceHandle workflowHandle(Class clazz, String k return toWorkflow(clazz, key); } + /** + * PREVIEW: Returns a {@link Scope} that routes all outgoing calls within the given scope. + * + *

NOTE: This API is in preview and is not enabled by default. To use it in + * restate-server 1.7, enable the flow control and protocol v7 experimental features, via {@code + * RESTATE_EXPERIMENTAL_ENABLE_PROTOCOL_V7=true} and {@code + * RESTATE_EXPERIMENTAL_ENABLE_VQUEUES=true}. These can be enabled only on new clusters, + * for more info check out https://docs.restate.dev/services/flow-control#enabling-flow-control. + * If these experimental features aren't enabled, the call fails with a retryable error and keeps + * retrying until they are. + * + *

A scope is a sub-grouping of resources (invocations, virtual object instances, workflow + * instances, concurrency limits) within the Restate cluster. It becomes part of the target + * identity tuple: + * + *

    + *
  • {@code scope, service, handler, idempotencyKey?} + *
  • {@code scope, virtualObject, objectKey, handler, idempotencyKey?} + *
  • {@code scope, workflow, workflowKey, handler} + *
+ * + *

Under the hood, the scope contributes to the partition key, so all resources in a scope get + * co-located by the restate-server. + * + *

Omitting the scope (i.e. using the regular {@link #service(Class)} / {@link #workflow(Class, + * String)} methods) is equivalent to calling with no scope, which is the existing behavior. + * + *

The scope key must consist only of {@code [a-zA-Z0-9_.-]} characters, with {@code 1 <= + * length <= 36} chars. + * + * 

{@code
+   * // Route a call into a named scope
+   * Restate.scope("tenant-123").service(MyService.class).process(payload);
+   *
+   * // Idempotency keys are scoped — "req-1" in "tenant-123" is distinct from "req-1" in "tenant-456"
+   * Restate.scope("tenant-123").serviceHandle(MyService.class)
+   *     .call(MyService::process, payload, InvocationOptions.idempotencyKey("req-1").build())
+   *     .await();
+   *
+   * // Combine with a limit key to enforce per-scope concurrency limits
+   * Restate.scope("tenant-123").workflowHandle(MyWorkflow.class, "wf-key")
+   *     .call(MyWorkflow::run, input, InvocationOptions.limitKey("api-key/user42").build())
+   *     .await();
+   * }
+ * + * @param scopeKey the scope identifier + * @return a {@link Scope} + * @see https://docs.restate.dev/services/flow-control + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public static Scope scope(String scopeKey) { + return new Scope(scopeKey); + } + /** Interface to interact with this Virtual Object/Workflow state. */ public interface State { diff --git a/sdk-api/src/main/java/dev/restate/sdk/Scope.java b/sdk-api/src/main/java/dev/restate/sdk/Scope.java new file mode 100644 index 000000000..dfb401c77 --- /dev/null +++ b/sdk-api/src/main/java/dev/restate/sdk/Scope.java @@ -0,0 +1,132 @@ +// 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 dev.restate.sdk; + +import dev.restate.common.Request; +import dev.restate.common.Target; +import dev.restate.common.reflections.MethodInfo; +import dev.restate.common.reflections.ProxySupport; +import dev.restate.common.reflections.ReflectionUtils; +import dev.restate.serde.TypeTag; + +/** + * PREVIEW: A context for making RPC calls within a specific scope. + * + *

Obtain an instance via {@link Restate#scope(String)}. + * + * @see Restate#scope(String) + */ +@org.jetbrains.annotations.ApiStatus.Experimental +public final class Scope { + + private final String scope; + + Scope(String scope) { + this.scope = scope; + } + + /** + * @see Restate#service(Class) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public SVC service(Class clazz) { + ReflectionUtils.mustHaveServiceAnnotation(clazz); + String serviceName = ReflectionUtils.extractServiceName(clazz); + return ProxySupport.createProxy( + clazz, + invocation -> { + var methodInfo = MethodInfo.fromMethod(invocation.getMethod()); + + //noinspection unchecked + return Context.current() + .call( + Request.of( + Target.virtualObject(scope, serviceName, null, methodInfo.getHandlerName()), + (TypeTag) methodInfo.getInputType(), + (TypeTag) methodInfo.getOutputType(), + invocation.getArguments().length == 0 ? null : invocation.getArguments()[0])) + .await(); + }); + } + + /** + * @see Restate#serviceHandle(Class) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public ServiceHandle serviceHandle(Class clazz) { + ReflectionUtils.mustHaveServiceAnnotation(clazz); + return new ServiceHandleImpl<>(clazz, null, scope); + } + + /** + * @see Restate#virtualObject(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public SVC virtualObject(Class clazz, String key) { + ReflectionUtils.mustHaveVirtualObjectAnnotation(clazz); + String serviceName = ReflectionUtils.extractServiceName(clazz); + return ProxySupport.createProxy( + clazz, + invocation -> { + var methodInfo = MethodInfo.fromMethod(invocation.getMethod()); + + //noinspection unchecked + return Context.current() + .call( + Request.of( + Target.virtualObject(scope, serviceName, key, methodInfo.getHandlerName()), + (TypeTag) methodInfo.getInputType(), + (TypeTag) methodInfo.getOutputType(), + invocation.getArguments().length == 0 ? null : invocation.getArguments()[0])) + .await(); + }); + } + + /** + * @see Restate#virtualObjectHandle(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public ServiceHandle virtualObjectHandle(Class clazz, String key) { + ReflectionUtils.mustHaveVirtualObjectAnnotation(clazz); + return new ServiceHandleImpl<>(clazz, key, scope); + } + + /** + * @see Restate#workflow(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public SVC workflow(Class clazz, String key) { + ReflectionUtils.mustHaveWorkflowAnnotation(clazz); + String serviceName = ReflectionUtils.extractServiceName(clazz); + return ProxySupport.createProxy( + clazz, + invocation -> { + var methodInfo = MethodInfo.fromMethod(invocation.getMethod()); + + //noinspection unchecked + return Context.current() + .call( + Request.of( + Target.virtualObject(scope, serviceName, key, methodInfo.getHandlerName()), + (TypeTag) methodInfo.getInputType(), + (TypeTag) methodInfo.getOutputType(), + invocation.getArguments().length == 0 ? null : invocation.getArguments()[0])) + .await(); + }); + } + + /** + * @see Restate#workflowHandle(Class, String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public ServiceHandle workflowHandle(Class clazz, String key) { + ReflectionUtils.mustHaveWorkflowAnnotation(clazz); + return new ServiceHandleImpl<>(clazz, key, scope); + } +} diff --git a/sdk-api/src/main/java/dev/restate/sdk/ServiceHandleImpl.java b/sdk-api/src/main/java/dev/restate/sdk/ServiceHandleImpl.java index bedfab092..78ab8d291 100644 --- a/sdk-api/src/main/java/dev/restate/sdk/ServiceHandleImpl.java +++ b/sdk-api/src/main/java/dev/restate/sdk/ServiceHandleImpl.java @@ -27,14 +27,20 @@ final class ServiceHandleImpl implements ServiceHandle { private final Class clazz; private final String serviceName; private final @Nullable String key; + private final @Nullable String scope; // To use call/send private MethodInfoCollector methodInfoCollector; ServiceHandleImpl(Class clazz, @Nullable String key) { + this(clazz, key, null); + } + + ServiceHandleImpl(Class clazz, @Nullable String key, @Nullable String scope) { this.clazz = clazz; this.serviceName = ReflectionUtils.extractServiceName(clazz); this.key = key; + this.scope = scope; } @SuppressWarnings("unchecked") @@ -45,6 +51,7 @@ public DurableFuture call( return Context.current() .call( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -62,6 +69,7 @@ public DurableFuture call( return Context.current() .call( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -78,6 +86,7 @@ public DurableFuture call(Function methodReference, InvocationOpt return Context.current() .call( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -93,6 +102,7 @@ public DurableFuture call(Consumer methodReference, InvocationOptions return Context.current() .call( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -110,6 +120,7 @@ public InvocationHandle send( return Context.current() .send( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -128,6 +139,7 @@ public InvocationHandle send( return Context.current() .send( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -146,6 +158,7 @@ public InvocationHandle send( return Context.current() .send( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), @@ -163,6 +176,7 @@ public InvocationHandle send( return Context.current() .send( toRequest( + scope, serviceName, key, methodInfo.getHandlerName(), diff --git a/sdk-common/build.gradle.kts b/sdk-common/build.gradle.kts index 1a878ddc4..8a7f43784 100644 --- a/sdk-common/build.gradle.kts +++ b/sdk-common/build.gradle.kts @@ -12,6 +12,7 @@ description = "Common interfaces of the Restate SDK" dependencies { compileOnly(libs.jspecify) + compileOnly(libs.jetbrains.annotations) api(libs.opentelemetry.api) api(project(":common")) diff --git a/sdk-common/src/main/java/dev/restate/sdk/common/HandlerRequest.java b/sdk-common/src/main/java/dev/restate/sdk/common/HandlerRequest.java index 1a0351dba..12d02609c 100644 --- a/sdk-common/src/main/java/dev/restate/sdk/common/HandlerRequest.java +++ b/sdk-common/src/main/java/dev/restate/sdk/common/HandlerRequest.java @@ -13,6 +13,7 @@ import java.nio.ByteBuffer; import java.util.Map; import java.util.Objects; +import org.jspecify.annotations.Nullable; /** This class encapsulates the inputs to a handler. */ public final class HandlerRequest { @@ -20,6 +21,9 @@ public final class HandlerRequest { private final Context otelContext; private final Slice body; private final Map headers; + private final @Nullable String scope; + private final @Nullable String limitKey; + private final @Nullable String idempotencyKey; private final String serviceName; private final String handlerName; @@ -28,16 +32,27 @@ public HandlerRequest( Context otelContext, Slice body, Map headers, + @Nullable String scope, + @Nullable String limitKey, + @Nullable String idempotencyKey, String serviceName, String handlerName) { this.invocationId = invocationId; this.otelContext = otelContext; this.body = body; this.headers = headers; + this.scope = scope; + this.limitKey = limitKey; + this.idempotencyKey = idempotencyKey; this.serviceName = serviceName; this.handlerName = handlerName; } + public HandlerRequest( + InvocationId invocationId, Context otelContext, Slice body, Map headers) { + this(invocationId, otelContext, body, headers, null, null, null, null, null); + } + public InvocationId invocationId() { return invocationId; } @@ -66,6 +81,28 @@ public Map headers() { return headers; } + /** + * PREVIEW: The scope key with which this invocation was submitted, if any. + * + * @see dev.restate.common.Target#scoped(String) + */ + @org.jetbrains.annotations.ApiStatus.Experimental + public @Nullable String scope() { + return scope; + } + + /** PREVIEW: The limit key with which this invocation was submitted, if any. */ + @org.jetbrains.annotations.ApiStatus.Experimental + public @Nullable String limitKey() { + return limitKey; + } + + /** The idempotency key with which this invocation was submitted, if any. */ + @org.jetbrains.annotations.ApiStatus.Experimental + public @Nullable String idempotencyKey() { + return idempotencyKey; + } + /** Name of the service being invoked. */ public String serviceName() { return serviceName; @@ -77,43 +114,59 @@ public String handlerName() { } @Override - public boolean equals(Object obj) { - if (obj == this) return true; - if (obj == null || obj.getClass() != this.getClass()) return false; - var that = (HandlerRequest) obj; - return Objects.equals(this.invocationId, that.invocationId) - && Objects.equals(this.otelContext, that.otelContext) - && Objects.equals(this.body, that.body) - && Objects.equals(this.headers, that.headers) - && Objects.equals(this.serviceName, that.serviceName) - && Objects.equals(this.handlerName, that.handlerName); + public boolean equals(Object o) { + if (!(o instanceof HandlerRequest that)) return false; + return Objects.equals(invocationId, that.invocationId) + && Objects.equals(otelContext, that.otelContext) + && Objects.equals(body, that.body) + && Objects.equals(headers, that.headers) + && Objects.equals(scope, that.scope) + && Objects.equals(limitKey, that.limitKey) + && Objects.equals(idempotencyKey, that.idempotencyKey) + && Objects.equals(serviceName, that.serviceName) + && Objects.equals(handlerName, that.handlerName); } @Override public int hashCode() { - return Objects.hash(invocationId, otelContext, body, headers, serviceName, handlerName); + return Objects.hash( + invocationId, + otelContext, + body, + headers, + scope, + limitKey, + idempotencyKey, + serviceName, + handlerName); } @Override public String toString() { - return "HandlerRequest[" + return "HandlerRequest{" + "invocationId=" + invocationId - + ", " - + "serviceName=" - + serviceName - + ", " - + "handlerName=" - + handlerName - + ", " - + "otelContext=" + + ", otelContext=" + otelContext - + ", " - + "body=" + + ", body=" + body - + ", " - + "headers=" + + ", headers=" + headers - + ']'; + + ", scope='" + + scope + + '\'' + + ", limitKey='" + + limitKey + + '\'' + + ", idempotencyKey='" + + idempotencyKey + + '\'' + + ", serviceName='" + + serviceName + + '\'' + + ", handlerName='" + + handlerName + + '\'' + + '}'; } } diff --git a/sdk-common/src/main/java/dev/restate/sdk/endpoint/definition/HandlerContext.java b/sdk-common/src/main/java/dev/restate/sdk/endpoint/definition/HandlerContext.java index 9ad11042f..328d2afd4 100644 --- a/sdk-common/src/main/java/dev/restate/sdk/endpoint/definition/HandlerContext.java +++ b/sdk-common/src/main/java/dev/restate/sdk/endpoint/definition/HandlerContext.java @@ -70,16 +70,37 @@ public interface HandlerContext { record CallResult( AsyncResult invocationIdAsyncResult, AsyncResult callAsyncResult) {} + @Deprecated + default CompletableFuture call( + Target target, + Slice parameter, + @Nullable String idempotencyKey, + @Nullable Collection> headers) { + return call(target, parameter, idempotencyKey, null, headers); + } + CompletableFuture call( Target target, Slice parameter, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable Collection> headers); + @Deprecated + default CompletableFuture> send( + Target target, + Slice parameter, + @Nullable String idempotencyKey, + @Nullable Collection> headers, + @Nullable Duration delay) { + return send(target, parameter, idempotencyKey, null, headers, delay); + } + CompletableFuture> send( Target target, Slice parameter, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable Collection> headers, @Nullable Duration delay); diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/ExecutorSwitchingHandlerContextImpl.java b/sdk-core/src/main/java/dev/restate/sdk/core/ExecutorSwitchingHandlerContextImpl.java index 67039ba88..de5fe4ea2 100644 --- a/sdk-core/src/main/java/dev/restate/sdk/core/ExecutorSwitchingHandlerContextImpl.java +++ b/sdk-core/src/main/java/dev/restate/sdk/core/ExecutorSwitchingHandlerContextImpl.java @@ -98,9 +98,10 @@ public CompletableFuture call( Target target, Slice parameter, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable Collection> headers) { return CompletableFuture.supplyAsync( - () -> super.call(target, parameter, idempotencyKey, headers), coreExecutor) + () -> super.call(target, parameter, idempotencyKey, limitKey, headers), coreExecutor) .thenCompose(Function.identity()); } @@ -109,10 +110,12 @@ public CompletableFuture> send( Target target, Slice parameter, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable Collection> headers, @Nullable Duration delay) { return CompletableFuture.supplyAsync( - () -> super.send(target, parameter, idempotencyKey, headers, delay), coreExecutor) + () -> super.send(target, parameter, idempotencyKey, limitKey, headers, delay), + coreExecutor) .thenCompose(Function.identity()); } diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java b/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java index 373ef7a30..4c9840a65 100644 --- a/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java +++ b/sdk-core/src/main/java/dev/restate/sdk/core/HandlerContextImpl.java @@ -66,6 +66,9 @@ class HandlerContextImpl implements HandlerContextInternal { otelContext, input.input(), input.headersAsMap(), + input.scope(), + input.limitKey(), + input.idempotencyKey(), serviceName, handlerName); this.attemptHeaders = attemptHeaders; @@ -225,12 +228,13 @@ public CompletableFuture call( Target target, Slice parameter, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable Collection> headers) { return catchExceptions( () -> { - // scope/limitKey are not yet surfaced by the public API; pass null for now. StateMachine.CallHandle callHandle = - this.stateMachine.call(target, parameter, idempotencyKey, null, null, headers); + this.stateMachine.call( + target, parameter, idempotencyKey, target.getScope(), limitKey, headers); AsyncResultInternal invocationIdAsyncResult = AsyncResults.single(this, callHandle.invocationIdHandle(), invocationIdCompleter()); @@ -248,13 +252,14 @@ public CompletableFuture> send( Target target, Slice parameter, @Nullable String idempotencyKey, + @Nullable String limitKey, @Nullable Collection> headers, @Nullable Duration delay) { return catchExceptions( () -> { - // scope/limitKey are not yet surfaced by the public API; pass null for now. int sendHandle = - this.stateMachine.send(target, parameter, idempotencyKey, null, null, headers, delay); + this.stateMachine.send( + target, parameter, idempotencyKey, target.getScope(), limitKey, headers, delay); return AsyncResults.single(this, sendHandle, invocationIdCompleter()); }); diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/ProtocolException.java b/sdk-core/src/main/java/dev/restate/sdk/core/ProtocolException.java index 172e6b71d..c65653d68 100644 --- a/sdk-core/src/main/java/dev/restate/sdk/core/ProtocolException.java +++ b/sdk-core/src/main/java/dev/restate/sdk/core/ProtocolException.java @@ -160,6 +160,14 @@ public static ProtocolException uncompletedDoProgressDuringReplay( return new ProtocolException(sb.toString(), JOURNAL_MISMATCH_CODE); } + public static ProtocolException scopeOrLimitKeyUnsupported() { + return new ProtocolException( + "Scope and limit key are not supported by the legacy state machine, which negotiates at most" + + " service protocol version V6. These features require the new state machine" + + " and service protocol V7, available on JDK 23+.", + UNSUPPORTED_FEATURE); + } + public static ProtocolException unsupportedFeature( String featureName, Protocol.ServiceProtocolVersion requiredVersion, diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/legacy/LegacyStateMachine.java b/sdk-core/src/main/java/dev/restate/sdk/core/legacy/LegacyStateMachine.java index cfce0ffe0..b94b205a8 100644 --- a/sdk-core/src/main/java/dev/restate/sdk/core/legacy/LegacyStateMachine.java +++ b/sdk-core/src/main/java/dev/restate/sdk/core/legacy/LegacyStateMachine.java @@ -342,6 +342,9 @@ public CallHandle call( @Nullable String limitKey, @Nullable Collection> headers) { LOG.debug("Executing 'Call {}'", target); + if (scope != null || limitKey != null) { + throw ProtocolException.scopeOrLimitKeyUnsupported(); + } if (idempotencyKey != null && idempotencyKey.isBlank()) { throw ProtocolException.idempotencyKeyIsEmpty(); } @@ -401,6 +404,9 @@ public int send( } else { LOG.debug("Executing 'Send {}'", target); } + if (scope != null || limitKey != null) { + throw ProtocolException.scopeOrLimitKeyUnsupported(); + } if (idempotencyKey != null && idempotencyKey.isBlank()) { throw ProtocolException.idempotencyKeyIsEmpty(); } diff --git a/sdk-fake-api/src/main/java/dev/restate/sdk/fake/FakeHandlerContext.java b/sdk-fake-api/src/main/java/dev/restate/sdk/fake/FakeHandlerContext.java index ee525e39f..0e26190c7 100644 --- a/sdk-fake-api/src/main/java/dev/restate/sdk/fake/FakeHandlerContext.java +++ b/sdk-fake-api/src/main/java/dev/restate/sdk/fake/FakeHandlerContext.java @@ -66,6 +66,9 @@ public String toString() { Context.root(), Slice.EMPTY, expectations.requestHeaders(), + null, + null, + null, "FakeService", "fakeHandler"); } @@ -147,7 +150,11 @@ public CompletableFuture> timer(Duration duration, String s) { @Override public CompletableFuture call( - Target target, Slice slice, String s, Collection> collection) { + Target target, + Slice slice, + String s, + String limitKey, + Collection> collection) { throw new UnsupportedOperationException( "FakeHandlerContext doesn't currently support mocking this operation"); } @@ -157,6 +164,7 @@ public CompletableFuture> send( Target target, Slice slice, String s, + String limitKey, Collection> collection, Duration duration) { throw new UnsupportedOperationException( From e7f3fdb997aaedd4ae895c6672ea3fb65cb8adf6 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Tue, 7 Jul 2026 11:25:58 +0200 Subject: [PATCH 2/2] Scope and Limit Key tests --- .github/workflows/integration-ci.yaml | 5 +++- .github/workflows/integration.yaml | 5 +++- .tools/exclusions-legacy-statemachine.yaml | 9 +++++++ .../dev/restate/sdk/testservices/ProxyImpl.kt | 24 +++++++++++++++---- .../sdk/testservices/contracts/Proxy.kt | 2 ++ 5 files changed, 38 insertions(+), 7 deletions(-) create mode 100644 .tools/exclusions-legacy-statemachine.yaml diff --git a/.github/workflows/integration-ci.yaml b/.github/workflows/integration-ci.yaml index 1dafe7df6..1734b9ff6 100644 --- a/.github/workflows/integration-ci.yaml +++ b/.github/workflows/integration-ci.yaml @@ -107,9 +107,12 @@ jobs: - name: Run test tool continue-on-error: ${{ inputs.continueOnError == 'true' }} - uses: restatedev/e2e/sdk-tests@v2.1 + uses: restatedev/e2e/sdk-tests@v2.2 with: envVars: ${{ inputs.envVars }} + # Scope/limit key need the FFM state machine (protocol v7, JRE >= 23). On the legacy + # state machine (JRE < 23) those tests can't pass, so exclude them there. + exclusionsFile: ${{ matrix.jreVersion < 23 && '.tools/exclusions-legacy-statemachine.yaml' || '' }} testArtifactOutput: ${{ inputs.testArtifactOutput != '' && format('{0}-jre{1}', inputs.testArtifactOutput, matrix.jreVersion) || format('sdk-java-jre{0}-integration-test-report', matrix.jreVersion) }} restateContainerImage: ${{ inputs.restateCommit != '' && 'localhost/restatedev/restate-commit-download:latest' || (inputs.restateImage != '' && inputs.restateImage || 'ghcr.io/restatedev/restate:main') }} serviceContainerImage: ${{ inputs.serviceImage != '' && inputs.serviceImage || 'restatedev/test-services-java' }} diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml index fec515199..62ac1b74d 100644 --- a/.github/workflows/integration.yaml +++ b/.github/workflows/integration.yaml @@ -117,9 +117,12 @@ jobs: - name: Run test tool continue-on-error: ${{ inputs.continueOnError == 'true' }} - uses: restatedev/e2e/sdk-tests@v2.1 + uses: restatedev/e2e/sdk-tests@v2.2 with: envVars: ${{ inputs.envVars }} + # Scope/limit key need the FFM state machine (protocol v7, JRE >= 23). On the legacy + # state machine (JRE < 23) those tests can't pass, so exclude them there. + exclusionsFile: ${{ matrix.jreVersion < 23 && '.tools/exclusions-legacy-statemachine.yaml' || '' }} testArtifactOutput: ${{ inputs.testArtifactOutput != '' && format('{0}-jre{1}', inputs.testArtifactOutput, matrix.jreVersion) || format('sdk-java-jre{0}-integration-test-report', matrix.jreVersion) }} restateContainerImage: ${{ inputs.restateCommit != '' && 'localhost/restatedev/restate-commit-download:latest' || (inputs.restateImage != '' && inputs.restateImage || 'ghcr.io/restatedev/restate:main') }} serviceContainerImage: ${{ inputs.serviceImage != '' && inputs.serviceImage || 'restatedev/test-services-java' }} diff --git a/.tools/exclusions-legacy-statemachine.yaml b/.tools/exclusions-legacy-statemachine.yaml new file mode 100644 index 000000000..879add01f --- /dev/null +++ b/.tools/exclusions-legacy-statemachine.yaml @@ -0,0 +1,9 @@ +# Tests excluded when the SDK runs on the legacy (pure-Java) state machine, i.e. on JRE < 23, +# where the Panama/FFM state machine is unavailable and only service protocol v6 is negotiated. +# +# Scope and limit key require service protocol v7 (native/FFM state machine, JRE >= 23), so these +# tests must be skipped on the legacy state machine. This file is passed via `--exclusions-file` +# only for the JRE 17 and 21 CI legs; the JRE 25 leg runs them. +exclusions: + "default": + - "dev.restate.sdktesting.tests.ServiceToServiceScopeConcurrency.scopeAndLimitKeyArePropagatedOnServiceToServiceCalls" diff --git a/test-services/src/main/kotlin/dev/restate/sdk/testservices/ProxyImpl.kt b/test-services/src/main/kotlin/dev/restate/sdk/testservices/ProxyImpl.kt index ba2795c6f..f9b01cd81 100644 --- a/test-services/src/main/kotlin/dev/restate/sdk/testservices/ProxyImpl.kt +++ b/test-services/src/main/kotlin/dev/restate/sdk/testservices/ProxyImpl.kt @@ -18,11 +18,13 @@ import kotlin.time.Duration.Companion.milliseconds class ProxyImpl : Proxy { private fun Proxy.ProxyRequest.toTarget(): Target { - return if (this.virtualObjectKey == null) { - Target.service(this.serviceName, this.handlerName) - } else { - Target.virtualObject(this.serviceName, this.virtualObjectKey, this.handlerName) - } + val target = + if (this.virtualObjectKey == null) { + Target.service(this.serviceName, this.handlerName) + } else { + Target.virtualObject(this.serviceName, this.virtualObjectKey, this.handlerName) + } + return if (this.scope != null) target.scoped(this.scope) else target } override suspend fun call(request: Proxy.ProxyRequest): ByteArray { @@ -32,6 +34,9 @@ class ProxyImpl : Proxy { if (request.idempotencyKey != null) { it.idempotencyKey = request.idempotencyKey } + if (request.limitKey != null) { + it.limitKey = request.limitKey + } } ) .await() @@ -44,6 +49,9 @@ class ProxyImpl : Proxy { if (request.idempotencyKey != null) { it.idempotencyKey = request.idempotencyKey } + if (request.limitKey != null) { + it.limitKey = request.limitKey + } }, request.delayMillis?.milliseconds ?: Duration.ZERO, ) @@ -66,6 +74,9 @@ class ProxyImpl : Proxy { if (request.proxyRequest.idempotencyKey != null) { it.idempotencyKey = request.proxyRequest.idempotencyKey } + if (request.proxyRequest.limitKey != null) { + it.limitKey = request.proxyRequest.limitKey + } }, request.proxyRequest.delayMillis?.milliseconds ?: Duration.ZERO, ) @@ -83,6 +94,9 @@ class ProxyImpl : Proxy { if (request.proxyRequest.idempotencyKey != null) { it.idempotencyKey = request.proxyRequest.idempotencyKey } + if (request.proxyRequest.limitKey != null) { + it.limitKey = request.proxyRequest.limitKey + } } ) if (request.awaitAtTheEnd) { diff --git a/test-services/src/main/kotlin/dev/restate/sdk/testservices/contracts/Proxy.kt b/test-services/src/main/kotlin/dev/restate/sdk/testservices/contracts/Proxy.kt index c7392cb55..48b6fefb1 100644 --- a/test-services/src/main/kotlin/dev/restate/sdk/testservices/contracts/Proxy.kt +++ b/test-services/src/main/kotlin/dev/restate/sdk/testservices/contracts/Proxy.kt @@ -24,6 +24,8 @@ interface Proxy { val message: ByteArray, val delayMillis: Int? = null, val idempotencyKey: String? = null, + val scope: String? = null, + val limitKey: String? = null, ) @Serializable