Skip to content

Commit 67bf833

Browse files
EvanthxEvanthx
committed
Reviewing
1 parent 7b4f3a8 commit 67bf833

6 files changed

Lines changed: 137 additions & 80 deletions

File tree

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

Lines changed: 7 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@
1010
import io.grpc.Status;
1111
import io.grpc.StatusRuntimeException;
1212
import io.temporal.api.common.v1.*;
13-
import io.temporal.api.enums.v1.EventType;
1413
import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
1514
import io.temporal.api.enums.v1.WorkflowExecutionStatus;
1615
import io.temporal.api.errordetails.v1.MultiOperationExecutionFailure;
@@ -103,29 +102,14 @@ public WorkflowStartOutput start(WorkflowStartInput input) {
103102
e);
104103
}
105104
}
105+
// If this start is being issued from inside a Nexus operation handler, stash only the
106+
// forward operation->workflow link from the start response so NexusStartWorkflowHelper can
107+
// attach it to the WorkflowExecutionStarted event. Unlike signal/signalWithStart, start
108+
// deliberately does NOT add a backlink here: the operation->workflow relationship is already
109+
// captured by the forward link, so re-adding response.getLink() as a backlink would duplicate
110+
// it on the caller's history event. Do not "restore symmetry" by calling addBacklink here.
106111
if (CurrentNexusOperationContext.isNexusContext()) {
107-
// Auto-capture the start-workflow backlink so the task handler drains it onto the
108-
// StartOperationResponse, the same path used for signal/signalWithStart responses.
109-
if (response.hasLink()) {
110-
CurrentNexusOperationContext.get().addBacklink(response.getLink());
111-
} else {
112-
// Older servers (pre-1.31) don't return a link on the start response. Fabricate one
113-
// pointing at the started workflow's WorkflowExecutionStarted event so the caller still
114-
// gets a backlink.
115-
CurrentNexusOperationContext.get()
116-
.addBacklink(
117-
Link.newBuilder()
118-
.setWorkflowEvent(
119-
Link.WorkflowEvent.newBuilder()
120-
.setNamespace(clientOptions.getNamespace())
121-
.setWorkflowId(execution.getWorkflowId())
122-
.setRunId(execution.getRunId())
123-
.setEventRef(
124-
Link.WorkflowEvent.EventReference.newBuilder()
125-
.setEventType(
126-
EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED)))
127-
.build());
128-
}
112+
CurrentNexusOperationContext.get().setStartWorkflowResponseLink(response.getLink());
129113
}
130114
return new WorkflowStartOutput(execution);
131115
}

‎temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java‎

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,11 @@ public class InternalNexusOperationContext {
1818
private final Scope metricScope;
1919
private final WorkflowClient client;
2020
NexusOperationOutboundCallsInterceptor outboundCalls;
21+
// Link returned by the StartWorkflowExecution response when the operation is backed by a workflow
22+
// (workflow-run operations). Read by NexusStartWorkflowHelper to attach the forward
23+
// operation->workflow link, fabricating a WORKFLOW_EXECUTION_STARTED link when the server omits
24+
// one. Distinct from the signal backlinks below.
25+
Link startWorkflowResponseLink;
2126
// Links extracted from the inbound Nexus task. Stored once at the task-handler boundary so the
2227
// workflow client can attach them to the outgoing requests it issues (e.g. signal,
2328
// signalWithStart) via the request's links field.
@@ -28,9 +33,9 @@ public class InternalNexusOperationContext {
2833
// by the task handler when building StartOperationResponse so each RPC the handler issued gets a
2934
// corresponding link on the caller workflow's history event.
3035
//
31-
// This context is only safe for use from the single thread that runs the operation handler (the
32-
// Nexus task executor's thread); the backing ArrayList is not synchronized. Handlers must not
33-
// mutate it from other threads.
36+
// A handler may issue RPCs from multiple threads, so every read and write of this list is guarded
37+
// by backlinksLock and getBacklinks() returns a defensive copy taken under the lock.
38+
private final Object backlinksLock = new Object();
3439
private final List<Link> responseBacklinks = new ArrayList<>();
3540

3641
public InternalNexusOperationContext(
@@ -90,14 +95,24 @@ public void setNexusOperationLinks(List<Link> links) {
9095
return Collections.unmodifiableList(nexusOperationLinks);
9196
}
9297

98+
public void setStartWorkflowResponseLink(Link link) {
99+
this.startWorkflowResponseLink = link;
100+
}
101+
102+
public Link getStartWorkflowResponseLink() {
103+
return startWorkflowResponseLink;
104+
}
105+
93106
/**
94107
* Append a backlink returned by an outbound RPC the operation handler issued (e.g. signal,
95108
* signalWithStart, etc). The task handler drains the list when building the operation's
96109
* StartOperationResponse.
97110
*/
98111
public void addBacklink(Link link) {
99112
if (link != null) {
100-
this.responseBacklinks.add(link);
113+
synchronized (backlinksLock) {
114+
responseBacklinks.add(link);
115+
}
101116
}
102117
}
103118

@@ -109,7 +124,9 @@ public void addBacklink(Link link) {
109124
* drained afterward by the task handler when building the StartOperationResponse.
110125
*/
111126
public @Nonnull List<Link> getBacklinks() {
112-
return Collections.unmodifiableList(responseBacklinks);
127+
synchronized (backlinksLock) {
128+
return Collections.unmodifiableList(new ArrayList<>(responseBacklinks));
129+
}
113130
}
114131

115132
private class NexusOperationContextImpl implements NexusOperationContext {

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

Lines changed: 28 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -235,48 +235,35 @@ public void mixedSignalAndSignalWithStartAccumulateAllBacklinks() {
235235
}
236236

237237
/**
238-
* Flag-enabled server: the start response carries a backlink; the SDK captures it verbatim onto
239-
* the operation context (used by the WorkflowRunOperation async path).
238+
* Post-rebase start contract: a plain {@code start()} issued from inside a Nexus operation
239+
* handler captures only the FORWARD operation->workflow link (via {@code
240+
* setStartWorkflowResponseLink}) and deliberately does NOT add a backlink (unlike
241+
* signal/signalWithStart). Replaces the two start-backlink tests removed by the rebase and guards
242+
* against a regression that re-adds a backlink on the start path.
240243
*/
241244
@Test
242-
public void startUsesServerStartLinkWhenPresent() {
243-
Link serverLink =
245+
public void startSetsForwardLinkOnlyAndCapturesNoBacklink() {
246+
Link startResponseLink =
244247
workflowEventLink(
245248
WORKFLOW_ID, "target-run", EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED);
246-
when(genericClient.start(any(StartWorkflowExecutionRequest.class)))
247-
.thenReturn(
248-
StartWorkflowExecutionResponse.newBuilder()
249-
.setRunId("target-run")
250-
.setLink(serverLink)
251-
.build());
252-
253-
invoker.start(newStartInput());
254-
255-
List<Link> captured = nexusCtx.getBacklinks();
256-
Assert.assertEquals("expected the server-provided start link", 1, captured.size());
257-
Assert.assertEquals(serverLink, captured.get(0));
258-
}
259-
260-
/**
261-
* Older server (pre-1.31): the start response has no link. The SDK must fabricate a backlink
262-
* pointing at the started workflow's WorkflowExecutionStarted event so the caller still links to
263-
* the callee.
264-
*/
265-
@Test
266-
public void startFabricatesStartLinkWhenServerOmitsIt() {
267-
when(genericClient.start(any(StartWorkflowExecutionRequest.class)))
268-
.thenReturn(StartWorkflowExecutionResponse.newBuilder().setRunId("target-run").build());
249+
StartWorkflowExecutionResponse response =
250+
StartWorkflowExecutionResponse.newBuilder()
251+
.setRunId("target-run")
252+
.setLink(startResponseLink)
253+
.build();
254+
when(genericClient.start(any(StartWorkflowExecutionRequest.class))).thenReturn(response);
269255

270256
invoker.start(newStartInput());
271257

272-
List<Link> captured = nexusCtx.getBacklinks();
273-
Assert.assertEquals("expected one fabricated backlink", 1, captured.size());
274-
Link.WorkflowEvent we = captured.get(0).getWorkflowEvent();
275-
Assert.assertEquals(NAMESPACE, we.getNamespace());
276-
Assert.assertEquals(WORKFLOW_ID, we.getWorkflowId());
277-
Assert.assertEquals("target-run", we.getRunId());
258+
// Forward direction: the start response link is stashed for NexusStartWorkflowHelper to read.
278259
Assert.assertEquals(
279-
EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, we.getEventRef().getEventType());
260+
"expected the forward start link on the context",
261+
startResponseLink,
262+
nexusCtx.getStartWorkflowResponseLink());
263+
264+
// Backward direction: start must not add a backlink.
265+
Assert.assertTrue(
266+
"expected no backlink captured on the start path", nexusCtx.getBacklinks().isEmpty());
280267
}
281268

282269
// ── helpers ──────────────────────────────────────────────────────────────────────────────
@@ -289,6 +276,13 @@ private static WorkflowSignalInput newSignalInput() {
289276
new Object[] {"payload"});
290277
}
291278

279+
private static WorkflowStartInput newStartInput() {
280+
WorkflowOptions options =
281+
WorkflowOptions.newBuilder().setTaskQueue("tq").setDisableEagerExecution(true).build();
282+
return new WorkflowStartInput(
283+
WORKFLOW_ID, "TestWorkflow", Header.empty(), new Object[] {}, options);
284+
}
285+
292286
private static WorkflowSignalWithStartInput newSignalWithStartInput() {
293287
WorkflowOptions options = WorkflowOptions.newBuilder().setTaskQueue("tq").build();
294288
WorkflowStartInput startInput =
@@ -298,14 +292,6 @@ private static WorkflowSignalWithStartInput newSignalWithStartInput() {
298292
startInput, "test-signal", new Object[] {"signal-payload"});
299293
}
300294

301-
private static WorkflowStartInput newStartInput() {
302-
// Disable eager execution so start() takes the plain RPC path (no local worker dispatch).
303-
WorkflowOptions options =
304-
WorkflowOptions.newBuilder().setTaskQueue("tq").setDisableEagerExecution(true).build();
305-
return new WorkflowStartInput(
306-
WORKFLOW_ID, "TestWorkflow", Header.empty(), new Object[] {}, options);
307-
}
308-
309295
private static Link workflowEventLink(String workflowId, String runId, EventType eventType) {
310296
return Link.newBuilder()
311297
.setWorkflowEvent(

‎temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerImplTest.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import com.uber.m3.tally.Scope;
88
import com.uber.m3.util.Duration;
99
import io.nexusrpc.Header;
10+
import io.nexusrpc.OperationException;
1011
import io.nexusrpc.handler.*;
1112
import io.temporal.api.common.v1.Link;
1213
import io.temporal.api.common.v1.Payload;
@@ -239,6 +240,49 @@ public void syncResponseIncludesSignalBacklinks() throws TimeoutException {
239240
sync.getLinks(0).getUrl().contains("callee-wf"));
240241
}
241242

243+
/**
244+
* Failure path: a handler that stashes a backlink (as a successful signal RPC would) and then
245+
* throws afterwards must NOT leak the captured backlink onto the failure response. Backlinks are
246+
* only drained on the success branch of {@link NexusTaskHandlerImpl#handleStartOperation}; the
247+
* failure branch builds a {@code StartOperationResponse.failure} that carries no links.
248+
*/
249+
@Test
250+
public void failureResponseDropsCapturedBacklinks() throws TimeoutException {
251+
WorkflowClient client = mock(WorkflowClient.class);
252+
NexusTaskHandlerImpl nexusTaskHandlerImpl =
253+
new NexusTaskHandlerImpl(
254+
client, NAMESPACE, TASK_QUEUE, dataConverter, new WorkerInterceptor[] {});
255+
nexusTaskHandlerImpl.registerNexusServiceImplementations(
256+
new Object[] {new BacklinkStashingThenThrowingServiceImpl()});
257+
nexusTaskHandlerImpl.start();
258+
259+
PollNexusTaskQueueResponse.Builder task =
260+
PollNexusTaskQueueResponse.newBuilder()
261+
.setRequest(
262+
Request.newBuilder()
263+
.setStartOperation(
264+
StartOperationRequest.newBuilder()
265+
.setOperation("operation")
266+
.setService("TestNexusService1")
267+
.setPayload(dataConverter.toPayload("input").get())
268+
.build()));
269+
270+
NexusTaskHandler.Result result =
271+
nexusTaskHandlerImpl.handle(new NexusTask(task, null, null), metricsScope);
272+
273+
Assert.assertNull(result.getHandlerException());
274+
StartOperationResponse response = result.getResponse().getStartOperation();
275+
Assert.assertEquals(
276+
"expected the failure response variant",
277+
StartOperationResponse.VariantCase.FAILURE,
278+
response.getVariantCase());
279+
// The handler captured a backlink before throwing; the failure response must not carry it
280+
// (and has no links field at all).
281+
Assert.assertFalse(
282+
"failure response variant should not expose any success-path links",
283+
response.hasSyncSuccess() || response.hasAsyncSuccess());
284+
}
285+
242286
/**
243287
* Handler that simulates what a real Nexus operation would do after issuing a signal: stash a
244288
* backlink on the operation context, then return an async result. Lets us exercise the
@@ -299,6 +343,35 @@ public OperationHandler<String, String> operation() {
299343
}
300344
}
301345

346+
/**
347+
* Stashes a backlink on the operation context (as a successful signal RPC would) and then throws
348+
* an {@link OperationException}, exercising the failure branch of {@link
349+
* NexusTaskHandlerImpl#handleStartOperation}.
350+
*/
351+
@ServiceImpl(service = TestNexusServices.TestNexusService1.class)
352+
public class BacklinkStashingThenThrowingServiceImpl {
353+
@OperationImpl
354+
public OperationHandler<String, String> operation() {
355+
return OperationHandler.sync(
356+
(ctx, details, input) -> {
357+
Link backlink =
358+
Link.newBuilder()
359+
.setWorkflowEvent(
360+
Link.WorkflowEvent.newBuilder()
361+
.setNamespace(NAMESPACE)
362+
.setWorkflowId("callee-wf")
363+
.setRunId("callee-run-id")
364+
.setEventRef(
365+
Link.WorkflowEvent.EventReference.newBuilder()
366+
.setEventType(
367+
EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED)))
368+
.build();
369+
CurrentNexusOperationContext.get().addBacklink(backlink);
370+
throw OperationException.failed("boom after capturing a backlink");
371+
});
372+
}
373+
}
374+
302375
@ServiceImpl(service = TestNexusServices.TestNexusService1.class)
303376
public class TestNexusServiceImpl {
304377
@OperationImpl

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/SignalOperationLinkingTest.java‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -95,14 +95,11 @@ public void testSignalOperationLinks() {
9595
@Test
9696
public void testMultiSignalOperationLinks() {
9797
WorkflowClient client = testWorkflowRule.getWorkflowClient();
98-
List<String> calleeIds =
99-
Arrays.asList(
100-
"multi-callee-a-" + UUID.randomUUID(),
101-
"multi-callee-b-" + UUID.randomUUID(),
102-
"multi-callee-c-" + UUID.randomUUID());
98+
List<String> calleeIds = Arrays.asList("multicallee-a", "multicallee-b", "multicallee-c");
10399

104100
TestWorkflows.TestWorkflow1 callerStub =
105-
testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class, "caller");
101+
testWorkflowRule.newWorkflowStubTimeoutOptions(
102+
TestWorkflows.TestWorkflow1.class, "multicaller");
106103
String result =
107104
callerStub.execute(MODE_MULTI_SIGNAL_WITH_START + ":" + String.join(",", calleeIds));
108105
Assert.assertEquals("ok:multi:" + String.join(",", calleeIds), result);
@@ -153,10 +150,11 @@ public void testMultiSignalOperationLinks() {
153150
@Test
154151
public void testAsyncSignalOperationLinks() {
155152
WorkflowClient client = testWorkflowRule.getWorkflowClient();
156-
String calleeWorkflowId = "async-callee-" + UUID.randomUUID();
153+
String calleeWorkflowId = "async-callee";
157154

158155
TestWorkflows.TestWorkflow1 callerStub =
159-
testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class, "caller");
156+
testWorkflowRule.newWorkflowStubTimeoutOptions(
157+
TestWorkflows.TestWorkflow1.class, "async-caller");
160158
String result = callerStub.execute(MODE_ASYNC_SIGNAL_WITH_START + ":" + calleeWorkflowId);
161159
Assert.assertEquals("async-started", result);
162160

@@ -184,7 +182,7 @@ public void testAsyncSignalOperationLinks() {
184182
/** Drive the two-signal flow (signalWithStart + plain signal) and assert link propagation. */
185183
private void runTwoSignalScenario() {
186184
WorkflowClient client = testWorkflowRule.getWorkflowClient();
187-
String calleeWorkflowId = "signal-callee-" + UUID.randomUUID();
185+
String calleeWorkflowId = "callee";
188186

189187
TestWorkflows.TestWorkflow1 callerStub =
190188
testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class, "caller");

‎temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@
3737
import io.temporal.api.taskqueue.v1.StickyExecutionAttributes;
3838
import io.temporal.api.update.v1.*;
3939
import io.temporal.api.workflow.v1.*;
40-
import io.temporal.api.workflow.v1.OnConflictOptions;
4140
import io.temporal.api.workflowservice.v1.*;
4241
import io.temporal.common.converter.DefaultDataConverter;
4342
import io.temporal.failure.ServerFailure;

0 commit comments

Comments
 (0)