Propagate backlinks on Signal and Signal-with-Start responses

Author: long-nt-tran

api/historyservice/v1/request_response.pb.go

@@ -491,7 +491,9 @@ message SignalWorkflowExecutionRequest {
   bool child_workflow_only = 4;
 }
 
-message SignalWorkflowExecutionResponse {}
+message SignalWorkflowExecutionResponse {
+  temporal.api.common.v1.Link link = 1;
+}
 
 message SignalWithStartWorkflowExecutionRequest {
   option (routing).workflow_id = "signal_with_start_request.workflow_id";
@@ -505,6 +507,8 @@ message SignalWithStartWorkflowExecutionRequest {
 message SignalWithStartWorkflowExecutionResponse {
   string run_id = 1;
   bool started = 2;
+  temporal.api.common.v1.Link signal_link = 3;
+  temporal.api.common.v1.Link workflow_start_link = 4;
 }
 
 message RemoveSignalMutableStateRequest {

proto/internal/temporal/server/api/historyservice/v1/request_response.proto

@@ -2246,15 +2246,17 @@ func (wh *WorkflowHandler) SignalWorkflowExecution(ctx context.Context, request
 		return nil, err
 	}
 
-	_, err = wh.historyClient.SignalWorkflowExecution(ctx, &historyservice.SignalWorkflowExecutionRequest{
+	resp, err := wh.historyClient.SignalWorkflowExecution(ctx, &historyservice.SignalWorkflowExecutionRequest{
 		NamespaceId:   namespaceID.String(),
 		SignalRequest: request,
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	return &workflowservice.SignalWorkflowExecutionResponse{}, nil
+	return &workflowservice.SignalWorkflowExecutionResponse{
+		Link: resp.GetLink(),
+	}, nil
 }
 
 // SignalWithStartWorkflowExecution is used to ensure sending signal to a workflow.
@@ -2364,8 +2366,10 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(ctx context.Context,
 	}
 
 	return &workflowservice.SignalWithStartWorkflowExecutionResponse{
-		RunId:   resp.GetRunId(),
-		Started: resp.Started,
+		RunId:             resp.GetRunId(),
+		Started:           resp.Started,
+		SignalLink:        resp.GetSignalLink(),
+		WorkflowStartLink: resp.GetWorkflowStartLink(),
 	}, nil
 }
 

service/frontend/workflow_handler.go

@@ -0,0 +1,50 @@
+package api
+
+import (
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/common"
+)
+
+// GenerateStartedEventRefLink builds a Link pointing to the WORKFLOW_EXECUTION_STARTED event.
+// Use this for backlinks to workflow start: the started event is always EventId=1 (FirstEventID)
+// and is never buffered, so a concrete EventReference is appropriate.
+func GenerateStartedEventRefLink(namespace, workflowID, runID string) *commonpb.Link {
+	return &commonpb.Link{
+		Variant: &commonpb.Link_WorkflowEvent_{
+			WorkflowEvent: &commonpb.Link_WorkflowEvent{
+				Namespace:  namespace,
+				WorkflowId: workflowID,
+				RunId:      runID,
+				Reference: &commonpb.Link_WorkflowEvent_EventRef{
+					EventRef: &commonpb.Link_WorkflowEvent_EventReference{
+						EventId:   common.FirstEventID,
+						EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
+					},
+				},
+			},
+		},
+	}
+}
+
+// GenerateRequestIdRefLink builds a Link with a RequestIdReference.
+// Use this for events that are buffered at signal time (e.g. SIGNALED), where the
+// concrete EventId is not yet known. The server resolves the RequestId to a real
+// EventId once the buffer flushes.
+func GenerateRequestIdRefLink(namespace, workflowID, runID, requestID string, eventType enumspb.EventType) *commonpb.Link {
+	return &commonpb.Link{
+		Variant: &commonpb.Link_WorkflowEvent_{
+			WorkflowEvent: &commonpb.Link_WorkflowEvent{
+				Namespace:  namespace,
+				WorkflowId: workflowID,
+				RunId:      runID,
+				Reference: &commonpb.Link_WorkflowEvent_RequestIdRef{
+					RequestIdRef: &commonpb.Link_WorkflowEvent_RequestIdReference{
+						RequestId: requestID,
+						EventType: eventType,
+					},
+				},
+			},
+		},
+	}
+}

service/history/api/link_util.go

@@ -10,7 +10,6 @@ import (
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
-	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/namespace"
@@ -338,20 +337,7 @@ func (uws *updateWithStart) updateWorkflow(
 		RunId:   currentWorkflowLease.GetContext().GetWorkflowKey().RunID,
 		Started: false, // set explicitly for emphasis
 		Status:  enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
-		Link: &commonpb.Link{
-			Variant: &commonpb.Link_WorkflowEvent_{
-				WorkflowEvent: &commonpb.Link_WorkflowEvent{
-					WorkflowId: wfKey.WorkflowID,
-					RunId:      wfKey.RunID,
-					Reference: &commonpb.Link_WorkflowEvent_EventRef{
-						EventRef: &commonpb.Link_WorkflowEvent_EventReference{
-							EventId:   common.FirstEventID,
-							EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
-						},
-					},
-				},
-			},
-		},
+		Link:    api.GenerateStartedEventRefLink("", wfKey.WorkflowID, wfKey.RunID),
 	}
 
 	return makeResponse(startResp, updateResp), nil

service/history/api/multioperation/api.go

@@ -99,8 +99,21 @@ func Invoke(
 		api.ReactivateVersionWorkflowIfPinned(ctx, namespaceEntry, request.GetVersioningOverride(), reactivationSignalCache, reactivationSignaler, shard.GetConfig().EnableVersionReactivationSignals())
 	}
 
+	swr := signalWithStartRequest.SignalWithStartRequest
 	return &historyservice.SignalWithStartWorkflowExecutionResponse{
 		RunId:   runID,
 		Started: started,
+		SignalLink: api.GenerateRequestIdRefLink(
+			swr.GetNamespace(),
+			swr.GetWorkflowId(),
+			runID,
+			swr.GetRequestId(),
+			enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED,
+		),
+		WorkflowStartLink: api.GenerateStartedEventRefLink(
+			swr.GetNamespace(),
+			swr.GetWorkflowId(),
+			runID,
+		),
 	}, nil
 }