#347

Committed 31 Jul 2026 06:59PM UTC coverage: 68.148% (-0.03%) from 68.181%

Build # #347

Build Type

push

github

Committed by

web-flow

Commit Message

NEXUS-485: Support Workflow Update as a Nexus Operation (#2945)

* NEXUS-485: Support Workflow Update as a Nexus Operation

* address comments, change signatures to newer

* address comments 2: add all workflow exec overloads

* address comments: log failed

Coverage Stats

7166 of 12554 branches covered (57.08%)

Branch coverage included in aggregate %.

118 of 296 new or added lines in 11 files covered. (39.86%)

28 existing lines in 7 files now uncovered.

29722 of 41575 relevant lines covered (71.49%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.33

/temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java

package io.temporal.internal.common;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.base.Defaults;
import com.google.common.base.Strings;
import io.nexusrpc.Header;
import io.nexusrpc.handler.HandlerException;
import io.nexusrpc.handler.ServiceImplInstance;
import io.temporal.api.common.v1.Callback;
import io.temporal.api.common.v1.Link;
import io.temporal.api.enums.v1.TaskQueueKind;
import io.temporal.api.taskqueue.v1.TaskQueue;
import io.temporal.client.OnConflictOptions;
import io.temporal.client.WorkflowOptions;
import io.temporal.client.WorkflowStub;
import io.temporal.common.metadata.POJOActivityMethodMetadata;
import io.temporal.common.metadata.POJOWorkflowMethodMetadata;
import io.temporal.common.metadata.WorkflowMethodType;
import io.temporal.internal.client.NexusStartWorkflowRequest;
import io.temporal.internal.nexus.CurrentNexusOperationContext;
import io.temporal.internal.nexus.InternalNexusOperationContext;
import io.temporal.internal.nexus.OperationTokenUtil;
import java.util.*;
import java.util.stream.Collectors;

/** Utility functions shared by the implementation code. */
public final class InternalUtils {
  public static final String TEMPORAL_RESERVED_PREFIX = "__temporal_";
  public static final String WORKFLOW_STREAM_RESERVED_PREFIX = "__temporal_workflow_stream_";

  private static String QUERY_TYPE_STACK_TRACE = "__stack_trace";
  private static String ENHANCED_QUERY_TYPE_STACK_TRACE = "__enhanced_stack_trace";

  public static TaskQueue createStickyTaskQueue(
      String stickyTaskQueueName, String normalTaskQueueName) {
    return TaskQueue.newBuilder()
        .setName(stickyTaskQueueName)
        .setKind(TaskQueueKind.TASK_QUEUE_KIND_STICKY)
        .setNormalName(normalTaskQueueName)
        .build();
  }

  public static TaskQueue createNormalTaskQueue(String taskQueueName) {
    return TaskQueue.newBuilder()
        .setName(taskQueueName)
        .setKind(TaskQueueKind.TASK_QUEUE_KIND_NORMAL)
        .build();
  }

  public static Object getValueOrDefault(Object value, Class<?> valueClass) {
    if (value != null) {
      return value;
    }
    return Defaults.defaultValue(valueClass);
  }

  /**
   * Creates a new stub that is bound to the same workflow as the given stub, but with the Nexus
   * callback URL and headers set.
   *
   * @param stub the stub to create a new stub from
   * @param request the request containing the Nexus callback URL and headers
   * @return a new stub bound to the same workflow as the given stub, but with the Nexus callback
   *     URL and headers set
   */
  @SuppressWarnings("deprecation") // Check the OPERATION_ID header for backwards compatibility
  public static NexusWorkflowStarter createNexusBoundStub(
      WorkflowStub stub, NexusStartWorkflowRequest request) {
    if (!stub.getOptions().isPresent()) {
      throw new IllegalArgumentException("Options are expected to be set on the stub");
    }
    WorkflowOptions options = stub.getOptions().get();
    if (options.getWorkflowId() == null) {
      throw new IllegalArgumentException(
          "WorkflowId is expected to be set on WorkflowOptions when used with Nexus");
    }
    InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get();
    // Generate the operation token for the new workflow.
    String operationToken;
    try {
      operationToken =
          OperationTokenUtil.generateWorkflowRunOperationToken(
              options.getWorkflowId(), nexusContext.getNamespace());
    } catch (JsonProcessingException e) {
      // Not expected as the link is constructed by the SDK.
      throw new HandlerException(
          HandlerException.ErrorType.BAD_REQUEST, "failed to generate workflow operation token", e);
    }
    List<Link> links =
        request.getLinks() == null
            ? null
            : request.getLinks().stream()
                .map(
                    (link) ->
                        LinkConverter.nexusLinkToLink(
                            io.temporal.api.nexus.v1.Link.newBuilder()
                                .setType(link.getType())
                                .setUrl(link.getUri().toString())
                                .build()))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    WorkflowOptions.Builder nexusWorkflowOptions =
        WorkflowOptions.newBuilder(options).setRequestId(request.getRequestId()).setLinks(links);

    // If a callback URL is provided, pass it as a completion callback.
    if (!Strings.isNullOrEmpty(request.getCallbackUrl())) {
      // Add the Nexus operation ID to the headers if it is not already present to support
      // fabricating
      // a NexusOperationStarted event if the completion is received before the response to a
      // StartOperation request.
      Map<String, String> headers =
          request.getCallbackHeaders().entrySet().stream()
              .collect(
                  Collectors.toMap(
                      (k) -> k.getKey().toLowerCase(),
                      Map.Entry::getValue,
                      (a, b) -> a,
                      () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER)));
      if (!headers.containsKey(Header.OPERATION_ID)) {
        headers.put(Header.OPERATION_ID.toLowerCase(), operationToken);
      }
      if (!headers.containsKey(Header.OPERATION_TOKEN)) {
        headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);
      }
      Callback.Builder cbBuilder =
          Callback.newBuilder()
              .setNexus(
                  Callback.Nexus.newBuilder()
                      .setUrl(request.getCallbackUrl())
                      .putAllHeader(headers)
                      .build());
      if (links != null) {
        cbBuilder.addAllLinks(links);
      }
      nexusWorkflowOptions.setCompletionCallbacks(Collections.singletonList(cbBuilder.build()));
    }

    if (options.getTaskQueue() == null) {
      nexusWorkflowOptions.setTaskQueue(request.getTaskQueue());
    }
    nexusWorkflowOptions.setOnConflictOptions(
        OnConflictOptions.newBuilder()
            .setAttachRequestId(true)
            .setAttachLinks(true)
            .setAttachCompletionCallbacks(true)
            .build());

    return new NexusWorkflowStarter(stub.newInstance(nexusWorkflowOptions.build()), operationToken);
  }

  /**
   * Returns true if the given name is in the {@code __temporal_workflow_stream_} sub-namespace,
   * which is reserved for the workflow streams contrib module and permitted for signal, update, and
   * query handler registration.
   */
  public static boolean isWorkflowStreamReservedName(String name) {
    return name.startsWith(WORKFLOW_STREAM_RESERVED_PREFIX);
  }

  /** Helper to build a Nexus Callback from the provided input. */
  public static Callback buildNexusCallback(
      Map<String, String> callbackHeaders,
      String callbackUrl,
      String operationToken,
      List<Link> links) {
    Map<String, String> headers =
        callbackHeaders.entrySet().stream()
            .collect(
                Collectors.toMap(
                    (k) -> k.getKey().toLowerCase(),
                    Map.Entry::getValue,
                    (a, b) -> a,
                    TreeMap::new));
    headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);
    Callback.Builder cbBuilder =
        Callback.newBuilder()
            .setNexus(
                Callback.Nexus.newBuilder().setUrl(callbackUrl).putAllHeader(headers).build());
    if (links != null) {
      cbBuilder.addAllLinks(links);
    }
    return cbBuilder.build();
  }

  /** Check the method name for reserved prefixes or names. */
  public static void checkMethodName(POJOWorkflowMethodMetadata methodMetadata) {
    boolean workflowStreamExempt =
        !methodMetadata.getType().equals(WorkflowMethodType.WORKFLOW)
            && isWorkflowStreamReservedName(methodMetadata.getName());
    if (methodMetadata.getName().startsWith(TEMPORAL_RESERVED_PREFIX) && !workflowStreamExempt) {
      throw new IllegalArgumentException(
          methodMetadata.getType().toString().toLowerCase()
              + " name \""
              + methodMetadata.getName()
              + "\" must not start with \""
              + TEMPORAL_RESERVED_PREFIX
              + "\"");
    }
    if (methodMetadata.getType().equals(WorkflowMethodType.QUERY)
        && (methodMetadata.getName().equals(QUERY_TYPE_STACK_TRACE)
            || methodMetadata.getName().equals(ENHANCED_QUERY_TYPE_STACK_TRACE))) {
      throw new IllegalArgumentException(
          "Query method name \"" + methodMetadata.getName() + "\" is reserved for internal use");
    }
  }

  public static void checkMethodName(POJOActivityMethodMetadata methodMetadata) {
    if (methodMetadata.getActivityTypeName().startsWith(TEMPORAL_RESERVED_PREFIX)) {
      throw new IllegalArgumentException(
          "Activity name \""
              + methodMetadata.getActivityTypeName()
              + "\" must not start with \""
              + TEMPORAL_RESERVED_PREFIX
              + "\"");
    }
  }

  /** Prohibit instantiation */
  private InternalUtils() {}

  public static void checkMethodName(ServiceImplInstance instance) {
    if (instance.getDefinition().getName().startsWith(TEMPORAL_RESERVED_PREFIX)) {
      throw new IllegalArgumentException(
          "Service name \""
              + instance.getDefinition().getName()
              + "\" must not start with \""
              + TEMPORAL_RESERVED_PREFIX
              + "\"");
    }
    for (String operationName : instance.getDefinition().getOperations().keySet()) {
      if (operationName.startsWith(TEMPORAL_RESERVED_PREFIX)) {
        throw new IllegalArgumentException(
            "Operation name \""
                + operationName
                + "\" must not start with \""
                + TEMPORAL_RESERVED_PREFIX
                + "\"");
      }
    }
  }
}

1	package io.temporal.internal.common;
2
3	import com.fasterxml.jackson.core.JsonProcessingException;
4	import com.google.common.base.Defaults;
5	import com.google.common.base.Strings;
6	import io.nexusrpc.Header;
7	import io.nexusrpc.handler.HandlerException;
8	import io.nexusrpc.handler.ServiceImplInstance;
9	import io.temporal.api.common.v1.Callback;
10	import io.temporal.api.common.v1.Link;
11	import io.temporal.api.enums.v1.TaskQueueKind;
12	import io.temporal.api.taskqueue.v1.TaskQueue;
13	import io.temporal.client.OnConflictOptions;
14	import io.temporal.client.WorkflowOptions;
15	import io.temporal.client.WorkflowStub;
16	import io.temporal.common.metadata.POJOActivityMethodMetadata;
17	import io.temporal.common.metadata.POJOWorkflowMethodMetadata;
18	import io.temporal.common.metadata.WorkflowMethodType;
19	import io.temporal.internal.client.NexusStartWorkflowRequest;
20	import io.temporal.internal.nexus.CurrentNexusOperationContext;
21	import io.temporal.internal.nexus.InternalNexusOperationContext;
22	import io.temporal.internal.nexus.OperationTokenUtil;
23	import java.util.*;
24	import java.util.stream.Collectors;
25
26	/** Utility functions shared by the implementation code. */
27	public final class InternalUtils {
28	public static final String TEMPORAL_RESERVED_PREFIX = "__temporal_";
29	public static final String WORKFLOW_STREAM_RESERVED_PREFIX = "__temporal_workflow_stream_";
30
31	private static String QUERY_TYPE_STACK_TRACE = "__stack_trace";	1✔
32	private static String ENHANCED_QUERY_TYPE_STACK_TRACE = "__enhanced_stack_trace";	1✔
33
34	public static TaskQueue createStickyTaskQueue(
35	String stickyTaskQueueName, String normalTaskQueueName) {
36	return TaskQueue.newBuilder()	1✔
37	.setName(stickyTaskQueueName)	1✔
38	.setKind(TaskQueueKind.TASK_QUEUE_KIND_STICKY)	1✔
39	.setNormalName(normalTaskQueueName)	1✔
40	.build();	1✔
41	}
42
43	public static TaskQueue createNormalTaskQueue(String taskQueueName) {
44	return TaskQueue.newBuilder()	1✔
45	.setName(taskQueueName)	1✔
46	.setKind(TaskQueueKind.TASK_QUEUE_KIND_NORMAL)	1✔
47	.build();	1✔
48	}
49
50	public static Object getValueOrDefault(Object value, Class<?> valueClass) {
51	if (value != null) {	1✔
52	return value;	1✔
53	}
54	return Defaults.defaultValue(valueClass);	1✔
55	}
56
57	/**
58	* Creates a new stub that is bound to the same workflow as the given stub, but with the Nexus
59	* callback URL and headers set.
60	*
61	* @param stub the stub to create a new stub from
62	* @param request the request containing the Nexus callback URL and headers
63	* @return a new stub bound to the same workflow as the given stub, but with the Nexus callback
64	* URL and headers set
65	*/
66	@SuppressWarnings("deprecation") // Check the OPERATION_ID header for backwards compatibility
67	public static NexusWorkflowStarter createNexusBoundStub(
68	WorkflowStub stub, NexusStartWorkflowRequest request) {
69	if (!stub.getOptions().isPresent()) {	1!
70	throw new IllegalArgumentException("Options are expected to be set on the stub");	×
71	}
72	WorkflowOptions options = stub.getOptions().get();	1✔
73	if (options.getWorkflowId() == null) {	1!
74	throw new IllegalArgumentException(	×
75	"WorkflowId is expected to be set on WorkflowOptions when used with Nexus");
76	}
77	InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get();	1✔
78	// Generate the operation token for the new workflow.
79	String operationToken;
80	try {
81	operationToken =	1✔
82	OperationTokenUtil.generateWorkflowRunOperationToken(	1✔
83	options.getWorkflowId(), nexusContext.getNamespace());	1✔
84	} catch (JsonProcessingException e) {	×
85	// Not expected as the link is constructed by the SDK.
86	throw new HandlerException(	×
87	HandlerException.ErrorType.BAD_REQUEST, "failed to generate workflow operation token", e);
88	}	1✔
89	List<Link> links =
90	request.getLinks() == null	1!
91	? null	×
92	: request.getLinks().stream()	1✔
93	.map(	1✔
94	(link) ->
95	LinkConverter.nexusLinkToLink(	1✔
96	io.temporal.api.nexus.v1.Link.newBuilder()	1✔
97	.setType(link.getType())	1✔
98	.setUrl(link.getUri().toString())	1✔
99	.build()))	1✔
100	.filter(Objects::nonNull)	1✔
101	.collect(Collectors.toList());	1✔
102	WorkflowOptions.Builder nexusWorkflowOptions =	1✔
103	WorkflowOptions.newBuilder(options).setRequestId(request.getRequestId()).setLinks(links);	1✔
104
105	// If a callback URL is provided, pass it as a completion callback.
106	if (!Strings.isNullOrEmpty(request.getCallbackUrl())) {	1!
107	// Add the Nexus operation ID to the headers if it is not already present to support
108	// fabricating
109	// a NexusOperationStarted event if the completion is received before the response to a
110	// StartOperation request.
111	Map<String, String> headers =	1✔
112	request.getCallbackHeaders().entrySet().stream()	1✔
113	.collect(	1✔
114	Collectors.toMap(	1✔
115	(k) -> k.getKey().toLowerCase(),	1✔
116	Map.Entry::getValue,
117	(a, b) -> a,	×
118	() -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER)));	1✔
119	if (!headers.containsKey(Header.OPERATION_ID)) {	1!
120	headers.put(Header.OPERATION_ID.toLowerCase(), operationToken);	1✔
121	}
122	if (!headers.containsKey(Header.OPERATION_TOKEN)) {	1!
123	headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);	1✔
124	}
125	Callback.Builder cbBuilder =
126	Callback.newBuilder()	1✔
127	.setNexus(	1✔
128	Callback.Nexus.newBuilder()	1✔
129	.setUrl(request.getCallbackUrl())	1✔
130	.putAllHeader(headers)	1✔
131	.build());	1✔
132	if (links != null) {	1!
133	cbBuilder.addAllLinks(links);	1✔
134	}
135	nexusWorkflowOptions.setCompletionCallbacks(Collections.singletonList(cbBuilder.build()));	1✔
136	}
137
138	if (options.getTaskQueue() == null) {	1!
139	nexusWorkflowOptions.setTaskQueue(request.getTaskQueue());	1✔
140	}
141	nexusWorkflowOptions.setOnConflictOptions(	1✔
142	OnConflictOptions.newBuilder()	1✔
143	.setAttachRequestId(true)	1✔
144	.setAttachLinks(true)	1✔
145	.setAttachCompletionCallbacks(true)	1✔
146	.build());	1✔
147
148	return new NexusWorkflowStarter(stub.newInstance(nexusWorkflowOptions.build()), operationToken);	1✔
149	}
150
151	/**
152	* Returns true if the given name is in the {@code __temporal_workflow_stream_} sub-namespace,
153	* which is reserved for the workflow streams contrib module and permitted for signal, update, and
154	* query handler registration.
155	*/
156	public static boolean isWorkflowStreamReservedName(String name) {
157	return name.startsWith(WORKFLOW_STREAM_RESERVED_PREFIX);	1✔
158	}
159
160	/** Helper to build a Nexus Callback from the provided input. */
161	public static Callback buildNexusCallback(
162	Map<String, String> callbackHeaders,
163	String callbackUrl,
164	String operationToken,
165	List<Link> links) {
166	Map<String, String> headers =	1✔
167	callbackHeaders.entrySet().stream()	1✔
168	.collect(	1✔
169	Collectors.toMap(	1✔
170	(k) -> k.getKey().toLowerCase(),	1✔
171	Map.Entry::getValue,
NEW 172	(a, b) -> a,	×
173	TreeMap::new));
174	headers.put(Header.OPERATION_TOKEN.toLowerCase(), operationToken);	1✔
175	Callback.Builder cbBuilder =
176	Callback.newBuilder()	1✔
177	.setNexus(	1✔
178	Callback.Nexus.newBuilder().setUrl(callbackUrl).putAllHeader(headers).build());	1✔
179	if (links != null) {	1!
180	cbBuilder.addAllLinks(links);	1✔
181	}
182	return cbBuilder.build();	1✔
183	}
184
185	/** Check the method name for reserved prefixes or names. */
186	public static void checkMethodName(POJOWorkflowMethodMetadata methodMetadata) {
187	boolean workflowStreamExempt =	1✔
188	!methodMetadata.getType().equals(WorkflowMethodType.WORKFLOW)	1✔
189	&& isWorkflowStreamReservedName(methodMetadata.getName());	1✔
190	if (methodMetadata.getName().startsWith(TEMPORAL_RESERVED_PREFIX) && !workflowStreamExempt) {	1✔
191	throw new IllegalArgumentException(	1✔
192	methodMetadata.getType().toString().toLowerCase()	1✔
193	+ " name \""
194	+ methodMetadata.getName()	1✔
195	+ "\" must not start with \""
196	+ TEMPORAL_RESERVED_PREFIX
197	+ "\"");
198	}
199	if (methodMetadata.getType().equals(WorkflowMethodType.QUERY)	1✔
200	&& (methodMetadata.getName().equals(QUERY_TYPE_STACK_TRACE)	1!
201	\|\| methodMetadata.getName().equals(ENHANCED_QUERY_TYPE_STACK_TRACE))) {	1!
202	throw new IllegalArgumentException(	×
203	"Query method name \"" + methodMetadata.getName() + "\" is reserved for internal use");	×
204	}
205	}	1✔
206
207	public static void checkMethodName(POJOActivityMethodMetadata methodMetadata) {
208	if (methodMetadata.getActivityTypeName().startsWith(TEMPORAL_RESERVED_PREFIX)) {	1✔
209	throw new IllegalArgumentException(	1✔
210	"Activity name \""
211	+ methodMetadata.getActivityTypeName()	1✔
212	+ "\" must not start with \""
213	+ TEMPORAL_RESERVED_PREFIX
214	+ "\"");
215	}
216	}	1✔
217
218	/** Prohibit instantiation */
219	private InternalUtils() {}
220
221	public static void checkMethodName(ServiceImplInstance instance) {
222	if (instance.getDefinition().getName().startsWith(TEMPORAL_RESERVED_PREFIX)) {	1!
223	throw new IllegalArgumentException(	×
224	"Service name \""
225	+ instance.getDefinition().getName()	×
226	+ "\" must not start with \""
227	+ TEMPORAL_RESERVED_PREFIX
228	+ "\"");
229	}
230	for (String operationName : instance.getDefinition().getOperations().keySet()) {	1✔
231	if (operationName.startsWith(TEMPORAL_RESERVED_PREFIX)) {	1!
232	throw new IllegalArgumentException(	×
233	"Operation name \""
234	+ operationName
235	+ "\" must not start with \""
236	+ TEMPORAL_RESERVED_PREFIX
237	+ "\"");
238	}
239	}	1✔
240	}	1✔
241	}

temporalio / sdk-java / #347

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous