Added the workflow listener of a notification channel.

openapi/openapiv3.yaml

@@ -12680,6 +12680,52 @@ components:
       description: |-
         A listener of a channel: a Workflow Execution woken with a Workflow Task, or
          a callback the server invokes with each notification.
+    ChannelSubscriptionInfo:
+      type: object
+      properties:
+        channel:
+          type: string
+          description: Channel name.
+        kind:
+          enum:
+            - CHANNEL_KIND_UNSPECIFIED
+            - CHANNEL_KIND_INDEPENDENT
+            - CHANNEL_KIND_LINKED
+          type: string
+          description: |-
+            CHANNEL_KIND_INDEPENDENT for a channel the workflow subscribed to with a
+             SubscribeNotificationChannel command. CHANNEL_KIND_LINKED for a channel linked to this
+             workflow, which lists it once the channel holds any state.
+          format: enum
+        subscribedEventId:
+          type: string
+          description: |-
+            Independent kind: id of the WorkflowNotificationChannelSubscribed event that recorded the
+             subscription. Zero for the linked kind.
+        lastCounter:
+          type: string
+          description: Highest counter the workflow has accepted from the channel. Zero when none has arrived.
+        pendingNotification:
+          allOf:
+            - $ref: '#/components/schemas/Notification'
+          description: The notification held for the workflow's next Workflow Task, when one is pending.
+        scheduledCounter:
+          type: string
+          description: |-
+            Counter carried by the scheduled event of a Workflow Task that has not started yet. Zero
+             otherwise.
+        listenerCount:
+          type: integer
+          description: 'Linked kind: callback listeners registered on the channel.'
+          format: int32
+        retainedCount:
+          type: integer
+          description: 'Linked kind: notifications retained for pollers.'
+          format: int32
+        acceptedCount:
+          type: string
+          description: 'Linked kind: notifications the channel has accepted over its life.'
+      description: A workflow's standing on a notification channel, as reported by DescribeWorkflowExecution.
     ChildWorkflowExecutionCanceledEventAttributes:
       type: object
       properties:
@@ -13885,6 +13931,13 @@ components:
             $ref: '#/components/schemas/PendingNexusOperationInfo'
         workflowExtendedInfo:
           $ref: '#/components/schemas/WorkflowExecutionExtendedInfo'
+        channelSubscriptions:
+          type: array
+          items:
+            $ref: '#/components/schemas/ChannelSubscriptionInfo'
+          description: |-
+            The notification channels this run stands on: the independent channels it subscribed to and
+             the channels linked to it that hold any state. Empty when there are none.
     DescribeWorkflowRuleResponse:
       type: object
       properties:
@@ -14632,6 +14685,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
           type: string
           format: enum
         version:
@@ -14797,6 +14852,10 @@ components:
           $ref: '#/components/schemas/WorkflowExecutionUnpausedEventAttributes'
         workflowExecutionTimeSkippingTransitionedEventAttributes:
           $ref: '#/components/schemas/WorkflowExecutionTimeSkippingTransitionedEventAttributes'
+        workflowNotificationChannelSubscribedEventAttributes:
+          $ref: '#/components/schemas/WorkflowNotificationChannelSubscribedEventAttributes'
+        workflowNotificationChannelUnsubscribedEventAttributes:
+          $ref: '#/components/schemas/WorkflowNotificationChannelUnsubscribedEventAttributes'
       description: |-
         History events are the method by which Temporal SDKs advance (or recreate) workflow state.
          See the `EventType` enum for more info about what each event is for.
@@ -17397,6 +17456,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
           type: string
           description: The event type of the history event generated by the request.
           format: enum
@@ -21701,6 +21762,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
           type: string
           format: enum
       description: EventReference is a direct reference to a history event through the event ID.
@@ -21772,6 +21835,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
           type: string
           format: enum
       description: RequestIdReference is a indirect reference to a history event through the request ID.
@@ -22768,6 +22833,34 @@ components:
             The run that subscribed. The server follows a continue-as-new to the
              chain's current run when it delivers.
       description: A Workflow Execution listening on a channel.
+    WorkflowNotificationChannelSubscribedEventAttributes:
+      type: object
+      properties:
+        workflowTaskCompletedEventId:
+          type: string
+          description: |-
+            The WorkflowTaskCompleted event of the task whose command created this
+             subscription.
+        channel:
+          type: string
+          description: The channel the Workflow listens on for the rest of this run.
+    WorkflowNotificationChannelUnsubscribedEventAttributes:
+      type: object
+      properties:
+        workflowTaskCompletedEventId:
+          type: string
+          description: |-
+            The WorkflowTaskCompleted event of the task whose command ended this
+             subscription.
+        channel:
+          type: string
+          description: The channel the Workflow stopped listening on.
+        subscribedEventId:
+          type: string
+          description: |-
+            The WorkflowNotificationChannelSubscribed event that recorded the
+             subscription this command ended. Zero when the run held no subscription
+             for the channel.
     WorkflowPropertiesModifiedEventAttributes:
       type: object
       properties:
@@ -23083,6 +23176,8 @@ components:
             - WORKFLOW_TASK_FAILED_CAUSE_EXTERNAL_STORAGE_FAILURE
             - WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_PAUSE_REQUESTED_BEFORE_TASK_STARTED
             - WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE
+            - WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_NOTIFICATION_CHANNEL_ATTRIBUTES
+            - WORKFLOW_TASK_FAILED_CAUSE_BAD_UNSUBSCRIBE_NOTIFICATION_CHANNEL_ATTRIBUTES
           type: string
           format: enum
         failure:
@@ -23137,6 +23232,14 @@ components:
           type: integer
           description: Starting at 1, how many attempts there have been to complete this task
           format: int32
+        notifications:
+          type: array
+          items:
+            $ref: '#/components/schemas/Notification'
+          description: |-
+            Notifications for channels this Workflow listens to, folded per channel
+             since the last task was scheduled. In History so a Workflow may act on
+             them deterministically and replay sees the same.
     WorkflowTaskStartedEventAttributes:
       type: object
       properties:

temporal/api/command/v1/message.proto

@@ -324,5 +324,26 @@ message Command {
 
         ScheduleNexusOperationCommandAttributes schedule_nexus_operation_command_attributes = 18;
         RequestCancelNexusOperationCommandAttributes request_cancel_nexus_operation_command_attributes = 19;
+        SubscribeNotificationChannelCommandAttributes
+            subscribe_notification_channel_command_attributes = 20;
+        UnsubscribeNotificationChannelCommandAttributes
+            unsubscribe_notification_channel_command_attributes = 21;
     }
 }
+
+// Makes the Workflow a listener of a notification channel for this run. The
+// next notifications on the channel arrive on the scheduled event of a Workflow
+// Task. The subscription ends with the run, and a successor subscribes again.
+message SubscribeNotificationChannelCommandAttributes {
+    // The channel to listen on, as the writers name it.
+    string channel = 1;
+}
+
+// Ends the run's subscription to a notification channel. Notifications already
+// recorded on a scheduled event still reach that Workflow Task; later ones do
+// not. A command naming a channel the run is not subscribed to records its
+// event and changes nothing, so replay matches every command to an event.
+message UnsubscribeNotificationChannelCommandAttributes {
+    // The channel to stop listening on, as the writers name it.
+    string channel = 1;
+}

temporal/api/enums/v1/command_type.proto

@@ -29,4 +29,6 @@ enum CommandType {
     COMMAND_TYPE_MODIFY_WORKFLOW_PROPERTIES = 16;
     COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION = 17;
     COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION = 18;
+    COMMAND_TYPE_SUBSCRIBE_NOTIFICATION_CHANNEL = 19;
+    COMMAND_TYPE_UNSUBSCRIBE_NOTIFICATION_CHANNEL = 20;
 }

temporal/api/enums/v1/event_type.proto

@@ -175,4 +175,11 @@ enum EventType {
     EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED = 59;
     // An event that indicates time skipping advanced time or was disabled automatically after a bound was reached.
     EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED = 60;
+    // A Workflow became a listener of a notification channel for its run.
+    // The notifications themselves ride the WorkflowTaskScheduled event.
+    EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED = 61;
+    // A Workflow stopped listening on a notification channel for its run.
+    // Recorded for every UnsubscribeNotificationChannel command, including one
+    // naming a channel the run was not subscribed to.
+    EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED = 62;
 }

temporal/api/enums/v1/failed_cause.proto

@@ -90,6 +90,11 @@ enum WorkflowTaskFailedCause {
     WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_PAUSE_REQUESTED_BEFORE_TASK_STARTED = 39;
     // A workflow task failed because the request exceeded a size limit.
     WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE = 40;
+    // A SubscribeNotificationChannel command named an empty or too-long channel, or hit a
+    // subscription or listener limit.
+    WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_NOTIFICATION_CHANNEL_ATTRIBUTES = 41;
+    // An UnsubscribeNotificationChannel command named an empty or too-long channel.
+    WORKFLOW_TASK_FAILED_CAUSE_BAD_UNSUBSCRIBE_NOTIFICATION_CHANNEL_ATTRIBUTES = 42;
 }
 
 // Activity tasks can fail for various reasons. Note that some of these reasons can only originate

temporal/api/history/v1/message.proto

@@ -17,6 +17,7 @@ import "temporal/api/enums/v1/failed_cause.proto";
 import "temporal/api/enums/v1/update.proto";
 import "temporal/api/enums/v1/workflow.proto";
 import "temporal/api/common/v1/message.proto";
+import "temporal/api/notification/v1/message.proto";
 import "temporal/api/deployment/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
 import "temporal/api/taskqueue/v1/message.proto";
@@ -302,6 +303,10 @@ message WorkflowTaskScheduledEventAttributes {
     google.protobuf.Duration start_to_close_timeout = 2;
     // Starting at 1, how many attempts there have been to complete this task
     int32 attempt = 3;
+    // Notifications for channels this Workflow listens to, folded per channel
+    // since the last task was scheduled. In History so a Workflow may act on
+    // them deterministically and replay sees the same.
+    repeated temporal.api.notification.v1.Notification notifications = 4;
 }
 
 message WorkflowTaskStartedEventAttributes {
@@ -955,6 +960,26 @@ message ActivityPropertiesModifiedExternallyEventAttributes {
     temporal.api.common.v1.RetryPolicy new_retry_policy = 2;
 }
 
+message WorkflowNotificationChannelSubscribedEventAttributes {
+    // The WorkflowTaskCompleted event of the task whose command created this
+    // subscription.
+    int64 workflow_task_completed_event_id = 1;
+    // The channel the Workflow listens on for the rest of this run.
+    string channel = 2;
+}
+
+message WorkflowNotificationChannelUnsubscribedEventAttributes {
+    // The WorkflowTaskCompleted event of the task whose command ended this
+    // subscription.
+    int64 workflow_task_completed_event_id = 1;
+    // The channel the Workflow stopped listening on.
+    string channel = 2;
+    // The WorkflowNotificationChannelSubscribed event that recorded the
+    // subscription this command ended. Zero when the run held no subscription
+    // for the channel.
+    int64 subscribed_event_id = 3;
+}
+
 message WorkflowExecutionUpdateAcceptedEventAttributes {
     // The instance ID of the update protocol that generated this event.
     string protocol_instance_id = 1;
@@ -1278,6 +1303,10 @@ message HistoryEvent {
         WorkflowExecutionPausedEventAttributes workflow_execution_paused_event_attributes = 63;
         WorkflowExecutionUnpausedEventAttributes workflow_execution_unpaused_event_attributes = 64;
         WorkflowExecutionTimeSkippingTransitionedEventAttributes workflow_execution_time_skipping_transitioned_event_attributes = 65;
+        WorkflowNotificationChannelSubscribedEventAttributes
+            workflow_notification_channel_subscribed_event_attributes = 66;
+        WorkflowNotificationChannelUnsubscribedEventAttributes
+            workflow_notification_channel_unsubscribed_event_attributes = 67;
     }
 }
 

temporal/api/workflow/v1/message.proto

@@ -21,6 +21,7 @@ import "temporal/api/enums/v1/workflow.proto";
 import "temporal/api/common/v1/message.proto";
 import "temporal/api/deployment/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
+import "temporal/api/notification/v1/message.proto";
 import "temporal/api/taskqueue/v1/message.proto";
 import "temporal/api/sdk/v1/user_metadata.proto";
 
@@ -586,6 +587,32 @@ message NexusOperationCancellationInfo {
     string blocked_reason = 7;
 }
 
+// A workflow's standing on a notification channel, as reported by DescribeWorkflowExecution.
+message ChannelSubscriptionInfo {
+    // Channel name.
+    string channel = 1;
+    // CHANNEL_KIND_INDEPENDENT for a channel the workflow subscribed to with a
+    // SubscribeNotificationChannel command. CHANNEL_KIND_LINKED for a channel linked to this
+    // workflow, which lists it once the channel holds any state.
+    temporal.api.notification.v1.ChannelKind kind = 2;
+    // Independent kind: id of the WorkflowNotificationChannelSubscribed event that recorded the
+    // subscription. Zero for the linked kind.
+    int64 subscribed_event_id = 3;
+    // Highest counter the workflow has accepted from the channel. Zero when none has arrived.
+    int64 last_counter = 4;
+    // The notification held for the workflow's next Workflow Task, when one is pending.
+    temporal.api.notification.v1.Notification pending_notification = 5;
+    // Counter carried by the scheduled event of a Workflow Task that has not started yet. Zero
+    // otherwise.
+    int64 scheduled_counter = 6;
+    // Linked kind: callback listeners registered on the channel.
+    int32 listener_count = 7;
+    // Linked kind: notifications retained for pollers.
+    int32 retained_count = 8;
+    // Linked kind: notifications the channel has accepted over its life.
+    int64 accepted_count = 9;
+}
+
 message WorkflowExecutionOptions {
     // If set, takes precedence over the Versioning Behavior sent by the SDK on Workflow Task completion.
     VersioningOverride versioning_override = 1;

temporal/api/workflowservice/v1/request_response.proto

@@ -1319,6 +1319,9 @@ message DescribeWorkflowExecutionResponse {
     repeated temporal.api.workflow.v1.CallbackInfo callbacks = 6;
     repeated temporal.api.workflow.v1.PendingNexusOperationInfo pending_nexus_operations = 7;
     temporal.api.workflow.v1.WorkflowExecutionExtendedInfo workflow_extended_info = 8;
+    // The notification channels this run stands on: the independent channels it subscribed to and
+    // the channels linked to it that hold any state. Empty when there are none.
+    repeated temporal.api.workflow.v1.ChannelSubscriptionInfo channel_subscriptions = 9;
 }
 
 // (-- api-linter: core::0203::optional=disabled