fix: trigger callbacks when resetting workflow to not base (#8321)

## What changed? When a user resets a workflow to not the base workflow, we should still trigger nexus callbacks. ## Why? Consistency in user experience, fixing a bug in the implementation. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [X] covered by existing tests - [ ] added new unit test(s) - [X] added new functional test(s) ## Potential risks Could have an increase in the number of callbacks.
2026-10-04 02:04:03 +00:00 · 2025-09-18 14:35:16 -06:00
parent b42eac62ab
commit 33d18224f8
2 changed files with 125 additions and 1 deletions
@@ -5906,7 +5906,8 @@ func (ms *MutableStateImpl) AddHistorySize(size int64) {
 // processCloseCallbacks triggers "WorkflowClosed" callbacks, applying the state machine transition that schedules
 // callback tasks.
 func (ms *MutableStateImpl) processCloseCallbacks() error {
-	if ms.GetExecutionInfo().GetWorkflowWasReset() {
+	resetRunID := ms.GetExecutionInfo().GetResetRunId()
+	if ms.GetExecutionInfo().GetWorkflowWasReset() && resetRunID != "" && resetRunID != ms.executionState.RunId {
 		return nil
 	}

@@ -601,3 +601,126 @@ func (s *CallbacksSuite) TestNexusResetWorkflowWithCallback() {
 		100*time.Millisecond,
 	)
 }
+
+func blockingWorkflow(ctx workflow.Context) error {
+	return workflow.Await(ctx, func() bool {
+		return false
+	})
+}
+
+func (s *CallbacksSuite) TestNexusResetWorkflowWithCallback_ResetToNotBaseRun() {
+	s.OverrideDynamicConfig(
+		callbacks.AllowedAddresses,
+		[]any{map[string]any{"Pattern": "*", "AllowInsecure": true}},
+	)
+
+	/*
+	 * 1. Start WF w/ no callbacks and immediately terminate
+	 * 2. Start WF second time w/ a callback
+	 * 3. Reset WF back to the first run
+	 * 4. Verify callback is called
+	 */
+
+	tv := testvars.New(s.T())
+	ctx := testcore.NewContext()
+	sdkClient, err := client.Dial(client.Options{
+		HostPort:  s.FrontendGRPCAddress(),
+		Namespace: s.Namespace().String(),
+	})
+	s.NoError(err)
+
+	taskQueue := tv.TaskQueue()
+	workflowID := tv.WorkflowID()
+
+	ch := &completionHandler{
+		requestCh:         make(chan *nexus.CompletionRequest, 1),
+		requestCompleteCh: make(chan error, 1),
+	}
+	callbackAddress := fmt.Sprintf("localhost:%d", freeport.MustGetFreePort())
+	s.runNexusCompletionHTTPServer(s.T(), ch, callbackAddress)
+
+	w := worker.New(sdkClient, taskQueue.GetName(), worker.Options{})
+
+	w.RegisterWorkflow(blockingWorkflow)
+	s.NoError(w.Start())
+	defer w.Stop()
+
+	// 1. Start WF w/ no callbacks and immediately terminate
+	request1 := &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:          uuid.NewString(),
+		Namespace:          s.Namespace().String(),
+		WorkflowId:         workflowID,
+		WorkflowType:       &commonpb.WorkflowType{Name: "blockingWorkflow"},
+		TaskQueue:          taskQueue,
+		Input:              nil,
+		WorkflowRunTimeout: durationpb.New(20 * time.Second),
+		Identity:           s.T().Name(),
+	}
+
+	startResponse1, err := s.FrontendClient().StartWorkflowExecution(ctx, request1)
+	s.NoError(err)
+
+	// Validate the workflow started, then terminate it
+	workflowExecution := &commonpb.WorkflowExecution{
+		WorkflowId: workflowID,
+		RunId:      startResponse1.RunId,
+	}
+	s.WaitForHistoryEvents(`
+			1 WorkflowExecutionStarted
+			2 WorkflowTaskScheduled
+			3 WorkflowTaskStarted
+			4 WorkflowTaskCompleted`,
+		s.GetHistoryFunc(s.Namespace().String(), workflowExecution),
+		5*time.Second,
+		10*time.Millisecond)
+
+	_, err = s.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
+		Namespace:         s.Namespace().String(),
+		WorkflowExecution: workflowExecution,
+		Reason:            s.T().Name(),
+		Identity:          tv.WorkerIdentity(),
+	})
+	s.NoError(err)
+
+	// 2. Start WF second time w/ callbacks (new run)
+	cbs := []*commonpb.Callback{
+		{Variant: &commonpb.Callback_Nexus_{Nexus: &commonpb.Callback_Nexus{Url: "http://" + callbackAddress + "/cb1"}}},
+	}
+
+	request2 := proto.Clone(request1).(*workflowservice.StartWorkflowExecutionRequest)
+	request2.RequestId = uuid.NewString()
+	request2.CompletionCallbacks = cbs
+
+	_, err = s.FrontendClient().StartWorkflowExecution(ctx, request2)
+	s.NoError(err)
+
+	// 3. Reset workflow back to the first (terminated) run as base; must copy callbacks
+	_, err = sdkClient.ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
+		Namespace:                 s.Namespace().String(),
+		WorkflowExecution:         workflowExecution, // base = first (terminated) run
+		Reason:                    s.T().Name(),
+		WorkflowTaskFinishEventId: 4,
+		RequestId:                 "test_id",
+	})
+	s.NoError(err)
+
+	// 4. Wait for callback deliveries via the handler channels
+	select {
+	case completion := <-ch.requestCh:
+		s.Equal(nexus.OperationStateFailed, completion.State)
+		ch.requestCompleteCh <- nil
+	case <-ctx.Done():
+		s.FailNow("timed out waiting for callback")
+	}
+
+	// Ensure the original workflow runs to completion to avoid leaving dangling runs
+	_, err = s.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
+		Namespace: s.Namespace().String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: workflowID,
+		},
+		Reason:   s.T().Name(),
+		Identity: tv.WorkerIdentity(),
+	})
+	s.NoError(err)
+}