Skip to content

Added the native stream protos. #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
moetemp wants to merge 1 commit into from
Closed

Added the native stream protos. #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
183 changes: 179 additions & 4 deletions openapi/openapiv2.json
View file
Open in desktop

Large diffs are not rendered by default.

197 changes: 197 additions & 0 deletions openapi/openapiv3.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14687,6 +14687,8 @@ components:
- EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
type: string
format: enum
version:
Expand Down Expand Up @@ -14856,6 +14858,10 @@ components:
$ref: '#/components/schemas/WorkflowNotificationChannelSubscribedEventAttributes'
workflowNotificationChannelUnsubscribedEventAttributes:
$ref: '#/components/schemas/WorkflowNotificationChannelUnsubscribedEventAttributes'
workflowStreamSubscribedEventAttributes:
$ref: '#/components/schemas/WorkflowStreamSubscribedEventAttributes'
workflowStreamRecordsAppendedEventAttributes:
$ref: '#/components/schemas/WorkflowStreamRecordsAppendedEventAttributes'
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.
Expand Down Expand Up @@ -16750,6 +16756,13 @@ components:
2. Try to assign the next poll to a group without any pending polls,
3. If every group has some pending polls, assign the next poll to a group randomly
according to the weights.
streamSlices:
type: array
items:
$ref: '#/components/schemas/StreamSlice'
description: |-
Stream data attached to this task. Delivered out of band so the payloads
never enter History; only the offset ranges are recorded there.
PollerGroupInfo:
type: object
properties:
Expand Down Expand Up @@ -17458,6 +17471,8 @@ components:
- EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
type: string
description: The event type of the history event generated by the request.
format: enum
Expand Down Expand Up @@ -19609,6 +19624,124 @@ components:
type:
type: string
description: The type of the driver, required.
StreamRange:
type: object
properties:
streamId:
type: string
description: |-
The stream, as the subscribing command addressed it: either the name of
a stream the consuming Workflow owns or the id of one in another
execution.
fromOffset:
type: string
description: |-
Inclusive.
(-- api-linter: core::0140::prepositions=disabled
aip.dev/not-precedent: "from" and "to" name a half-open offset range. --)
toOffset:
type: string
description: |-
Exclusive.
(-- api-linter: core::0140::prepositions=disabled
aip.dev/not-precedent: "from" and "to" name a half-open offset range. --)
description: |-
The offsets a Workflow Task consumed, without the payloads. Recorded on
WorkflowTaskCompleted so History grows with Workflow Tasks rather than with
records.
StreamRecord:
type: object
properties:
body:
allOf:
- $ref: '#/components/schemas/Payload'
description: |-
The value the producer published, stored as sent.

A payload codec applies on the paths this API owns: the append command on
RespondWorkflowTaskCompleted, and the slices on PollWorkflowTaskQueue.
Records a producer writes or reads through the stream service take a
different path, whose messages are not part of this API yet and so are
outside what a codec-applying proxy walks.
metadata:
type: object
additionalProperties:
$ref: '#/components/schemas/Payload'
description: Producer-supplied provenance, stored as sent.
topic:
type: string
description: Producer-supplied grouping label, stored as sent.
kind:
enum:
- STREAM_RECORD_KIND_UNSPECIFIED
- STREAM_RECORD_KIND_DATA
- STREAM_RECORD_KIND_FINISH
type: string
description: How to read this record. Unspecified is read as DATA.
format: enum
producerId:
type: string
description: Who wrote the record. Empty when the owning Workflow did.
attempt:
type: string
description: |-
The producer's attempt. Readers treat a later attempt by the same
producer as superseding what the earlier one wrote.
sequence:
type: string
description: |-
The producer's position within its attempt, zero when it does not number
its records. Stored as sent; the server does not assign, validate or
order by it, and the stream's own offsets are what order a read.
description: |-
One entry in a stream. The record is the wire format: stores keep it
serialized as is and readers in every language decode the same bytes.
StreamSlice:
type: object
properties:
streamId:
type: string
description: |-
The stream, as the subscribing command addressed it: either the name of
a stream the consuming Workflow owns or the id of one in another
execution.
runId:
type: string
description: |-
Run id of the execution that owns the stream. Set on both a slice for the
task being started and a re-supplied one.
fromOffset:
type: string
description: |-
Inclusive.
(-- api-linter: core::0140::prepositions=disabled
aip.dev/not-precedent: "from" and "to" name a half-open offset range. --)
toOffset:
type: string
description: |-
Exclusive. Equal to from_offset when the subscription observed nothing,
which is a fact replay has to reproduce rather than an absence of one.
(-- api-linter: core::0140::prepositions=disabled
aip.dev/not-precedent: "from" and "to" name a half-open offset range. --)
records:
type: array
items:
$ref: '#/components/schemas/StreamRecord'
workflowTaskCompletedEventId:
type: string
description: |-
The WorkflowTaskCompleted event whose consumed_stream_ranges recorded
this range. Set only when the server is re-supplying a range for a task
being replayed; a slice for the task now being started leaves it unset,
because the event closing that task does not exist yet.

Replay needs this because a Workflow Task response carries one slice set
while a cache miss replays every prior task, so the ranges have to be
matched to the events that recorded them rather than to the response.
description: |-
A contiguous range of a stream delivered to a Workflow Task, along with the
offsets it covers. The offsets are what History records; the records
themselves are never written to History.
StructuredCalendarSpec:
type: object
properties:
Expand Down Expand Up @@ -21764,6 +21897,8 @@ components:
- EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
type: string
format: enum
description: EventReference is a direct reference to a history event through the event ID.
Expand Down Expand Up @@ -21837,6 +21972,8 @@ components:
- EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
- EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
type: string
format: enum
description: RequestIdReference is a indirect reference to a history event through the request ID.
Expand Down Expand Up @@ -22998,6 +23135,54 @@ components:
* BETWEEN ... AND
STARTS_WITH
description: Activity trigger will be triggered when an activity is about to start.
WorkflowStreamRecordsAppendedEventAttributes:
type: object
properties:
workflowTaskCompletedEventId:
type: string
description: |-
The WorkflowTaskCompleted event of the task whose command appended this
batch.
streamId:
type: string
description: Name of the stream the Workflow appended to.
fromOffset:
type: string
description: |-
Inclusive. Same range vocabulary as StreamRange and StreamSlice, so a
reader does not have to remember which of the three counts and which
bounds.
(-- api-linter: core::0140::prepositions=disabled
aip.dev/not-precedent: "from" and "to" name a half-open offset range. --)
toOffset:
type: string
description: |-
Exclusive. With from_offset this names the range without carrying any of
it, which is what keeps this event a fixed size no matter how large the
batch or its payloads are.
(-- api-linter: core::0140::prepositions=disabled
aip.dev/not-precedent: "from" and "to" name a half-open offset range. --)
WorkflowStreamSubscribedEventAttributes:
type: object
properties:
workflowTaskCompletedEventId:
type: string
description: |-
The WorkflowTaskCompleted event of the task whose command created this
subscription.
streamId:
type: string
description: |-
The stream the Workflow subscribed to, as the command addressed it:
either the name of a stream this Workflow owns or the id of one in
another execution.
startOffset:
type: string
description: |-
The offset the subscription actually starts from. Resolved by the server
when the subscription is registered and recorded here, so replay reads
the resolved value rather than resolving it again against a stream that
has since moved.
WorkflowTaskCompletedEventAttributes:
type: object
properties:
Expand Down Expand Up @@ -23069,6 +23254,15 @@ components:
description: |-
The Worker Deployment Version that completed this task. Must be set if `versioning_behavior`
is set. This value updates workflow execution's `versioning_info.deployment_version`.
consumedStreamRanges:
type: array
items:
$ref: '#/components/schemas/StreamRange'
description: |-
Offset ranges this Workflow Task consumed from streams it subscribes to.
Recorded on every task where a subscription is active, including when it
observed nothing: an empty range is a fact replay must reproduce, and
omitting it would let replay deliver records the Workflow did not have.
WorkflowTaskCompletedMetadata:
type: object
properties:
Expand Down Expand Up @@ -23178,6 +23372,9 @@ components:
- 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
- WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES
- WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES
- WORKFLOW_TASK_FAILED_CAUSE_STREAM_RANGE_UNAVAILABLE
type: string
format: enum
failure:
Expand Down
43 changes: 43 additions & 0 deletions temporal/api/command/v1/message.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import "google/protobuf/duration.proto";
import "temporal/api/enums/v1/workflow.proto";
import "temporal/api/enums/v1/command_type.proto";
import "temporal/api/common/v1/message.proto";
import "temporal/api/stream/v1/message.proto";
import "temporal/api/failure/v1/message.proto";
import "temporal/api/taskqueue/v1/message.proto";
import "temporal/api/workflow/v1/message.proto";
Expand Down Expand Up @@ -328,6 +329,8 @@ message Command {
subscribe_notification_channel_command_attributes = 20;
UnsubscribeNotificationChannelCommandAttributes
unsubscribe_notification_channel_command_attributes = 21;
AppendStreamRecordsCommandAttributes append_stream_records_command_attributes = 22;
SubscribeStreamCommandAttributes subscribe_stream_command_attributes = 23;
}
}

Expand All @@ -347,3 +350,43 @@ message UnsubscribeNotificationChannelCommandAttributes {
// The channel to stop listening on, as the writers name it.
string channel = 1;
}

// Appends records to a stream the Workflow owns. Applied inside the Workflow
// Task's own commit. Produces one `WorkflowStreamRecordsAppended` event
// carrying the offset range and none of the payload; it schedules no further
// work.
message AppendStreamRecordsCommandAttributes {
// Name of a stream this Workflow owns, scoped to the Workflow. Created on
// first use. Empty means the Workflow's default output stream. A Workflow
// cannot append to a stream in another execution, so this is never the id
// of a standalone stream.
string stream_name = 1;
// Stored in order. The server sets `producer_id` to empty on each record,
// because the owning Workflow is the producer here.
repeated temporal.api.stream.v1.StreamRecord records = 2;
}

// Subscribe this Workflow to a stream, so later Workflow Tasks carry the ranges
// it has not consumed yet.
//
// The stream's addressing is resolved by the server rather than supplied here.
// A Workflow cannot look it up without doing I/O, and a value it carried would
// be a reading rather than a fact, so it could differ on replay.
message SubscribeStreamCommandAttributes {
// Stream to consume, named either way round: a stream this Workflow owns
// by the name it appends under, a stream in another execution by its id.
// The server tries them in that order, so a Workflow that owns a stream
// under this name cannot reach a standalone stream with the same id. When
// neither exists the Workflow gets a stream of its own by that name, which
// is how a reader subscribes before the first record is written.
string stream_name_or_id = 1;
// Where to start, as an absolute offset. Read only when `start_position`
// is unset. A negative value is refused: the head of the stream is asked
// for with `start_position.tail`.
int64 start_offset = 2;
// Where to start. The server resolves it once, when it registers the
// subscription, and records the resolved absolute offset on the subscribed
// event, so replay does not resolve it again. Setting it together with a
// non-zero `start_offset` fails the command.
temporal.api.stream.v1.StreamStartPosition start_position = 3;
}
2 changes: 2 additions & 0 deletions temporal/api/enums/v1/command_type.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,4 +31,6 @@ enum CommandType {
COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION = 18;
COMMAND_TYPE_SUBSCRIBE_NOTIFICATION_CHANNEL = 19;
COMMAND_TYPE_UNSUBSCRIBE_NOTIFICATION_CHANNEL = 20;
COMMAND_TYPE_APPEND_STREAM_RECORDS = 21;
COMMAND_TYPE_SUBSCRIBE_STREAM = 22;
}
8 changes: 8 additions & 0 deletions temporal/api/enums/v1/event_type.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,4 +182,12 @@ enum EventType {
// Recorded for every UnsubscribeNotificationChannel command, including one
// naming a channel the run was not subscribed to.
EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED = 62;
// A Workflow subscribed to a stream. Recorded once per subscription, not
// per record: the offsets a task consumed ride WorkflowTaskCompleted and
// the payloads never enter History at all.
EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED = 63;
// A Workflow appended a batch of records to a stream. Recorded per
// batch, and carrying only the offset range it landed at: the bodies go to
// the stream's own log, never into History.
EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED = 64;
}
8 changes: 8 additions & 0 deletions temporal/api/enums/v1/failed_cause.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,14 @@ enum WorkflowTaskFailedCause {
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;
// A workflow task completed with an invalid AppendStreamRecords command.
WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES = 43;
// A workflow task completed with an invalid SubscribeStream command.
WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES = 44;
// A workflow task could not be started because a stream range it consumed and recorded in
// History can no longer be served, for example after truncation or because it exceeds the
// replay bound. Check the workflow task failure message for more information.
WORKFLOW_TASK_FAILED_CAUSE_STREAM_RANGE_UNAVAILABLE = 45;
}

// Activity tasks can fail for various reasons. Note that some of these reasons can only originate
Expand Down
Loading
Loading