Standalone operations for Nexus (#2872)

Author: Evanthx

.github/workflows/ci.yml

@@ -114,6 +114,7 @@ jobs:
               --dynamic-config-value 'component.callbacks.allowedAddresses=[{"Pattern":"localhost:7243","AllowInsecure":true}]' \
               --dynamic-config-value frontend.activityAPIsEnabled=true \
               --dynamic-config-value activity.enableStandalone=true \
+              --dynamic-config-value nexusoperation.enableStandalone=true \
               --dynamic-config-value history.enableChasm=true \
               --dynamic-config-value history.enableTransitionHistory=true &
           sleep 10s

temporal-sdk/src/main/java/io/temporal/client/NexusClient.java

@@ -0,0 +1,153 @@
+package io.temporal.client;
+
+import io.temporal.common.Experimental;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import java.lang.reflect.Type;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+
+/**
+ * Client for managing standalone Nexus operation executions. Obtain an instance via {@link
+ * #newInstance(WorkflowServiceStubs)} or {@link #newInstance(WorkflowServiceStubs,
+ * NexusClientOptions)}. Do not create this object per request; share it for the lifetime of the
+ * process.
+ *
+ * <p>Standalone Nexus operations run independently of any workflow — they are scheduled, monitored,
+ * and managed directly through this client (and the service-bound clients it produces) rather than
+ * from within a workflow execution.
+ *
+ * <p>To start operations, build a service-bound client and call {@code start}/{@code execute}:
+ *
+ * <pre>{@code
+ * NexusClient client = NexusClient.newInstance(stubs, options);
+ *
+ * // Typed: bind to an @ServiceInterface and invoke a method reference.
+ * NexusServiceClient<MyService> svc =
+ *     client.newNexusServiceClient(MyService.class, "my-endpoint");
+ * String result = svc.execute(MyService::greet, "world");
+ *
+ * // Untyped: dispatch by operation name string.
+ * UntypedNexusServiceClient untyped =
+ *     client.newUntypedNexusServiceClient("my-endpoint", "MyService");
+ * UntypedNexusOperationHandle handle = untyped.start("greet", null, "world");
+ * }</pre>
+ *
+ * <p>To act on an existing operation (describe, cancel, terminate, get result), obtain a handle via
+ * {@link #getHandle}:
+ *
+ * <pre>{@code
+ * NexusOperationHandle<String> handle = client.getHandle(operationId, runId, String.class);
+ * String result = handle.getResult();
+ * handle.cancel("user requested");
+ * }</pre>
+ *
+ * <p>For visibility queries across all operations in the namespace, see {@link
+ * #listNexusOperationExecutions} and {@link #countNexusOperationExecutions}.
+ *
+ * @see NexusServiceClient
+ * @see UntypedNexusServiceClient
+ * @see NexusOperationHandle
+ */
+@Experimental
+public interface NexusClient {
+
+  /**
+   * Creates a client with default {@link NexusClientOptions}.
+   *
+   * @param service gRPC stubs connected to a Temporal Service endpoint
+   */
+  static NexusClient newInstance(WorkflowServiceStubs service) {
+    return NexusClientImpl.newInstance(service, NexusClientOptions.getDefaultInstance());
+  }
+
+  /**
+   * Creates a client with the supplied options.
+   *
+   * @param service gRPC stubs connected to a Temporal Service endpoint
+   * @param options namespace, data converter, interceptors, and defaults applied to operations
+   *     started through this client
+   */
+  static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
+    return NexusClientImpl.newInstance(service, options);
+  }
+
+  /** Returns the underlying gRPC stubs this client routes RPCs through. */
+  WorkflowServiceStubs getWorkflowServiceStubs();
+
+  /**
+   * Returns an untyped handle to an existing operation execution, optionally pinned to a specific
+   * run.
+   *
+   * @param operationId the user-assigned operation ID
+   * @param runId the server-assigned run ID, or {@code null} to target the latest run
+   * @return an untyped handle
+   */
+  UntypedNexusOperationHandle getHandle(String operationId, @Nullable String runId);
+
+  /**
+   * Returns a typed handle to an existing operation execution, bound to {@code resultClass}.
+   *
+   * @param operationId the user-assigned operation ID
+   * @param runId the server-assigned run ID, or {@code null} to target the latest run
+   * @param resultClass expected result type
+   * @param <R> result type
+   */
+  <R> NexusOperationHandle<R> getHandle(
+      String operationId, @Nullable String runId, Class<R> resultClass);
+
+  /**
+   * Returns a typed handle to an existing operation execution, bound to {@code resultClass}/{@code
+   * resultType}. Use the {@code resultType} variant when the result is a generic type whose
+   * parameters cannot be captured by {@link Class} alone (e.g. {@code List<String>}).
+   *
+   * @param operationId the user-assigned operation ID
+   * @param runId the server-assigned run ID, or {@code null} to target the latest run
+   * @param resultClass expected result class
+   * @param resultType generic type for deserialization; may be {@code null}
+   * @param <R> result type
+   */
+  <R> NexusOperationHandle<R> getHandle(
+      String operationId, @Nullable String runId, Class<R> resultClass, @Nullable Type resultType);
+
+  /**
+   * Builds a typed service-bound client targeting the given endpoint, dispatching operations by
+   * method reference on the {@code @ServiceInterface}-annotated {@code service}. Reuses this
+   * client's stubs, options, and interceptor chain.
+   *
+   * @param service the {@code @ServiceInterface}-annotated service type
+   * @param endpoint Nexus endpoint name registered on the Temporal Service
+   * @param <T> the service interface type
+   */
+  <T> NexusServiceClient<T> newNexusServiceClient(Class<T> service, String endpoint);
+
+  /**
+   * Builds an untyped service-bound client targeting the given endpoint and service. Use this to
+   * dispatch operations by name string when no service interface is available.
+   *
+   * @param endpoint Nexus endpoint name registered on the Temporal Service
+   * @param serviceName Nexus service name on that endpoint
+   */
+  UntypedNexusServiceClient newUntypedNexusServiceClient(String endpoint, String serviceName);
+
+  /**
+   * Returns a stream of standalone Nexus operation executions matching the given visibility query.
+   * The stream paginates lazily over server-side results — pages are fetched on demand as the
+   * stream is consumed.
+   *
+   * @param query Temporal visibility query string, or {@code null} to return all executions in the
+   *     client namespace
+   * @return a lazy stream of matching executions
+   */
+  Stream<NexusOperationExecutionMetadata> listNexusOperationExecutions(@Nullable String query);
+
+  /**
+   * Returns the count of standalone Nexus operation executions matching the given visibility query,
+   * optionally with aggregation groups.
+   *
+   * @param query Temporal visibility query string, or {@code null} to count all executions in the
+   *     client namespace
+   * @return execution count, optionally with aggregation groups when the query uses {@code GROUP
+   *     BY}
+   */
+  NexusOperationExecutionCount countNexusOperationExecutions(@Nullable String query);
+}

temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java

@@ -0,0 +1,140 @@
+package io.temporal.client;
+
+import static io.temporal.internal.WorkflowThreadMarker.enforceNonWorkflowThread;
+
+import com.uber.m3.tally.Scope;
+import io.temporal.common.Experimental;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.CountNexusOperationExecutionsInput;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.CountNexusOperationExecutionsOutput;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.ListNexusOperationExecutionsInput;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.ListNexusOperationExecutionsOutput;
+import io.temporal.common.interceptors.NexusClientInterceptor;
+import io.temporal.internal.WorkflowThreadMarker;
+import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
+import io.temporal.internal.client.NexusOperationHandleImpl;
+import io.temporal.internal.client.RootNexusClientInvoker;
+import io.temporal.internal.client.external.GenericWorkflowClient;
+import io.temporal.internal.client.external.GenericWorkflowClientImpl;
+import io.temporal.serviceclient.MetricsTag;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import java.util.List;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Experimental
+public class NexusClientImpl implements NexusClient {
+
+  private static final Logger log = LoggerFactory.getLogger(NexusClientImpl.class);
+
+  private final WorkflowServiceStubs workflowServiceStubs;
+  private final NexusClientOptions options;
+  private final GenericWorkflowClient genericClient;
+  private final Scope metricsScope;
+  private final NexusClientCallsInterceptor nexusClientCallsInvoker;
+  private final List<NexusClientInterceptor> interceptors;
+
+  public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
+    enforceNonWorkflowThread();
+    return WorkflowThreadMarker.protectFromWorkflowThread(
+        new NexusClientImpl(service, options), NexusClient.class);
+  }
+
+  NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientOptions options) {
+    workflowServiceStubs =
+        new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace());
+    this.workflowServiceStubs = workflowServiceStubs;
+    this.options = options;
+    this.metricsScope =
+        workflowServiceStubs
+            .getOptions()
+            .getMetricsScope()
+            .tagged(MetricsTag.defaultTags(options.getNamespace()));
+    this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
+    this.interceptors = options.getInterceptors();
+    this.nexusClientCallsInvoker = initializeClientInvoker();
+    if (log.isDebugEnabled()) {
+      log.debug(
+          "NexusClient initialized: namespace={}, interceptors={}",
+          options.getNamespace(),
+          interceptors.size());
+    }
+  }
+
+  private NexusClientCallsInterceptor initializeClientInvoker() {
+    NexusClientCallsInterceptor invoker = new RootNexusClientInvoker(genericClient, options);
+    for (NexusClientInterceptor clientInterceptor : interceptors) {
+      NexusClientCallsInterceptor wrapped = clientInterceptor.nexusClientCallsInterceptor(invoker);
+      if (wrapped == null) {
+        throw new IllegalStateException(
+            "NexusClientInterceptor "
+                + clientInterceptor.getClass().getName()
+                + " returned null from nexusClientCallsInterceptor; expected a non-null"
+                + " NexusClientCallsInterceptor wrapping the supplied next link");
+      }
+      invoker = wrapped;
+    }
+    return invoker;
+  }
+
+  @Override
+  public WorkflowServiceStubs getWorkflowServiceStubs() {
+    return workflowServiceStubs;
+  }
+
+  @Override
+  public UntypedNexusOperationHandle getHandle(String operationId, @Nullable String runId) {
+    return new NexusOperationHandleImpl(operationId, runId, nexusClientCallsInvoker);
+  }
+
+  @Override
+  public <R> NexusOperationHandle<R> getHandle(
+      String operationId, @Nullable String runId, Class<R> resultClass) {
+    return getHandle(operationId, runId, resultClass, null);
+  }
+
+  @Override
+  public <R> NexusOperationHandle<R> getHandle(
+      String operationId,
+      @Nullable String runId,
+      Class<R> resultClass,
+      @Nullable java.lang.reflect.Type resultType) {
+    return NexusOperationHandle.fromUntyped(getHandle(operationId, runId), resultClass, resultType);
+  }
+
+  @Override
+  public <T> NexusServiceClient<T> newNexusServiceClient(Class<T> service, String endpoint) {
+    enforceNonWorkflowThread();
+    return WorkflowThreadMarker.protectFromWorkflowThread(
+        new NexusServiceClientImpl<>(nexusClientCallsInvoker, service, endpoint, options),
+        NexusServiceClient.class);
+  }
+
+  @Override
+  public UntypedNexusServiceClient newUntypedNexusServiceClient(
+      String endpoint, String serviceName) {
+    enforceNonWorkflowThread();
+    return WorkflowThreadMarker.protectFromWorkflowThread(
+        new UntypedNexusServiceClientImpl(nexusClientCallsInvoker, endpoint, serviceName, options),
+        UntypedNexusServiceClient.class);
+  }
+
+  @Override
+  public Stream<NexusOperationExecutionMetadata> listNexusOperationExecutions(
+      @Nullable String query) {
+    ListNexusOperationExecutionsOutput out =
+        nexusClientCallsInvoker.listNexusOperationExecutions(
+            new ListNexusOperationExecutionsInput(query));
+    return out.getOperations();
+  }
+
+  @Override
+  public NexusOperationExecutionCount countNexusOperationExecutions(@Nullable String query) {
+    CountNexusOperationExecutionsOutput out =
+        nexusClientCallsInvoker.countNexusOperationExecutions(
+            new CountNexusOperationExecutionsInput(query));
+    return out.getCount();
+  }
+}