Skip to content

feat(brook): add the Stream service messages (as experimental) #32

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

feat(brook): add the Stream service messages (as experimental) #32

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
61 changes: 61 additions & 0 deletions temporal/api/enums/v1/stream.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
syntax = "proto3";

package temporal.api.enums.v1;

option go_package = "go.temporal.io/api/enums/v1;enums";
option java_package = "io.temporal.api.enums.v1";
option java_multiple_files = true;
option java_outer_classname = "StreamProto";
option ruby_package = "Temporalio::Api::Enums::V1";
option csharp_namespace = "Temporalio.Api.Enums.V1";

// Experimental. Might change in the future.
enum StreamProducerKind {
STREAM_PRODUCER_KIND_UNSPECIFIED = 0;
STREAM_PRODUCER_KIND_WORKFLOW = 1;
STREAM_PRODUCER_KIND_LOCAL_ACTIVITY = 2;
STREAM_PRODUCER_KIND_ACTIVITY = 3;
STREAM_PRODUCER_KIND_CLIENT = 4;
STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY = 5;
}

// Experimental. Might change in the future.
enum StreamRecordKind {
STREAM_RECORD_KIND_UNSPECIFIED = 0;
STREAM_RECORD_KIND_DATA = 1;
// Lua flags the first batch of a higher epoch. The read makes this Record from the flag.
STREAM_RECORD_KIND_SUPERSEDED = 2;
STREAM_RECORD_KIND_CLOSE = 3;
}

// Experimental. Might change in the future.
// How a Stream closed, so a reader tells a terminated or timed out owner from a failed one
// without parsing the failure.
// (-- api-linter: core::0216::synonyms=disabled
// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --)
enum StreamCloseStatus {
STREAM_CLOSE_STATUS_UNSPECIFIED = 0;
STREAM_CLOSE_STATUS_COMPLETED = 1;
STREAM_CLOSE_STATUS_FAILED = 2;
STREAM_CLOSE_STATUS_CANCELED = 3;
// TERMINATED and TIMED_OUT come only from the owner's end, never from a caller.
STREAM_CLOSE_STATUS_TERMINATED = 4;
STREAM_CLOSE_STATUS_TIMED_OUT = 5;
}

// Experimental. Might change in the future.
// (-- api-linter: core::0216::synonyms=disabled
// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --)
enum StreamStatus {
STREAM_STATUS_UNSPECIFIED = 0;
STREAM_STATUS_OPEN = 1;
STREAM_STATUS_CLOSED = 2;
STREAM_STATUS_EXPIRED = 3;
}

// Experimental. Might change in the future.
enum StreamReadStart {
STREAM_READ_START_UNSPECIFIED = 0;
STREAM_READ_START_BEGINNING = 1;
STREAM_READ_START_END = 2;
}
111 changes: 111 additions & 0 deletions temporal/api/stream/v1/message.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
syntax = "proto3";

package temporal.api.stream.v1;

option go_package = "go.temporal.io/api/stream/v1;stream";
option java_package = "io.temporal.api.stream.v1";
option java_multiple_files = true;
option java_outer_classname = "MessageProto";
option ruby_package = "Temporalio::Api::Stream::V1";
option csharp_namespace = "Temporalio.Api.Stream.V1";

import "google/protobuf/timestamp.proto";

import "temporal/api/common/v1/message.proto";
import "temporal/api/enums/v1/stream.proto";
import "temporal/api/failure/v1/message.proto";

// Experimental. Might change in the future.
// Identifies one Stream. The id is opaque to callers.
message StreamRef {
string id = 1;
// Unset for a client's Stream. For a Workflow, business_id is the Workflow ID and run_id is
// the first execution run id of its run chain. For a Standalone Activity, business_id is the
// activity ID and run_id is its run id. A Store keys on id, and reads only a Workflow owner's
// business_id, to check that a close's ref names the owner's run chain.
temporal.api.common.v1.Execution owner = 2;
// The name of the backend that holds the Stream, as the Worker and client config register
// it, so every process routes a ref it receives. Empty is the default backend. The Store
// never reads it.
string backend = 3;
}

// Experimental. Might change in the future.
// The ordered attempt of a Producer. Compared in this order: task_started_event_id,
// task_attempt, task_started_time, attempt_scheduled_time, attempt. A field that a kind does
// not use is empty and compares as zero. A timestamp compares seconds, then nanos.
message ProducerEpoch {
// The Workflow Task of a Workflow, or the one that dispatched a Local Activity.
int64 task_started_event_id = 1;
int32 task_attempt = 2;
// Tells a reused started event id apart, for example after a dropped speculative task.
google.protobuf.Timestamp task_started_time = 5;
// When the server scheduled an Activity's attempt. A reset makes it later.
google.protobuf.Timestamp attempt_scheduled_time = 3;
// Activity, Local Activity and client attempt.
int32 attempt = 4;
}

// Experimental. Might change in the future.
message Producer {
temporal.api.enums.v1.StreamProducerKind kind = 1;
string id = 2;
ProducerEpoch epoch = 3;
// The run of the Producer, so nobody parses the id. Empty for a client.
string run_id = 4;
}

// Experimental. Might change in the future.
message WorkflowTaskEpoch {
int64 started_event_id = 1;
int32 attempt = 2;
google.protobuf.Timestamp started_time = 3;
}

// Experimental. Might change in the future.
message AppendRecord {
// Empty is the default Topic.
string topic = 1;
temporal.api.common.v1.Payload body = 2;
}

// Experimental. Might change in the future.
message Supersession {
temporal.api.enums.v1.StreamProducerKind producer_kind = 1;
string producer_id = 2;
ProducerEpoch epoch = 3;
// Workflow only.
WorkflowTaskEpoch previous_completed_task = 4;
repeated int64 failed_tasks = 5;
// Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn.
string withdraw_run = 6;
// The run of the Producer, so nobody parses the id. Empty for a client.
string run_id = 7;
}

// Experimental. Might change in the future.
// An offset is the Redis entry id packed as ms << 20 | seq. It is ordered and unique in one
// Stream, and it is sparse. Offsets have gaps, so they don't count Records. Zero is never an
// offset. A read names its Stream, so the read checks the offset against that Stream.
message StreamRecord {
temporal.api.enums.v1.StreamRecordKind kind = 1;
int64 offset = 2;
string topic = 3;
Producer producer = 4;
temporal.api.common.v1.Payload body = 5;
Supersession supersession = 6;
temporal.api.common.v1.Payload close_result = 7;
temporal.api.failure.v1.Failure close_failure = 8;
// CLOSE only. The result or the failure stays next to it.
temporal.api.enums.v1.StreamCloseStatus close_status = 9;
}

// Experimental. Might change in the future.
// The details of a failed Stream call, in the gRPC status details and in the details of the
// application failure of a Nexus operation.
message StreamErrorDetails {
// A token such as `stream_closed`, `stream_fenced` or `stream_format_too_new`.
// `seq_divergent` is a resend with the sequence of the last batch and other content.
// `stream_not_owner` is a close or a Producer that the Stream's owner doesn't allow.
string reason = 1;
}
177 changes: 177 additions & 0 deletions temporal/api/streamservice/v1/request_response.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,177 @@
syntax = "proto3";

package temporal.api.streamservice.v1;

option go_package = "go.temporal.io/api/streamservice/v1;streamservice";
option java_package = "io.temporal.api.streamservice.v1";
option java_multiple_files = true;
option java_outer_classname = "RequestResponseProto";
option ruby_package = "Temporalio::Api::StreamService::V1";
option csharp_namespace = "Temporalio.Api.StreamService.V1";

// The messages of the Stream service. An SDK Store (Redis or memory) and the server answer the
// same five operations with them. A field that only the server reads says so.

import "google/protobuf/duration.proto";

import "temporal/api/common/v1/message.proto";
import "temporal/api/enums/v1/stream.proto";
import "temporal/api/failure/v1/message.proto";
import "temporal/api/stream/v1/message.proto";

// Experimental. Might change in the future.
// Idempotent for one id and one Retention. ALREADY_EXISTS for another Retention.
// (-- api-linter: core::0133=disabled
// aip.dev/not-precedent: The request names a Stream by namespace and stream_id. --)
message CreateStreamRequest {
string namespace = 1;
// A Store requires it. Outside Workflow code, the server makes an id when it is empty.
string stream_id = 2;
// The age after which a Record is dropped. Empty is the Store default.
google.protobuf.Duration retention = 3;
// Workflow only. The previous run on a retry run's first create, to withdraw its Records.
string withdraw_run = 4;
// The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch.
temporal.api.stream.v1.Producer owner = 6;
// Only the server reads it. The server closes the Stream with the outcome of the owner's run
// chain, terminate and timeout included. It needs owner, and a ref with a Workflow owner.
// (-- api-linter: core::0140::prepositions=disabled
// aip.dev/not-precedent: "with" ties the close to the owner's run chain. --)
bool close_with_run = 7;
// The ref of the Stream. A Workflow-owned Stream sets the ref's owner, which names the
// owner's run chain.
temporal.api.stream.v1.StreamRef ref = 8;
// Only the server reads it. The server checks that the token is the Activity's current
// attempt, so a caller can't put a Stream into a Standalone Activity it isn't.
bytes task_token = 9;
}

// Experimental. Might change in the future.
message CreateStreamResponse {
temporal.api.stream.v1.StreamRef stream = 1;
bool created = 2;
temporal.api.enums.v1.StreamStatus status = 3;
}

// Experimental. Might change in the future.
// FAILED_PRECONDITION if closed, fenced, or if the Stream's meta has a higher format version v
// (detail stream_format_too_new). NOT_FOUND if absent.
message AppendRecordsRequest {
string namespace = 1;
temporal.api.stream.v1.StreamRef stream = 2;
// Required outside a Workflow. In a Workflow the transport sets it.
temporal.api.stream.v1.Producer producer = 3;
// The batch number in this epoch. With the Producer and epoch it is the dedupe key.
int64 sequence = 4;
repeated temporal.api.stream.v1.AppendRecord records = 5;
// The run's last completed Workflow Task (started_event_id 0 before the first) and its failed
// tasks, from a Local Activity's start. The batch withdraws the run's Local Activity Records
// whose task never completed.
temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6;
repeated int64 failed_tasks = 7;
// Local Activity only. The previous run on a retry run's first batch, to withdraw its Records.
string withdraw_run = 8;
// Only the server reads it, on a Workflow Producer's batch. The batch withdraws no Workflow
// Task and keeps no task facts, but withdraw_run still applies.
bool defer_withdrawal = 9;
// Only the server reads it. The server checks the Producer against the Activity's current
// attempt, so nobody else writes as that Activity.
bytes task_token = 10;
}

// Experimental. Might change in the future.
message AppendRecordsResponse {
int64 first_offset = 1;
int64 last_offset = 2;
}

// Experimental. Might change in the future.
message CloseStreamRequest {
string namespace = 1;
temporal.api.stream.v1.StreamRef stream = 2;
temporal.api.stream.v1.Producer producer = 3;
temporal.api.common.v1.Payload result = 4;
temporal.api.failure.v1.Failure failure = 5;
// Only the server reads these. They are the run's task facts, as on the append.
temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6;
repeated int64 failed_tasks = 7;
// Only the server reads it. The previous run on a retry run's close, to withdraw its Records.
string withdraw_run = 8;
// Empty is COMPLETED with no failure and FAILED with one. A caller sets CANCELED itself.
temporal.api.enums.v1.StreamCloseStatus status = 9;
// Only the server reads it. It checks the token as on the append.
bytes task_token = 10;
}

// Experimental. Might change in the future.
message CloseStreamResponse {
}

// Experimental. Might change in the future.
message ReadRecordsRequest {
string namespace = 1;
temporal.api.stream.v1.StreamRef stream = 2;
// Start at this offset. Zero uses start. INVALID_ARGUMENT if past the newest offset + 1.
// (-- api-linter: core::0140::prepositions=disabled
// aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --)
int64 from_offset = 3;
temporal.api.enums.v1.StreamReadStart start = 4;
// Empty reads every Topic in one order.
repeated string topics = 5;
int32 max_records = 6;
google.protobuf.Duration wait = 7;
// How long to wait for a Stream that nobody created yet. Empty fails at once with
// NOT_FOUND. At most 60 s.
google.protobuf.Duration create_wait = 8;
// Also read the Topics that start with one of these. A Record passes when its Topic is in
// topics or starts with a prefix. Both empty reads every Topic.
repeated string topic_prefixes = 10;
// The exclusive end of the read. The read scans no entry at or after it, the close Record
// included, so a replay reads the same range. Zero means no end. Only the server serves it,
// and a Store refuses it.
// (-- api-linter: core::0140::prepositions=disabled
// aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --)
int64 to_offset = 11;
// Park as a normal read, then answer with no Records and next_offset = from_offset, when a
// Record that passes the filter exists, when the Stream closes (done), or when wait ends.
// Its wait may be days, since a parked read adds nothing to History while it waits. Only the
// server serves it, and a Store refuses it.
bool wake_only = 12;
// Read the newest `last` Records that pass the filter, oldest first. It needs no from_offset
// and no start. Not with wake_only, and not in Workflow code, where replay needs a fixed range.
int32 last = 13;
// The most Record bytes in one page, so a page fits the caller's limit, such as a History
// marker. A larger Record still comes alone, so the read always moves on. Zero is no cap.
int64 max_bytes = 14;
// The lowest offset this run read on replay. A reset run copies the earlier run's reads into
// its History, so its floor must cover them too, not only its first live page. Only the
// server reads it.
// (-- api-linter: core::0140::prepositions=disabled
// aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --)
int64 floor_from = 15;
}

// Experimental. Might change in the future.
message ReadRecordsResponse {
repeated temporal.api.stream.v1.StreamRecord records = 1;
// The from_offset of the next read, which is the newest offset read + 1.
int64 next_offset = 2;
bool done = 3;
// This reader's floor keeps more than half of the Stream's byte cap. The writer fails when the
// cap fills, and only the reader's continue-as-new frees it. Only the server sets it.
bool continue_as_new_suggested = 4;
}

// Experimental. Might change in the future.
message DescribeStreamRequest {
string namespace = 1;
temporal.api.stream.v1.StreamRef stream = 2;
}

// Experimental. Might change in the future.
message DescribeStreamResponse {
temporal.api.enums.v1.StreamStatus status = 1;
// Zero when the Stream has no Record.
int64 last_offset = 2;
google.protobuf.Duration retention = 3;
}