Add support for NexusHandler callbacks

openapi/openapiv3.yaml

@@ -10826,6 +10826,8 @@ components:
           $ref: '#/components/schemas/Callback_Nexus'
         internal:
           $ref: '#/components/schemas/Callback_Internal'
+        nexusHandler:
+          $ref: '#/components/schemas/Callback_NexusHandler'
         links:
           type: array
           items:
@@ -10878,6 +10880,17 @@ components:
         blockedReason:
           type: string
           description: If the state is BLOCKED, blocked reason provides additional information.
+        requestId:
+          type: string
+          description: |-
+            Server-generated request ID used as an idempotency token when invoking callbacks.
+             It has no relation to caller-side request_id sent in operations like StartNexusOperationExecutionRequest.
+        failure:
+          allOf:
+            - $ref: '#/components/schemas/Failure'
+          description: |-
+            The failure if the callback was not able to complete successfully. e.g. timed out, received an
+             unretriable error, etc.
       description: Common callback information. Specific CallbackInfo messages should embed this and may include additional fields.
     Callback_Internal:
       type: object
@@ -10902,6 +10915,45 @@ components:
           additionalProperties:
             type: string
           description: Header to attach to callback request.
+      description: "Nexus callbacks are used to delivery Nexus operation completions, as defined in the Nexus RPC spec: \n https://github.com/nexus-rpc/api/blob/main/SPEC.md#callback-urls"
+    Callback_NexusHandler:
+      type: object
+      properties:
+        taskQueueName:
+          type: string
+          description: |-
+            Nexus task queue the Temporal worker is listening on.
+
+             NOTE: This is not a temporal.api.taskqueue.v1.TaskQueue to avoid a circular dependency.
+        service:
+          type: string
+          description: Target Nexus service, e.g. "HTTPAdapter".
+        operation:
+          type: string
+          description: Target operation, e.g. "DeliverAsWebhook".
+        sourceContext:
+          allOf:
+            - $ref: '#/components/schemas/Payload'
+          description: |-
+            Arbitrary user-supplied data from the source operation's callsite. (As applicable, not all operations
+             support attaching context data.)
+
+             There are restrictions on the maxium payload size a single callback can carry, as well as the
+             total sum of all source context payloads attached to an execution. See dynamic configuration
+             settings: "callback.worker.sourceContext.maxSize", "callback.worker.sourceContext.aggregateMaxSize".
+      description: |-
+        NexusHandler callbacks are requests to invoke a specific shape of Nexus operation on a Temporal worker.
+         The specified Nexus operation must have the following:
+         - Input:  temporal.api.notificationservice.v1.OnCompleteRequest
+         - Output: temporal.api.notificationservice.v1.OnCompleteResponse
+
+         The targeted Nexus service must be registered within the same namespace as the source operation
+         the callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks
+         are strictly caller-side.)
+
+         NexusHandler callbacks are only supported for certain types of operations, e.g. standalone Nexus operations.
+         Attempting to attach a Worker callback for an unsupported operation will result in an INVALID_ARGUMENT
+         error from the server.
     CanceledFailureInfo:
       type: object
       properties:
@@ -11886,6 +11938,13 @@ components:
           type: string
           description: Token for follow-on long-poll requests. Absent only if the operation is complete.
           format: bytes
+        completionCallbacks:
+          type: array
+          items:
+            $ref: '#/components/schemas/CallbackInfo'
+          description: |-
+            Completion callbacks to be invoked once the Nexus operation reaches a terminal state.
+             They will remain in the CALLBACK_STATE_STANDBY state until the Nexus operation is finished.
     DescribeScheduleResponse:
       type: object
       properties:
@@ -12355,15 +12414,14 @@ components:
             - EXECUTION_TYPE_UNSPECIFIED
             - EXECUTION_TYPE_WORKFLOW
             - EXECUTION_TYPE_ACTIVITY
+            - EXECUTION_TYPE_NEXUS_OPERATION
           type: string
           format: enum
         businessId:
           type: string
         runId:
           type: string
-      description: |-
-        Identifies a specific execution within a namespace. This is used for standalone activities
-         executions in batch jobs currently.
+      description: Identifies a specific execution within a namespace.
     ExternalWorkflowExecutionCancelRequestedEventAttributes:
       type: object
       properties:
@@ -13070,6 +13128,8 @@ components:
           $ref: '#/components/schemas/Link_NexusOperation'
         workflow:
           $ref: '#/components/schemas/Link_Workflow'
+        callback:
+          $ref: '#/components/schemas/Link_Callback'
       description: |-
         Link can be associated with history events. It might contain information about an external entity
          related to the history event. For example, workflow A makes a Nexus call that starts workflow B:
@@ -13094,6 +13154,25 @@ components:
         A link to a built-in batch job.
          Batch jobs can be used to perform operations on a set of workflows (e.g. terminate, signal, cancel, etc).
          This link can be put on workflow history events generated by actions taken by a batch job.
+    Link_Callback:
+      type: object
+      properties:
+        namespace:
+          type: string
+        execution:
+          $ref: '#/components/schemas/Execution'
+        componentId:
+          type: string
+          description: |-
+            In most cases, the Execution is sufficient to identify the callback's source. But for some types of execution,
+             a separate "component ID" is required. e.g. for completion callbacks attached to a workflow update. The
+             execution would be the workflow itself, with the component_id being the update ID of the update operation.
+        requestId:
+          type: string
+          description: Server-generate request ID sent when the callback was dispatched.
+      description: |-
+        A link to a worker callback attached to an execution. An execution (e.g. standalone Nexus operation) can have
+         multiple callbacks attached, and will be differentiated by the request_id used when the callback is invoked.
     Link_NexusOperation:
       type: object
       properties:
@@ -17330,6 +17409,13 @@ components:
             Defines how to resolve an operation id conflict with a *running* operation.
              The default policy is NEXUS_OPERATION_ID_CONFLICT_POLICY_FAIL.
           format: enum
+        onConflictOptions:
+          allOf:
+            - $ref: '#/components/schemas/OnConflictOptions'
+          description: |-
+            Defines actions to be done to the existing running standalone Nexus when the conflict policy
+             NEXUS_OPERATION_ID_CONFLICT_POLICY_USE_EXISTING is used. If not set or set to a empty object
+             (all options with default value), it will not modify the running operation.
         searchAttributes:
           allOf:
             - $ref: '#/components/schemas/SearchAttributes'
@@ -17349,6 +17435,18 @@ components:
           allOf:
             - $ref: '#/components/schemas/UserMetadata'
           description: Metadata for use by user interfaces to display the fixed as-of-start summary and details of the operation.
+        completionCallbacks:
+          type: array
+          items:
+            $ref: '#/components/schemas/Callback'
+          description: Completion callbacks to be invoked once the Nexus operation reaches a terminal state.
+        links:
+          type: array
+          items:
+            $ref: '#/components/schemas/Link'
+          description: |-
+            Links to be associated with the Nexus operation. Callbacks may also have associated links;
+             links already included with a callback should not be duplicated here.
     StartNexusOperationExecutionResponse:
       type: object
       properties:

temporal/api/callback/v1/message.proto

@@ -9,6 +9,7 @@ option java_outer_classname = "MessageProto";
 option ruby_package = "Temporalio::Api::Callback::V1";
 option csharp_namespace = "Temporalio.Api.Callback.V1";
 
+import "google/protobuf/empty.proto";
 import "google/protobuf/timestamp.proto";
 
 import "temporal/api/common/v1/message.proto";
@@ -34,4 +35,17 @@ message CallbackInfo {
     google.protobuf.Timestamp next_attempt_schedule_time = 7;
     // If the state is BLOCKED, blocked reason provides additional information.
     string blocked_reason = 8;
-}
+
+    // Server-generated request ID used as an idempotency token when invoking callbacks.
+    // It has no relation to caller-side request_id sent in operations like StartNexusOperationExecutionRequest.
+    string request_id = 9;
+
+    // Result of the callback's execution, only set when the callback reaches a terminal state.
+    oneof result {
+        // The callback completed successfully. (Which may include delivering a "failed" result successfully.)
+        google.protobuf.Empty success = 10;
+        // The failure if the callback was not able to complete successfully. e.g. timed out, received an
+        // unretriable error, etc.
+        temporal.api.failure.v1.Failure failure = 11;
+    }
+}

temporal/api/common/v1/message.proto

@@ -68,8 +68,7 @@ message WorkflowExecution {
     string run_id = 2;
 }
 
-// Identifies a specific execution within a namespace. This is used for standalone activities
-// executions in batch jobs currently.
+// Identifies a specific execution within a namespace.
 message Execution {
     temporal.api.enums.v1.ExecutionType type = 1;
     string business_id = 2;
@@ -185,6 +184,8 @@ message ResetOptions {
 
 // Callback to attach to various events in the system, e.g. workflow run completion.
 message Callback {
+    // Nexus callbacks are used to delivery Nexus operation completions, as defined in the Nexus RPC spec: 
+    // https://github.com/nexus-rpc/api/blob/main/SPEC.md#callback-urls
     message Nexus {
         // Callback URL.
         string url = 1;
@@ -201,10 +202,43 @@ message Callback {
         bytes data = 1;
     }
 
+    // NexusHandler callbacks are requests to invoke a specific shape of Nexus operation on a Temporal worker.
+    // The specified Nexus operation must have the following:
+    // - Input:  temporal.api.notificationservice.v1.OnCompleteRequest
+    // - Output: temporal.api.notificationservice.v1.OnCompleteResponse
+    //
+    // The targeted Nexus service must be registered within the same namespace as the source operation
+    // the callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks
+    // are strictly caller-side.)
+    //
+    // NexusHandler callbacks are only supported for certain types of operations, e.g. standalone Nexus operations.
+    // Attempting to attach a Worker callback for an unsupported operation will result in an INVALID_ARGUMENT
+    // error from the server.
+    message NexusHandler {
+        // Nexus task queue the Temporal worker is listening on.
+        //
+        // NOTE: This is not a temporal.api.taskqueue.v1.TaskQueue to avoid a circular dependency.
+        string task_queue_name = 1;
+
+        // Target Nexus service, e.g. "HTTPAdapter".
+        string service = 2;
+        // Target operation, e.g. "DeliverAsWebhook".
+        string operation = 3;
+
+        // Arbitrary user-supplied data from the source operation's callsite. (As applicable, not all operations
+        // support attaching context data.)
+        //
+        // There are restrictions on the maxium payload size a single callback can carry, as well as the
+        // total sum of all source context payloads attached to an execution. See dynamic configuration:
+        // "callback.nexusHandler.sourceContext.maxSize", "callback.nexusHandler.sourceContext.aggregateMaxSize".
+        temporal.api.common.v1.Payload source_context = 4;
+    }
+
     reserved 1; // For a generic callback mechanism to be added later.
     oneof variant {
         Nexus nexus = 2;
         Internal internal = 3;
+        NexusHandler nexus_handler = 4;
     }
 
     // Links associated with the callback. It can be used to link to underlying resources of the
@@ -273,12 +307,27 @@ message Link {
         string reason = 4;
     }
 
+    // A link to a worker callback attached to an execution. An execution (e.g. standalone Nexus operation) can have
+    // multiple callbacks attached, and will be differentiated by the request_id used when the callback is invoked.
+    message Callback {
+        string namespace = 1;
+        Execution execution = 2;
+        // In most cases, the Execution is sufficient to identify the callback's source. But for some types of execution,
+        // a separate "component ID" is required. e.g. for completion callbacks attached to a workflow update. The
+        // execution would be the workflow itself, with the component_id being the update ID of the update operation.
+        string component_id = 3;
+
+        // Server-generate request ID sent when the callback was dispatched.
+        string request_id  = 4;
+    }
+
     oneof variant {
         WorkflowEvent workflow_event = 1;
         BatchJob batch_job = 2;
         Activity activity = 3;
         NexusOperation nexus_operation = 4;
         Workflow workflow = 5;
+        Callback callback = 6;
     }
 }
 

temporal/api/enums/v1/common.proto

@@ -47,7 +47,7 @@ enum CallbackState {
     CALLBACK_STATE_FAILED = 4;
     // Callback has succeeded.
     CALLBACK_STATE_SUCCEEDED = 5;
-    // Callback is blocked (eg: by circuit breaker).
+    // Callback is blocked, e.g. by circuit breaker.
     CALLBACK_STATE_BLOCKED = 6;
 }
 
@@ -113,4 +113,6 @@ enum ExecutionType {
     EXECUTION_TYPE_WORKFLOW = 1;
     // An activity execution archetype. This is reserved for standalone activities.
     EXECUTION_TYPE_ACTIVITY = 2;
-}
+    // A Nexus operation execution archetype. This is reserved for standalone Nexus operations.
+    EXECUTION_TYPE_NEXUS_OPERATION = 3;
+}

temporal/api/nexusoperation/v1/message.proto

@@ -0,0 +1,42 @@
+syntax = "proto3";
+
+package temporal.api.nexusoperation.v1;
+
+option go_package = "go.temporal.io/api/nexusoperation/v1;nexusoperation";
+option java_package = "io.temporal.api.nexusoperation.v1";
+option java_multiple_files = true;
+option java_outer_classname = "MessageProto";
+option ruby_package = "Temporalio::Api::NexusOperation::V1";
+option csharp_namespace = "Temporalio.Api.NexusOperation.V1";
+
+import "temporal/api/callback/v1/message.proto";
+
+// CallbackInfo contains the state of a callback attached to a standalone Nexus operation.
+message CallbackInfo {
+    // Trigger for when the Nexus operation is completed, covering both success cases as
+    // well as any type of failure.
+    message OperationCompleted {}
+
+    message Trigger {
+        oneof variant {
+            OperationCompleted operation_completed = 1;
+        }
+    }
+
+    // Trigger for this callback.
+    Trigger trigger = 1;
+    // Common callback info.
+    temporal.api.callback.v1.CallbackInfo info = 2;
+}
+
+// When StartNexusOperationExecutionRequest uses the conflict policy NEXUS_OPERATION_ID_CONFLICT_POLICY_USE_EXISTING
+// and there is already an existing, running standalone Nexus operation, OnConflictOptions defines actions to be
+// taken.
+message OnConflictOptions {
+    // Attaches the request ID to the running operation.
+    bool attach_request_id = 1;
+    // Attaches the completion callbacks to the running operation.
+    bool attach_completion_callbacks = 2;
+    // Attaches any new links to the running operation.
+    bool attach_links = 3;
+}

temporal/api/notificationservice/v1/request_response.proto

@@ -0,0 +1,37 @@
+syntax = "proto3";
+
+package temporal.api.notificationservice.v1;
+
+option go_package = "go.temporal.io/api/notificationservice/v1;notificationservice";
+option java_package = "io.temporal.api.notificationservice.v1";
+option java_multiple_files = true;
+option java_outer_classname = "RequestResponseProto";
+option ruby_package = "Temporalio::Api::NotificationService::V1";
+option csharp_namespace = "Temporalio.Api.NotificationService.V1";
+
+import "temporal/api/common/v1/message.proto";
+import "temporal/api/failure/v1/message.proto";
+
+// OnCompleteRequest is the request type to the NotificationService's OnComplete operation,
+// allowing for defining completion handlers for arbitrary operations.
+//
+// Information about the source operation will be available in the form of a commonpb.Link,
+// which will be available separately from this OnCompleteRequest. e.g. a link to the source
+// standalone Nexus operation would be found in the nexuspb.StartOperationRequest parameter
+// sent to the worker callback. (In addition to this OnCompleteRequest.)
+message OnCompleteRequest {
+
+    // The result of the source operation.
+    oneof result {
+        // The operation was successful, and resulted in the given payload.
+        temporal.api.common.v1.Payload success = 1;
+        // The operation failed. Includes timeout, cancellation, and application errors.
+        temporal.api.failure.v1.Failure failure = 2;
+    }
+
+    // User-supplied data which was added to the source invocation. (As applicable.)
+    temporal.api.common.v1.Payload source_context = 3;
+}
+
+// OnCompleteResponse is the return type of the OnComplete operation.
+message OnCompleteResponse {}