Query commit

Author: Evanthx

temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java

@@ -434,6 +434,14 @@ public <R> QueryOutput<R> query(QueryInput<R> input) {
     QueryWorkflowResponse result;
     result = genericClient.query(request);
 
+    // A query writes nothing to history, so the server returns a link to the workflow execution
+    // that processed it rather than to an event. When the query is issued from inside a Nexus
+    // operation handler, propagate that link so the caller's Nexus operation event points at the
+    // queried workflow. Older servers leave it unset.
+    if (CurrentNexusOperationContext.isNexusContext() && result.hasLink()) {
+      CurrentNexusOperationContext.get().addResponseLink(result.getLink());
+    }
+
     boolean queryRejected = result.hasQueryRejected();
     WorkflowExecutionStatus rejectStatus =
         queryRejected ? result.getQueryRejected().getStatus() : null;

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

@@ -25,10 +25,12 @@ public class LinkConverter {
       "temporal:///namespaces/%s/nexus-operations/%s/%s/details";
   private static final String activityLinkPathFormat =
       "temporal:///namespaces/%s/activities/%s/%s/details";
+  private static final String workflowLinkPathFormat = "temporal:///namespaces/%s/workflows/%s/%s";
   private static final String linkReferenceTypeKey = "referenceType";
   private static final String linkEventIDKey = "eventID";
   private static final String linkEventTypeKey = "eventType";
   private static final String linkRequestIDKey = "requestID";
+  private static final String linkReasonKey = "reason";
 
   private static final String eventReferenceType =
       Link.WorkflowEvent.EventReference.getDescriptor().getName();
@@ -98,14 +100,28 @@ public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.Workfl
     return null;
   }
 
+  /**
+   * Converts a {@link Link.Workflow} to a Nexus link. A workflow link addresses a workflow
+   * execution as a whole rather than one event within it, so the URL uses the workflow path and
+   * carries no event path suffix and no reference query params. It is used when there is no history
+   * event to point at, for example a Query or a rejected Update. The optional {@code reason}
+   * explaining why the link exists is carried as a query param.
+   */
   public static io.temporal.api.nexus.v1.Link workflowLinkToNexusLink(Link.Workflow w) {
     try {
-      String namespace = URLEncoder.encode(w.getNamespace(), StandardCharsets.UTF_8.toString());
-      String workflowId =
-          URLEncoder.encode(w.getWorkflowId(), StandardCharsets.UTF_8.toString())
-              .replace("+", "%20"); // handle workflowIds supporting spaces
-      String runId = URLEncoder.encode(w.getRunId(), StandardCharsets.UTF_8.toString());
-      String url = String.format(linkPathFormat, namespace, workflowId, runId);
+      String url =
+          String.format(
+              workflowLinkPathFormat,
+              encodePathSegment(w.getNamespace()),
+              encodePathSegment(w.getWorkflowId()),
+              encodePathSegment(w.getRunId()));
+      if (!w.getReason().isEmpty()) {
+        url +=
+            "?"
+                + linkReasonKey
+                + "="
+                + URLEncoder.encode(w.getReason(), StandardCharsets.UTF_8.toString());
+      }
       return io.temporal.api.nexus.v1.Link.newBuilder()
           .setUrl(url)
           .setType(workflowLinkType)
@@ -190,16 +206,25 @@ public static Link nexusLinkToWorkflowEvent(io.temporal.api.nexus.v1.Link nexusL
   }
 
   public static Link nexusLinkToWorkflowLink(io.temporal.api.nexus.v1.Link nexusLink) {
+    if (!workflowLinkType.equals(nexusLink.getType())) {
+      log.error(
+          "Failed to parse Nexus link URL: cannot parse link type {} to {}",
+          nexusLink.getType(),
+          workflowLinkType);
+      return null;
+    }
     Link.Builder link = Link.newBuilder();
     try {
       URI uri = new URI(nexusLink.getUrl());
-      log.debug("Parsing nexus link URL: {}", uri.getRawPath());
-      if (!uri.getScheme().equals(temporalUrlScheme)) {
+
+      // Compared in this order so a URL with no scheme at all reports the invalid scheme rather
+      // than throwing.
+      if (!temporalUrlScheme.equals(uri.getScheme())) {
         log.error("Failed to parse Nexus link URL: invalid scheme: {}", uri.getScheme());
         return null;
       }
+
       StringTokenizer st = new StringTokenizer(uri.getRawPath(), "/");
-      // maybe add constants for "namespaces", "workflows" too
       if (!st.nextToken().equals("namespaces")) {
         log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
         return null;
@@ -210,18 +235,28 @@ public static Link nexusLinkToWorkflowLink(io.temporal.api.nexus.v1.Link nexusLi
         return null;
       }
       String workflowID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
-      if (!st.hasMoreTokens()) {
+      String runID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
+      // The run ID ends a workflow link, so anything trailing means this is a different link
+      // shape. In particular this rejects the workflow-event form, which ends in "/history".
+      if (st.hasMoreTokens()) {
         log.error("Failed to parse Nexus link URL: invalid path: {}", uri.getRawPath());
         return null;
       }
-      String runID = URLDecoder.decode(st.nextToken(), StandardCharsets.UTF_8.toString());
-      link.setWorkflow(
+
+      Link.Workflow.Builder w =
           Link.Workflow.newBuilder()
               .setNamespace(namespace)
               .setWorkflowId(workflowID)
-              .setRunId(runID));
+              .setRunId(runID);
+      String reason = rawQueryParam(uri, linkReasonKey);
+      if (reason != null) {
+        w.setReason(reason);
+      }
+
+      link.setWorkflow(w);
     } catch (Exception e) {
-      log.error("Failed to convert NexusLink {} to WorkflowLink", nexusLink, e);
+      // Swallow un-parsable links since they are not critical to processing.
+      log.error("Failed to parse Nexus link URL", e);
       return null;
     }
     return link.build();
@@ -406,6 +441,33 @@ public static Link nexusLinkToNexusOperation(io.temporal.api.nexus.v1.Link nexus
     return link.build();
   }
 
+  /**
+   * Percent-encodes a single URL path segment. {@link URLEncoder} targets form encoding, where a
+   * space becomes '+', so rewrite it to "%20" as required for a path.
+   */
+  private static String encodePathSegment(String value) throws UnsupportedEncodingException {
+    return URLEncoder.encode(value, StandardCharsets.UTF_8.toString()).replace("+", "%20");
+  }
+
+  /**
+   * Reads a single param out of the raw, still-encoded query string, or returns null when the param
+   * is absent. Unlike {@link #parseQueryParams} the value is decoded exactly once, so values that
+   * themselves contain '=' or '&' survive the round trip.
+   */
+  private static String rawQueryParam(URI uri, String key) throws UnsupportedEncodingException {
+    final String rawQuery = uri.getRawQuery();
+    if (rawQuery == null || rawQuery.isEmpty()) {
+      return null;
+    }
+    for (String pair : rawQuery.split("&")) {
+      final String[] kv = pair.split("=", 2);
+      if (kv[0].equals(key)) {
+        return kv.length == 2 ? URLDecoder.decode(kv[1], StandardCharsets.UTF_8.toString()) : "";
+      }
+    }
+    return null;
+  }
+
   private static Map<String, String> parseQueryParams(URI uri) throws UnsupportedEncodingException {
     final String query = uri.getQuery();
     if (query == null || query.isEmpty()) {

temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java

@@ -7,10 +7,15 @@
 import com.uber.m3.tally.RootScopeBuilder;
 import com.uber.m3.tally.Scope;
 import io.temporal.api.common.v1.Link;
+import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.api.enums.v1.EventType;
 import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
+import io.temporal.api.enums.v1.WorkflowExecutionStatus;
+import io.temporal.api.query.v1.QueryRejected;
 import io.temporal.api.update.v1.UpdateRef;
+import io.temporal.api.workflowservice.v1.QueryWorkflowRequest;
+import io.temporal.api.workflowservice.v1.QueryWorkflowResponse;
 import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest;
 import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse;
 import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest;
@@ -23,7 +28,9 @@
 import io.temporal.client.WorkflowClientOptions;
 import io.temporal.client.WorkflowOptions;
 import io.temporal.client.WorkflowUpdateStage;
+import io.temporal.common.converter.DefaultDataConverter;
 import io.temporal.common.interceptors.Header;
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
 import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.StartUpdateInput;
 import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput;
 import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalWithStartInput;
@@ -348,6 +355,116 @@ private static StartUpdateInput<String> newStartUpdateInput() {
             .build());
   }
 
+  /**
+   * A query never writes to history, so the server answers with a {@code Link.Workflow} naming the
+   * execution that processed it instead of a {@code Link.WorkflowEvent}. That link has to reach the
+   * operation context so the caller's Nexus operation event points back at the queried workflow.
+   */
+  @Test
+  public void queryCapturesWorkflowResponseLink() {
+    Link responseLink = workflowLink(WORKFLOW_ID, "target-run", "Query processed");
+    when(genericClient.query(any(QueryWorkflowRequest.class)))
+        .thenReturn(
+            QueryWorkflowResponse.newBuilder()
+                .setLink(responseLink)
+                .setQueryResult(queryResult("answer"))
+                .build());
+
+    WorkflowClientCallsInterceptor.QueryOutput<String> output = invoker.query(newQueryInput());
+
+    List<Link> captured = nexusCtx.getResponseLinks();
+    Assert.assertEquals("expected one captured response link", 1, captured.size());
+    Assert.assertEquals(responseLink, captured.get(0));
+
+    // Capturing the link must not disturb the query's own result.
+    Assert.assertFalse(output.isQueryRejected());
+    Assert.assertEquals("answer", output.getResult());
+  }
+
+  /**
+   * Two queries in a row each contribute a response link; both must accumulate in call order on the
+   * shared list, exactly as the signal path does.
+   */
+  @Test
+  public void multipleQueriesAccumulateAllResponseLinks() {
+    Link firstResponseLink = workflowLink("callee-a", "run-a", "Query processed");
+    Link secondResponseLink = workflowLink("callee-b", "run-b", "Query processed");
+    when(genericClient.query(any(QueryWorkflowRequest.class)))
+        .thenReturn(QueryWorkflowResponse.newBuilder().setLink(firstResponseLink).build())
+        .thenReturn(QueryWorkflowResponse.newBuilder().setLink(secondResponseLink).build());
+
+    invoker.query(newQueryInput());
+    invoker.query(newQueryInput());
+
+    Assert.assertEquals(
+        "expected one response link per query call, in call order",
+        Arrays.asList(firstResponseLink, secondResponseLink),
+        nexusCtx.getResponseLinks());
+  }
+
+  /**
+   * A rejected query still carries a link to the workflow that rejected it, and the link is
+   * captured before the rejection is surfaced. This matches sdk-go, where the link is recorded
+   * ahead of the QueryRejected branch. Pins the ordering so it is not "fixed" into the wrong
+   * behavior later.
+   */
+  @Test
+  public void rejectedQueryStillCapturesResponseLink() {
+    Link responseLink = workflowLink(WORKFLOW_ID, "target-run", "Query processed");
+    when(genericClient.query(any(QueryWorkflowRequest.class)))
+        .thenReturn(
+            QueryWorkflowResponse.newBuilder()
+                .setLink(responseLink)
+                .setQueryRejected(
+                    QueryRejected.newBuilder()
+                        .setStatus(WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED))
+                .build());
+
+    WorkflowClientCallsInterceptor.QueryOutput<String> output = invoker.query(newQueryInput());
+
+    Assert.assertTrue("expected the query to be reported as rejected", output.isQueryRejected());
+    Assert.assertEquals(
+        "expected the response link to be captured even for a rejected query",
+        Collections.singletonList(responseLink),
+        nexusCtx.getResponseLinks());
+  }
+
+  /**
+   * Older-server compatibility: {@code QueryWorkflowResponse.link} is unset, so nothing is captured
+   * and the query itself still succeeds.
+   */
+  @Test
+  public void queryAgainstOlderServerCapturesNoResponseLink() {
+    when(genericClient.query(any(QueryWorkflowRequest.class)))
+        .thenReturn(QueryWorkflowResponse.getDefaultInstance());
+
+    invoker.query(newQueryInput());
+
+    Assert.assertTrue(
+        "expected no captured response link when server returned no link",
+        nexusCtx.getResponseLinks().isEmpty());
+  }
+
+  /**
+   * A query issued outside a Nexus operation handler must not touch the operation context at all.
+   * Guards against the propagation being reached without a context, which would throw.
+   */
+  @Test
+  public void queryOutsideNexusContextIgnoresResponseLink() {
+    CurrentNexusOperationContext.unset();
+    when(genericClient.query(any(QueryWorkflowRequest.class)))
+        .thenReturn(
+            QueryWorkflowResponse.newBuilder()
+                .setLink(workflowLink(WORKFLOW_ID, "target-run", "Query processed"))
+                .build());
+
+    invoker.query(newQueryInput());
+
+    Assert.assertTrue(
+        "a query outside a Nexus context must not record response links",
+        nexusCtx.getResponseLinks().isEmpty());
+  }
+
   // ── helpers ──────────────────────────────────────────────────────────────────────────────
 
   private static WorkflowSignalInput newSignalInput() {
@@ -374,6 +491,33 @@ private static WorkflowSignalWithStartInput newSignalWithStartInput() {
         startInput, "test-signal", new Object[] {"signal-payload"});
   }
 
+  private static WorkflowClientCallsInterceptor.QueryInput<String> newQueryInput() {
+    return new WorkflowClientCallsInterceptor.QueryInput<>(
+        WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).build(),
+        "test-query",
+        Header.empty(),
+        new Object[] {},
+        String.class,
+        String.class);
+  }
+
+  private static Payloads queryResult(String value) {
+    return DefaultDataConverter.STANDARD_INSTANCE
+        .toPayloads(value)
+        .orElseThrow(() -> new IllegalStateException("expected payloads"));
+  }
+
+  private static Link workflowLink(String workflowId, String runId, String reason) {
+    return Link.newBuilder()
+        .setWorkflow(
+            Link.Workflow.newBuilder()
+                .setNamespace(NAMESPACE)
+                .setWorkflowId(workflowId)
+                .setRunId(runId)
+                .setReason(reason))
+        .build();
+  }
+
   private static Link workflowEventLink(String workflowId, String runId, EventType eventType) {
     return Link.newBuilder()
         .setWorkflowEvent(