diff --git a/chasm/lib/stream/config.go b/chasm/lib/stream/config.go new file mode 100644 index 0000000000..f4d1008313 --- /dev/null +++ b/chasm/lib/stream/config.go @@ -0,0 +1,248 @@ +package stream + +import ( + "time" + + "go.temporal.io/server/common/dynamicconfig" +) + +// DefaultMaxMessagesPerPoll bounds a read page when the caller does not. +const DefaultMaxMessagesPerPoll = 1000 + +// MaxRecordsPerBatch bounds one append. It is not only an admission limit: a +// batch is keyed by its first offset, so to serve a read starting inside a +// batch the reader has to find the batch that contains it. Bounding the batch +// bounds how far back it has to look, which turns an unbounded scan into a +// fixed overread. +const MaxRecordsPerBatch = 1000 + +// LongPollTimeout matches the convention used by the history long polls: on +// expiry the caller gets an empty response and polls again, rather than an +// error it would have to special-case. +const LongPollTimeout = 20 * time.Second + +// LongPollBuffer leaves room to return an empty response before the caller's +// own deadline fires. +const LongPollBuffer = 3 * time.Second + +// RoutedCallTimeout bounds a call to another shard made while a workflow's lock +// is held. The request's own deadline can be far longer, and a slow stream +// shard would otherwise stretch the lock hold to match it. +const RoutedCallTimeout = 5 * time.Second + +// MaxListPageSize bounds a visibility page when the caller does not. +const MaxListPageSize = 1000 + +// MaxStreamNameLength bounds a name before it becomes a map key in mutable +// state. The name comes from the caller, and any caller in the namespace can +// pick a new one. +const MaxStreamNameLength = 255 + +// The limits below bound resource use and are namespace-scoped dynamic config, +// with these values as the defaults. They stay as constants too so component +// code driven without a config, as the unit tests do, has something to fall +// back on. +const ( + // MaxConsumeItemsPerTask bounds one Workflow Task's slice. A byte cap alone + // is not enough: a burst of tiny records stays under it while still making + // one task's drain arbitrarily long. Whichever bound binds first, the rest + // is delivered on the following task. + MaxConsumeItemsPerTask = 1000 + + // MaxConsumeBytesPerTask bounds one Workflow Task's slice by size. Paired + // with MaxConsumeItemsPerTask because neither bound alone is enough: a + // burst of tiny records slips under the byte budget, and a few large ones + // slip under the item count. + MaxConsumeBytesPerTask = 2 << 20 + + // MaxProducersPerStream bounds the per-producer dedup table. The table is + // part of the component state written on every append, so a caller that + // sends a fresh producer id per request would grow the state until the + // mutable-state size limit rejects every further append, leaving the + // stream unwritable for good. The bound turns that into a clear error on + // the offending call. + MaxProducersPerStream = 1000 + + // MaxConsumersPerStream bounds the registered consumer table for the same + // reason. Each consumer also holds a truncation floor, so an unbounded + // table would pin storage as well as grow state. + MaxConsumersPerStream = 1000 + + // MaxMessageBytes bounds one message. A message is never split, so this is + // also the smallest unit a reader can be asked to materialise. + MaxMessageBytes = 1 << 20 + + // MaxBatchBytes bounds one append. It is deliberately equal to + // MaxConsumeBytesPerTask: a batch is written as one node and read back + // whole, so a batch larger than a task's byte budget could never be + // delivered. + MaxBatchBytes = MaxConsumeBytesPerTask + + // MaxOwnedStreamsPerWorkflow bounds how many named streams one execution + // can carry. Each is a component in the workflow's mutable state, so an + // unbounded count grows that state until the size limit terminates the + // execution, and any caller in the namespace can name a new one. + MaxOwnedStreamsPerWorkflow = 100 + + // OwnedStreamMaxBytes is the byte budget of a stream a workflow owns. Its + // batches are the owning execution's mutable state, and the execution size + // limit terminates the workflow rather than refusing an append, so the + // stream refuses first. The sum over a workflow's streams is still bounded + // by that limit. + OwnedStreamMaxBytes = 2 << 20 + + // OwnedStreamMaxItems is the item budget of a stream a workflow owns, for + // the same reason: each batch is a node, and many small ones cost state + // that the byte budget alone does not see. + OwnedStreamMaxItems = 10_000 +) + +var ( + MaxConsumeItemsPerTaskSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxConsumeItemsPerTask", + MaxConsumeItemsPerTask, + `Most stream records one workflow task carries per subscription.`, + ) + MaxConsumeBytesPerTaskSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxConsumeBytesPerTask", + MaxConsumeBytesPerTask, + `Most stream record bytes one workflow task carries per subscription.`, + ) + MaxProducersPerStreamSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxProducersPerStream", + MaxProducersPerStream, + `Most producer ids a stream tracks for deduplication.`, + ) + MaxConsumersPerStreamSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxConsumersPerStream", + MaxConsumersPerStream, + `Most workflow consumers a stream registers.`, + ) + MaxMessageBytesSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxMessageBytes", + MaxMessageBytes, + `Largest single stream record accepted.`, + ) + MaxBatchBytesSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxBatchBytes", + MaxBatchBytes, + `Largest stream append accepted, summed over its records.`, + ) + MaxOwnedStreamsPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxOwnedStreamsPerWorkflow", + MaxOwnedStreamsPerWorkflow, + `Most named streams one workflow execution can own.`, + ) + OwnedStreamMaxBytesSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.ownedStreamMaxBytes", + OwnedStreamMaxBytes, + `Byte budget of a stream a workflow owns. Appends past it are refused. Keep it well +under limit.mutableStateSize.error, which would otherwise terminate the workflow.`, + ) + OwnedStreamMaxItemsSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.ownedStreamMaxItems", + OwnedStreamMaxItems, + `Message budget of a stream a workflow owns. Appends past it are refused.`, + ) + RetentionRecheckIntervalSetting = dynamicconfig.NewGlobalDurationSetting( + "stream.retentionRecheckInterval", + time.Minute, + `How long a closed stream past its retention waits before asking again whether the +consumers holding it are still running.`, + ) +) + +// Config holds the settings as live property functions. +type Config struct { + // The id length limit shared with workflow ids. A stream id becomes an + // execution's business id, and a stream name a key in mutable state. + MaxIDLength dynamicconfig.IntPropertyFn + RetentionRecheckInterval dynamicconfig.DurationPropertyFn + MaxConsumeItemsPerTask dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxConsumeBytesPerTask dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxProducersPerStream dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxConsumersPerStream dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxMessageBytes dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxBatchBytes dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxOwnedStreamsPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter + OwnedStreamMaxBytes dynamicconfig.IntPropertyFnWithNamespaceFilter + OwnedStreamMaxItems dynamicconfig.IntPropertyFnWithNamespaceFilter +} + +func NewConfig(dc *dynamicconfig.Collection) *Config { + return &Config{ + MaxIDLength: dynamicconfig.MaxIDLengthLimit.Get(dc), + RetentionRecheckInterval: RetentionRecheckIntervalSetting.Get(dc), + MaxConsumeItemsPerTask: MaxConsumeItemsPerTaskSetting.Get(dc), + MaxConsumeBytesPerTask: MaxConsumeBytesPerTaskSetting.Get(dc), + MaxProducersPerStream: MaxProducersPerStreamSetting.Get(dc), + MaxConsumersPerStream: MaxConsumersPerStreamSetting.Get(dc), + MaxMessageBytes: MaxMessageBytesSetting.Get(dc), + MaxBatchBytes: MaxBatchBytesSetting.Get(dc), + MaxOwnedStreamsPerWorkflow: MaxOwnedStreamsPerWorkflowSetting.Get(dc), + OwnedStreamMaxBytes: OwnedStreamMaxBytesSetting.Get(dc), + OwnedStreamMaxItems: OwnedStreamMaxItemsSetting.Get(dc), + } +} + +// Limits is one namespace's resolved limits, read once per request so a +// transition sees one consistent set. +type Limits struct { + MaxConsumeItemsPerTask int + MaxConsumeBytesPerTask int + MaxProducersPerStream int + MaxConsumersPerStream int + MaxMessageBytes int + MaxBatchBytes int + MaxOwnedStreamsPerWorkflow int + OwnedStreamMaxBytes int + OwnedStreamMaxItems int +} + +// LimitsFor resolves the limits for a namespace. A nil Config, which is what +// code driven without a service gets, resolves to the defaults. +func (c *Config) LimitsFor(namespaceName string) Limits { + if c == nil { + return DefaultLimits() + } + return Limits{ + MaxConsumeItemsPerTask: c.MaxConsumeItemsPerTask(namespaceName), + MaxConsumeBytesPerTask: c.MaxConsumeBytesPerTask(namespaceName), + MaxProducersPerStream: c.MaxProducersPerStream(namespaceName), + MaxConsumersPerStream: c.MaxConsumersPerStream(namespaceName), + MaxMessageBytes: c.MaxMessageBytes(namespaceName), + MaxBatchBytes: c.MaxBatchBytes(namespaceName), + MaxOwnedStreamsPerWorkflow: c.MaxOwnedStreamsPerWorkflow(namespaceName), + OwnedStreamMaxBytes: c.OwnedStreamMaxBytes(namespaceName), + OwnedStreamMaxItems: c.OwnedStreamMaxItems(namespaceName), + }.withDefaults() +} + +// DefaultLimits is the constant set above. +func DefaultLimits() Limits { + return Limits{}.withDefaults() +} + +// withDefaults fills any limit left at zero, so a zero Limits value means the +// defaults rather than a stream that accepts nothing. +// +// Zero therefore cannot be configured: setting one of these to 0 restores its +// default rather than turning the thing off. Switching streams off for a +// namespace is what the enablement setting is for. +func (l Limits) withDefaults() Limits { + fill := func(v *int, def int) { + if *v <= 0 { + *v = def + } + } + fill(&l.MaxConsumeItemsPerTask, MaxConsumeItemsPerTask) + fill(&l.MaxConsumeBytesPerTask, MaxConsumeBytesPerTask) + fill(&l.MaxProducersPerStream, MaxProducersPerStream) + fill(&l.MaxConsumersPerStream, MaxConsumersPerStream) + fill(&l.MaxMessageBytes, MaxMessageBytes) + fill(&l.MaxBatchBytes, MaxBatchBytes) + fill(&l.MaxOwnedStreamsPerWorkflow, MaxOwnedStreamsPerWorkflow) + fill(&l.OwnedStreamMaxBytes, OwnedStreamMaxBytes) + fill(&l.OwnedStreamMaxItems, OwnedStreamMaxItems) + return l +} diff --git a/chasm/lib/stream/cursor.go b/chasm/lib/stream/cursor.go new file mode 100644 index 0000000000..adb433ce70 --- /dev/null +++ b/chasm/lib/stream/cursor.go @@ -0,0 +1,141 @@ +package stream + +import ( + "go.temporal.io/api/serviceerror" + "go.temporal.io/server/chasm" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +// Cursor is a consuming workflow's position in a stream. +// +// It is a subcomponent of the consumer, not of the stream. That placement is +// the whole point: the consumer's mutable state and its History events commit +// in one transaction, so folding a delivered range into the cursor lands +// atomically with the event that records the range. Holding the cursor on the +// stream instead would make every advance a cross-execution write, and a crash +// between the two writes would either redeliver a range or skip it silently. +type Cursor struct { + chasm.UnimplementedComponent + + State *streampb.WorkflowStreamCursor +} + +type NewCursorRequest struct { + StreamID string + // External marks a stream in another execution, whose frontier this + // workflow is told about rather than reads. + External bool + + // Where to start reading. Resolving "from the tail" against the stream's + // head happens before this is called, so the value recorded here is already + // a fact rather than a reading that would differ on replay. + StartOffset int64 +} + +func NewCursor(_ chasm.MutableContext, req NewCursorRequest) (*Cursor, error) { + if req.StreamID == "" { + return nil, serviceerror.NewInvalidArgument("stream id is required") + } + if req.StartOffset < 0 { + return nil, serviceerror.NewInvalidArgument("start offset cannot be negative") + } + + return &Cursor{ + State: &streampb.WorkflowStreamCursor{ + StreamId: req.StreamID, + Offset: req.StartOffset, + StartOffset: req.StartOffset, + External: req.External, + KnownHead: req.StartOffset, + }, + }, nil +} + +// LifecycleState reports the cursor as running for as long as the workflow +// holding it exists. Deregistration is an explicit act, not a state the +// component reaches on its own. +func (c *Cursor) LifecycleState(_ chasm.Context) chasm.LifecycleState { + return chasm.LifecycleStateRunning +} + +// Offset is the next offset that has not yet been delivered and folded in. +func (c *Cursor) Offset() int64 { + return c.State.Offset +} + +func (c *Cursor) StreamID() string { + return c.State.StreamId +} + +// StagePending records the range attached to the workflow task now in flight. +// +// A redelivery overwrites whatever was staged before. That is safe because a +// range only becomes history when the task completes: if the previous task +// failed or timed out, nothing was recorded, so the replacement range is the +// first one the workflow will ever have observed at this point. +func (c *Cursor) StagePending(_ chasm.MutableContext, from int64, to int64) error { + if from < c.State.Offset { + return serviceerror.NewInvalidArgumentf( + "cannot deliver from offset %d, cursor is already at %d", from, c.State.Offset) + } + if to < from { + return serviceerror.NewInvalidArgumentf("range end %d precedes range start %d", to, from) + } + + c.State.PendingFrom = from + c.State.PendingTo = to + c.State.HasPending = true + return nil +} + +// Pending reports the staged range. The second result distinguishes "no task in +// flight" from "a task in flight that was given nothing", which are different +// facts: the latter must still be recorded. +func (c *Cursor) Pending() (from int64, to int64, ok bool) { + if !c.State.HasPending { + return 0, 0, false + } + return c.State.PendingFrom, c.State.PendingTo, true +} + +// Commit folds the staged range into the cursor and returns it for recording. +// Caller writes the returned range onto the event that closes the task, in the +// same transaction that persists this advance. +func (c *Cursor) Commit(_ chasm.MutableContext) (from int64, to int64, ok bool) { + if !c.State.HasPending { + return 0, 0, false + } + + from, to = c.State.PendingFrom, c.State.PendingTo + c.State.Offset = to + c.State.PendingFrom = 0 + c.State.PendingTo = 0 + c.State.HasPending = false + return from, to, true +} + +// IsExternal reports whether the stream lives in another execution. +func (c *Cursor) IsExternal() bool { + return c.State.External +} + +// KnownHead is the stream's frontier as last pushed to this workflow. +func (c *Cursor) KnownHead() int64 { + return c.State.KnownHead +} + +// AdvanceKnownHead moves the recorded frontier forward. It never moves back: a +// stale push arriving after a fresher one must not hide offsets already known +// to exist. +func (c *Cursor) AdvanceKnownHead(_ chasm.MutableContext, head int64) { + if head > c.State.KnownHead { + c.State.KnownHead = head + } +} + +// StartOffset is where this subscription began reading. Replay needs it to tell +// a consumer that has committed nothing apart from one whose recording events +// are simply not in the history page it was handed. +func (c *Cursor) StartOffset() int64 { + return c.State.StartOffset +} diff --git a/chasm/lib/stream/cursor_test.go b/chasm/lib/stream/cursor_test.go new file mode 100644 index 0000000000..deb2b95212 --- /dev/null +++ b/chasm/lib/stream/cursor_test.go @@ -0,0 +1,100 @@ +package stream + +import ( + "testing" + + "github.com/stretchr/testify/require" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +func newTestCursor(offset int64) *Cursor { + return &Cursor{ + State: &streampb.WorkflowStreamCursor{ + StreamId: "s-1", + Offset: offset, + }, + } +} + +func TestNewCursorRejectsIncompleteRequests(t *testing.T) { + cases := map[string]NewCursorRequest{ + "no stream id": {}, + "negative start": {StreamID: "s-1", StartOffset: -1}, + } + + for name, req := range cases { + t.Run(name, func(t *testing.T) { + _, err := NewCursor(nil, req) + require.Error(t, err) + }) + } +} + +func TestCursorCommitAdvancesAndClears(t *testing.T) { + c := newTestCursor(4) + + require.NoError(t, c.StagePending(nil, 4, 7)) + + from, to, ok := c.Pending() + require.True(t, ok) + require.Equal(t, int64(4), from) + require.Equal(t, int64(7), to) + + from, to, ok = c.Commit(nil) + require.True(t, ok) + require.Equal(t, int64(4), from) + require.Equal(t, int64(7), to) + require.Equal(t, int64(7), c.Offset()) + + _, _, ok = c.Pending() + require.False(t, ok, "a committed range must not be staged twice") + + _, _, ok = c.Commit(nil) + require.False(t, ok, "committing again must not re-record the range") +} + +// A task that observed nothing still has to be recorded, so an empty range is a +// pending range, not the absence of one. Replay depends on the difference. +func TestCursorTreatsAnEmptyRangeAsAFact(t *testing.T) { + c := newTestCursor(9) + + _, _, ok := c.Pending() + require.False(t, ok, "no task in flight yet") + + require.NoError(t, c.StagePending(nil, 9, 9)) + + from, to, ok := c.Pending() + require.True(t, ok, "a task given nothing is still a task that must be recorded") + require.Equal(t, from, to) + + from, to, ok = c.Commit(nil) + require.True(t, ok) + require.Equal(t, int64(9), from) + require.Equal(t, int64(9), to) + require.Equal(t, int64(9), c.Offset(), "an empty range must not move the cursor") +} + +func TestCursorRejectsARangeBehindItself(t *testing.T) { + c := newTestCursor(12) + + err := c.StagePending(nil, 11, 14) + require.ErrorContains(t, err, "cursor is already at 12") + + err = c.StagePending(nil, 12, 11) + require.ErrorContains(t, err, "precedes range start") +} + +// A task that failed recorded nothing, so the range it was given never became +// history and the replacement is free to differ. +func TestCursorRedeliveryReplacesTheStagedRange(t *testing.T) { + c := newTestCursor(2) + + require.NoError(t, c.StagePending(nil, 2, 5)) + require.NoError(t, c.StagePending(nil, 2, 9)) + + from, to, ok := c.Pending() + require.True(t, ok) + require.Equal(t, int64(2), from) + require.Equal(t, int64(9), to) + require.Equal(t, int64(2), c.Offset(), "staging alone must never advance the cursor") +} diff --git a/chasm/lib/stream/gen/streampb/v1/message.go-helpers.pb.go b/chasm/lib/stream/gen/streampb/v1/message.go-helpers.pb.go new file mode 100644 index 0000000000..ef69741081 --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/message.go-helpers.pb.go @@ -0,0 +1,80 @@ +// Code generated by protoc-gen-go-helpers. DO NOT EDIT. +package streampb + +import ( + "google.golang.org/protobuf/proto" +) + +// Marshal an object of type StreamRecord to the protobuf v3 wire format +func (val *StreamRecord) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamRecord from the protobuf v3 wire format +func (val *StreamRecord) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamRecord) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamRecord values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamRecord) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamRecord + switch t := that.(type) { + case *StreamRecord: + that1 = t + case StreamRecord: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type StreamRecordBatch to the protobuf v3 wire format +func (val *StreamRecordBatch) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamRecordBatch from the protobuf v3 wire format +func (val *StreamRecordBatch) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamRecordBatch) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamRecordBatch values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamRecordBatch) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamRecordBatch + switch t := that.(type) { + case *StreamRecordBatch: + that1 = t + case StreamRecordBatch: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} diff --git a/chasm/lib/stream/gen/streampb/v1/message.pb.go b/chasm/lib/stream/gen/streampb/v1/message.pb.go new file mode 100644 index 0000000000..bce3b3730a --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/message.pb.go @@ -0,0 +1,265 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// plugins: +// protoc-gen-go +// protoc +// source: temporal/server/chasm/lib/stream/proto/v1/message.proto + +package streampb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + v1 "go.temporal.io/api/common/v1" + v11 "go.temporal.io/api/stream/v1" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// The stored shape of temporal.api.stream.v1.StreamRecord plus the offset a +// read assigns. Field for field the public record, so one crosses the frontend +// in either direction without translation. +type StreamRecord struct { + state protoimpl.MessageState `protogen:"open.v1"` + Body *v1.Payload `protobuf:"bytes,1,opt,name=body,proto3" json:"body,omitempty"` + // Producer-supplied provenance, stored as sent. + Metadata map[string]*v1.Payload `protobuf:"bytes,2,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Topic string `protobuf:"bytes,3,opt,name=topic,proto3" json:"topic,omitempty"` + // The producer's position within its attempt, zero when it does not number + // its records. Stored as sent; the global offset is what orders the stream. + Sequence int64 `protobuf:"varint,4,opt,name=sequence,proto3" json:"sequence,omitempty"` + // Settled to DATA on append when left unspecified, so a retry hashes the + // same bytes and a reader never sees the zero value. + Kind v11.StreamRecordKind `protobuf:"varint,5,opt,name=kind,proto3,enum=temporal.api.stream.v1.StreamRecordKind" json:"kind,omitempty"` + // Position in the whole stream, set on read and never stored. A consumer + // that resumes at record granularity needs it, and a topic-filtered read + // leaves gaps that make it underivable from the response alone. + Offset int64 `protobuf:"varint,6,opt,name=offset,proto3" json:"offset,omitempty"` + // Who wrote the record. Empty when the owning Workflow did: its publish + // command clears the field, whatever the worker sent. + ProducerId string `protobuf:"bytes,7,opt,name=producer_id,json=producerId,proto3" json:"producer_id,omitempty"` + // The producer's attempt, stored as sent. Readers treat a later attempt by + // the same producer as superseding what the earlier one wrote. + Attempt int64 `protobuf:"varint,8,opt,name=attempt,proto3" json:"attempt,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamRecord) Reset() { + *x = StreamRecord{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_message_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamRecord) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamRecord) ProtoMessage() {} + +func (x *StreamRecord) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_message_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamRecord.ProtoReflect.Descriptor instead. +func (*StreamRecord) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_message_proto_rawDescGZIP(), []int{0} +} + +func (x *StreamRecord) GetBody() *v1.Payload { + if x != nil { + return x.Body + } + return nil +} + +func (x *StreamRecord) GetMetadata() map[string]*v1.Payload { + if x != nil { + return x.Metadata + } + return nil +} + +func (x *StreamRecord) GetTopic() string { + if x != nil { + return x.Topic + } + return "" +} + +func (x *StreamRecord) GetSequence() int64 { + if x != nil { + return x.Sequence + } + return 0 +} + +func (x *StreamRecord) GetKind() v11.StreamRecordKind { + if x != nil { + return x.Kind + } + return v11.StreamRecordKind(0) +} + +func (x *StreamRecord) GetOffset() int64 { + if x != nil { + return x.Offset + } + return 0 +} + +func (x *StreamRecord) GetProducerId() string { + if x != nil { + return x.ProducerId + } + return "" +} + +func (x *StreamRecord) GetAttempt() int64 { + if x != nil { + return x.Attempt + } + return 0 +} + +// One append is one batch, and one batch is one data node. The server stores +// this serialized and opaque; it decodes only to trim a partial first page or +// to apply a topic filter. +type StreamRecordBatch struct { + state protoimpl.MessageState `protogen:"open.v1"` + Records []*StreamRecord `protobuf:"bytes,1,rep,name=records,proto3" json:"records,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamRecordBatch) Reset() { + *x = StreamRecordBatch{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_message_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamRecordBatch) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamRecordBatch) ProtoMessage() {} + +func (x *StreamRecordBatch) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_message_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamRecordBatch.ProtoReflect.Descriptor instead. +func (*StreamRecordBatch) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_message_proto_rawDescGZIP(), []int{1} +} + +func (x *StreamRecordBatch) GetRecords() []*StreamRecord { + if x != nil { + return x.Records + } + return nil +} + +var File_temporal_server_chasm_lib_stream_proto_v1_message_proto protoreflect.FileDescriptor + +const file_temporal_server_chasm_lib_stream_proto_v1_message_proto_rawDesc = "" + + "\n" + + "7temporal/server/chasm/lib/stream/proto/v1/message.proto\x12)temporal.server.chasm.lib.stream.proto.v1\x1a$temporal/api/common/v1/message.proto\x1a$temporal/api/stream/v1/message.proto\"\xc7\x03\n" + + "\fStreamRecord\x123\n" + + "\x04body\x18\x01 \x01(\v2\x1f.temporal.api.common.v1.PayloadR\x04body\x12a\n" + + "\bmetadata\x18\x02 \x03(\v2E.temporal.server.chasm.lib.stream.proto.v1.StreamRecord.MetadataEntryR\bmetadata\x12\x14\n" + + "\x05topic\x18\x03 \x01(\tR\x05topic\x12\x1a\n" + + "\bsequence\x18\x04 \x01(\x03R\bsequence\x12<\n" + + "\x04kind\x18\x05 \x01(\x0e2(.temporal.api.stream.v1.StreamRecordKindR\x04kind\x12\x16\n" + + "\x06offset\x18\x06 \x01(\x03R\x06offset\x12\x1f\n" + + "\vproducer_id\x18\a \x01(\tR\n" + + "producerId\x12\x18\n" + + "\aattempt\x18\b \x01(\x03R\aattempt\x1a\\\n" + + "\rMetadataEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x125\n" + + "\x05value\x18\x02 \x01(\v2\x1f.temporal.api.common.v1.PayloadR\x05value:\x028\x01\"f\n" + + "\x11StreamRecordBatch\x12Q\n" + + "\arecords\x18\x01 \x03(\v27.temporal.server.chasm.lib.stream.proto.v1.StreamRecordR\arecordsB>Z temporal.api.common.v1.Payload + 2, // 1: temporal.server.chasm.lib.stream.proto.v1.StreamRecord.metadata:type_name -> temporal.server.chasm.lib.stream.proto.v1.StreamRecord.MetadataEntry + 4, // 2: temporal.server.chasm.lib.stream.proto.v1.StreamRecord.kind:type_name -> temporal.api.stream.v1.StreamRecordKind + 0, // 3: temporal.server.chasm.lib.stream.proto.v1.StreamRecordBatch.records:type_name -> temporal.server.chasm.lib.stream.proto.v1.StreamRecord + 3, // 4: temporal.server.chasm.lib.stream.proto.v1.StreamRecord.MetadataEntry.value:type_name -> temporal.api.common.v1.Payload + 5, // [5:5] is the sub-list for method output_type + 5, // [5:5] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name +} + +func init() { file_temporal_server_chasm_lib_stream_proto_v1_message_proto_init() } +func file_temporal_server_chasm_lib_stream_proto_v1_message_proto_init() { + if File_temporal_server_chasm_lib_stream_proto_v1_message_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_stream_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_stream_proto_v1_message_proto_rawDesc)), + NumEnums: 0, + NumMessages: 3, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_temporal_server_chasm_lib_stream_proto_v1_message_proto_goTypes, + DependencyIndexes: file_temporal_server_chasm_lib_stream_proto_v1_message_proto_depIdxs, + MessageInfos: file_temporal_server_chasm_lib_stream_proto_v1_message_proto_msgTypes, + }.Build() + File_temporal_server_chasm_lib_stream_proto_v1_message_proto = out.File + file_temporal_server_chasm_lib_stream_proto_v1_message_proto_goTypes = nil + file_temporal_server_chasm_lib_stream_proto_v1_message_proto_depIdxs = nil +} diff --git a/chasm/lib/stream/gen/streampb/v1/stream_state.go-helpers.pb.go b/chasm/lib/stream/gen/streampb/v1/stream_state.go-helpers.pb.go new file mode 100644 index 0000000000..c97d64015d --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/stream_state.go-helpers.pb.go @@ -0,0 +1,228 @@ +// Code generated by protoc-gen-go-helpers. DO NOT EDIT. +package streampb + +import ( + "google.golang.org/protobuf/proto" +) + +// Marshal an object of type StreamState to the protobuf v3 wire format +func (val *StreamState) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamState from the protobuf v3 wire format +func (val *StreamState) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamState) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamState values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamState) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamState + switch t := that.(type) { + case *StreamState: + that1 = t + case StreamState: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type StreamBudget to the protobuf v3 wire format +func (val *StreamBudget) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamBudget from the protobuf v3 wire format +func (val *StreamBudget) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamBudget) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamBudget values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamBudget) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamBudget + switch t := that.(type) { + case *StreamBudget: + that1 = t + case StreamBudget: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type ProducerCursor to the protobuf v3 wire format +func (val *ProducerCursor) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type ProducerCursor from the protobuf v3 wire format +func (val *ProducerCursor) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *ProducerCursor) Size() int { + return proto.Size(val) +} + +// Equal returns whether two ProducerCursor values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *ProducerCursor) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *ProducerCursor + switch t := that.(type) { + case *ProducerCursor: + that1 = t + case ProducerCursor: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type ConsumerCursor to the protobuf v3 wire format +func (val *ConsumerCursor) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type ConsumerCursor from the protobuf v3 wire format +func (val *ConsumerCursor) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *ConsumerCursor) Size() int { + return proto.Size(val) +} + +// Equal returns whether two ConsumerCursor values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *ConsumerCursor) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *ConsumerCursor + switch t := that.(type) { + case *ConsumerCursor: + that1 = t + case ConsumerCursor: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type WorkflowStreamCursor to the protobuf v3 wire format +func (val *WorkflowStreamCursor) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type WorkflowStreamCursor from the protobuf v3 wire format +func (val *WorkflowStreamCursor) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *WorkflowStreamCursor) Size() int { + return proto.Size(val) +} + +// Equal returns whether two WorkflowStreamCursor values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *WorkflowStreamCursor) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *WorkflowStreamCursor + switch t := that.(type) { + case *WorkflowStreamCursor: + that1 = t + case WorkflowStreamCursor: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type StreamLifecycle to the protobuf v3 wire format +func (val *StreamLifecycle) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamLifecycle from the protobuf v3 wire format +func (val *StreamLifecycle) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamLifecycle) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamLifecycle values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamLifecycle) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamLifecycle + switch t := that.(type) { + case *StreamLifecycle: + that1 = t + case StreamLifecycle: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} diff --git a/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go new file mode 100644 index 0000000000..223672a119 --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go @@ -0,0 +1,715 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// plugins: +// protoc-gen-go +// protoc +// source: temporal/server/chasm/lib/stream/proto/v1/stream_state.proto + +package streampb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + v1 "go.temporal.io/api/common/v1" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + durationpb "google.golang.org/protobuf/types/known/durationpb" + timestamppb "google.golang.org/protobuf/types/known/timestamppb" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Size is O(producers + consumers), never O(messages). Payload bytes live in +// the component's data nodes, not here, which keeps this off the CHASM +// partial-read path. +type StreamState struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Visibility frontier. Readers never observe an offset at or past this. + HeadOffset int64 `protobuf:"varint,1,opt,name=head_offset,json=headOffset,proto3" json:"head_offset,omitempty"` + // Truncation floor. Offsets below this are gone. + BaseOffset int64 `protobuf:"varint,2,opt,name=base_offset,json=baseOffset,proto3" json:"base_offset,omitempty"` + Closed bool `protobuf:"varint,4,opt,name=closed,proto3" json:"closed,omitempty"` + CloseReason *v1.Payload `protobuf:"bytes,5,opt,name=close_reason,json=closeReason,proto3" json:"close_reason,omitempty"` + Producers map[string]*ProducerCursor `protobuf:"bytes,9,rep,name=producers,proto3" json:"producers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Consumers map[string]*ConsumerCursor `protobuf:"bytes,10,rep,name=consumers,proto3" json:"consumers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Lifecycle *StreamLifecycle `protobuf:"bytes,11,opt,name=lifecycle,proto3" json:"lifecycle,omitempty"` + // Set when a successor run takes ownership, so an in-flight poll can follow + // the chain instead of stalling on a superseded run. + RedirectRunId string `protobuf:"bytes,12,opt,name=redirect_run_id,json=redirectRunId,proto3" json:"redirect_run_id,omitempty"` + // Wall-clock close time, used to schedule retention deletion. + CloseTime *timestamppb.Timestamp `protobuf:"bytes,13,opt,name=close_time,json=closeTime,proto3" json:"close_time,omitempty"` + // Set on a stream a workflow owns. Its batches are the owning execution's + // mutable state, so appends past the budget are refused rather than left to + // the execution size limit, which terminates the workflow. + Budget *StreamBudget `protobuf:"bytes,14,opt,name=budget,proto3" json:"budget,omitempty"` + // Bytes appended over the stream's life, kept for the budget check. A stream + // with a budget carries no lifecycle cap and is never truncated, so nothing + // reclaims behind it and this is also what it holds. The item side of the + // budget measures head minus base for the same reason: offsets are global + // and a stream can begin above zero. + AppendedBytes int64 `protobuf:"varint,15,opt,name=appended_bytes,json=appendedBytes,proto3" json:"appended_bytes,omitempty"` + // A notify task is scheduled and has not run yet. Appends while it is set + // schedule none of their own; the task reads the head when it runs, so it + // carries every append that landed before it. + NotifyPending bool `protobuf:"varint,16,opt,name=notify_pending,json=notifyPending,proto3" json:"notify_pending,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamState) Reset() { + *x = StreamState{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamState) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamState) ProtoMessage() {} + +func (x *StreamState) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamState.ProtoReflect.Descriptor instead. +func (*StreamState) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDescGZIP(), []int{0} +} + +func (x *StreamState) GetHeadOffset() int64 { + if x != nil { + return x.HeadOffset + } + return 0 +} + +func (x *StreamState) GetBaseOffset() int64 { + if x != nil { + return x.BaseOffset + } + return 0 +} + +func (x *StreamState) GetClosed() bool { + if x != nil { + return x.Closed + } + return false +} + +func (x *StreamState) GetCloseReason() *v1.Payload { + if x != nil { + return x.CloseReason + } + return nil +} + +func (x *StreamState) GetProducers() map[string]*ProducerCursor { + if x != nil { + return x.Producers + } + return nil +} + +func (x *StreamState) GetConsumers() map[string]*ConsumerCursor { + if x != nil { + return x.Consumers + } + return nil +} + +func (x *StreamState) GetLifecycle() *StreamLifecycle { + if x != nil { + return x.Lifecycle + } + return nil +} + +func (x *StreamState) GetRedirectRunId() string { + if x != nil { + return x.RedirectRunId + } + return "" +} + +func (x *StreamState) GetCloseTime() *timestamppb.Timestamp { + if x != nil { + return x.CloseTime + } + return nil +} + +func (x *StreamState) GetBudget() *StreamBudget { + if x != nil { + return x.Budget + } + return nil +} + +func (x *StreamState) GetAppendedBytes() int64 { + if x != nil { + return x.AppendedBytes + } + return 0 +} + +func (x *StreamState) GetNotifyPending() bool { + if x != nil { + return x.NotifyPending + } + return false +} + +// Hard bounds on what a stream may hold. Distinct from StreamLifecycle.max_items, +// which reclaims the oldest messages: a budget refuses the newest. +type StreamBudget struct { + state protoimpl.MessageState `protogen:"open.v1"` + MaxItems int64 `protobuf:"varint,1,opt,name=max_items,json=maxItems,proto3" json:"max_items,omitempty"` + MaxBytes int64 `protobuf:"varint,2,opt,name=max_bytes,json=maxBytes,proto3" json:"max_bytes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamBudget) Reset() { + *x = StreamBudget{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamBudget) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamBudget) ProtoMessage() {} + +func (x *StreamBudget) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamBudget.ProtoReflect.Descriptor instead. +func (*StreamBudget) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDescGZIP(), []int{1} +} + +func (x *StreamBudget) GetMaxItems() int64 { + if x != nil { + return x.MaxItems + } + return 0 +} + +func (x *StreamBudget) GetMaxBytes() int64 { + if x != nil { + return x.MaxBytes + } + return 0 +} + +type ProducerCursor struct { + state protoimpl.MessageState `protogen:"open.v1"` + Seq int64 `protobuf:"varint,1,opt,name=seq,proto3" json:"seq,omitempty"` + FirstOffset int64 `protobuf:"varint,2,opt,name=first_offset,json=firstOffset,proto3" json:"first_offset,omitempty"` + Count int64 `protobuf:"varint,3,opt,name=count,proto3" json:"count,omitempty"` + // Distinguishes a genuine retry from a client reusing a sequence with + // different content, which must be rejected rather than deduplicated. + ContentHash []byte `protobuf:"bytes,4,opt,name=content_hash,json=contentHash,proto3" json:"content_hash,omitempty"` + // Set by FinishWriting. Ends this producer's writes without closing the + // stream for anyone else. + Fenced bool `protobuf:"varint,5,opt,name=fenced,proto3" json:"fenced,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ProducerCursor) Reset() { + *x = ProducerCursor{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ProducerCursor) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ProducerCursor) ProtoMessage() {} + +func (x *ProducerCursor) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ProducerCursor.ProtoReflect.Descriptor instead. +func (*ProducerCursor) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDescGZIP(), []int{2} +} + +func (x *ProducerCursor) GetSeq() int64 { + if x != nil { + return x.Seq + } + return 0 +} + +func (x *ProducerCursor) GetFirstOffset() int64 { + if x != nil { + return x.FirstOffset + } + return 0 +} + +func (x *ProducerCursor) GetCount() int64 { + if x != nil { + return x.Count + } + return 0 +} + +func (x *ProducerCursor) GetContentHash() []byte { + if x != nil { + return x.ContentHash + } + return nil +} + +func (x *ProducerCursor) GetFenced() bool { + if x != nil { + return x.Fenced + } + return false +} + +type ConsumerCursor struct { + state protoimpl.MessageState `protogen:"open.v1"` + WorkflowId string `protobuf:"bytes,1,opt,name=workflow_id,json=workflowId,proto3" json:"workflow_id,omitempty"` + RunId string `protobuf:"bytes,2,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"` + // How far this consumer has read. Used to decide whether it needs waking, + // not to decide what the stream may drop. + Offset int64 `protobuf:"varint,3,opt,name=offset,proto3" json:"offset,omitempty"` + Active bool `protobuf:"varint,4,opt,name=active,proto3" json:"active,omitempty"` + // Set when the consumer is a workflow in another execution, which is the + // only case that has to be told the frontier moved. A workflow consuming a + // stream it owns sees that while closing its own transaction. + External bool `protobuf:"varint,5,opt,name=external,proto3" json:"external,omitempty"` + // The oldest offset this consumer's History still depends on. + // + // It is where the subscription started, not where it has read to. A range + // this consumer already consumed is recorded in its History and has to be + // re-readable for the workflow to replay, so bytes below the read position + // are exactly the ones a replay needs most. It stays put while the consumer + // is active and is released when it deregisters. + ReplayFloor int64 `protobuf:"varint,6,opt,name=replay_floor,json=replayFloor,proto3" json:"replay_floor,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ConsumerCursor) Reset() { + *x = ConsumerCursor{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ConsumerCursor) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ConsumerCursor) ProtoMessage() {} + +func (x *ConsumerCursor) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ConsumerCursor.ProtoReflect.Descriptor instead. +func (*ConsumerCursor) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDescGZIP(), []int{3} +} + +func (x *ConsumerCursor) GetWorkflowId() string { + if x != nil { + return x.WorkflowId + } + return "" +} + +func (x *ConsumerCursor) GetRunId() string { + if x != nil { + return x.RunId + } + return "" +} + +func (x *ConsumerCursor) GetOffset() int64 { + if x != nil { + return x.Offset + } + return 0 +} + +func (x *ConsumerCursor) GetActive() bool { + if x != nil { + return x.Active + } + return false +} + +func (x *ConsumerCursor) GetExternal() bool { + if x != nil { + return x.External + } + return false +} + +func (x *ConsumerCursor) GetReplayFloor() int64 { + if x != nil { + return x.ReplayFloor + } + return 0 +} + +// A consuming Workflow's position in a stream. This lives in the consuming +// Workflow's own state rather than on the stream, so advancing it commits in +// the same transaction as the WorkflowTaskCompleted event that records the +// range. Keeping it on the stream would make the advance a cross-execution +// write, and a crash between the two would either redeliver or skip. +type WorkflowStreamCursor struct { + state protoimpl.MessageState `protogen:"open.v1"` + StreamId string `protobuf:"bytes,1,opt,name=stream_id,json=streamId,proto3" json:"stream_id,omitempty"` + // Next offset to deliver. + Offset int64 `protobuf:"varint,4,opt,name=offset,proto3" json:"offset,omitempty"` + // The stream's frontier as of the last delivery. A workflow consuming a + // stream in another execution cannot read the real frontier while closing its + // own transaction, so this is what tells it a capped slice left more behind. + KnownHead int64 `protobuf:"varint,8,opt,name=known_head,json=knownHead,proto3" json:"known_head,omitempty"` + // Set when the stream lives in another execution, which decides whether the + // frontier is read locally or taken from known_head. + External bool `protobuf:"varint,9,opt,name=external,proto3" json:"external,omitempty"` + // The range attached to the Workflow Task currently in flight. Recorded on + // the event that closes that task, then folded into offset. An empty range + // is still recorded: a task where the subscription saw nothing is a fact + // replay has to reproduce. + PendingFrom int64 `protobuf:"varint,5,opt,name=pending_from,json=pendingFrom,proto3" json:"pending_from,omitempty"` + PendingTo int64 `protobuf:"varint,6,opt,name=pending_to,json=pendingTo,proto3" json:"pending_to,omitempty"` + HasPending bool `protobuf:"varint,7,opt,name=has_pending,json=hasPending,proto3" json:"has_pending,omitempty"` + // Where this subscription began reading. Retained separately from offset, + // which advances, because replay has to tell "committed nothing yet" apart + // from "the events recording what was committed are not in this page". + StartOffset int64 `protobuf:"varint,10,opt,name=start_offset,json=startOffset,proto3" json:"start_offset,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *WorkflowStreamCursor) Reset() { + *x = WorkflowStreamCursor{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *WorkflowStreamCursor) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*WorkflowStreamCursor) ProtoMessage() {} + +func (x *WorkflowStreamCursor) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use WorkflowStreamCursor.ProtoReflect.Descriptor instead. +func (*WorkflowStreamCursor) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDescGZIP(), []int{4} +} + +func (x *WorkflowStreamCursor) GetStreamId() string { + if x != nil { + return x.StreamId + } + return "" +} + +func (x *WorkflowStreamCursor) GetOffset() int64 { + if x != nil { + return x.Offset + } + return 0 +} + +func (x *WorkflowStreamCursor) GetKnownHead() int64 { + if x != nil { + return x.KnownHead + } + return 0 +} + +func (x *WorkflowStreamCursor) GetExternal() bool { + if x != nil { + return x.External + } + return false +} + +func (x *WorkflowStreamCursor) GetPendingFrom() int64 { + if x != nil { + return x.PendingFrom + } + return 0 +} + +func (x *WorkflowStreamCursor) GetPendingTo() int64 { + if x != nil { + return x.PendingTo + } + return 0 +} + +func (x *WorkflowStreamCursor) GetHasPending() bool { + if x != nil { + return x.HasPending + } + return false +} + +func (x *WorkflowStreamCursor) GetStartOffset() int64 { + if x != nil { + return x.StartOffset + } + return 0 +} + +type StreamLifecycle struct { + state protoimpl.MessageState `protogen:"open.v1"` + // How long a closed stream stays readable before it is deleted. + Retention *durationpb.Duration `protobuf:"bytes,1,opt,name=retention,proto3" json:"retention,omitempty"` + // Cap on readable messages. Whole batches are reclaimed once the floor + // passes them, so a capped stream has bounded storage. + // + // While a workflow consumer is registered the cap stops being a rolling + // window: the consumer's floor sits at the offset it subscribed from, + // because replay re-reads every range its History recorded, so the cap + // behaves as a lifetime quota measured from that subscription and appends + // past it are refused rather than reclaiming behind the consumer. + MaxItems int64 `protobuf:"varint,2,opt,name=max_items,json=maxItems,proto3" json:"max_items,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamLifecycle) Reset() { + *x = StreamLifecycle{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamLifecycle) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamLifecycle) ProtoMessage() {} + +func (x *StreamLifecycle) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamLifecycle.ProtoReflect.Descriptor instead. +func (*StreamLifecycle) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDescGZIP(), []int{5} +} + +func (x *StreamLifecycle) GetRetention() *durationpb.Duration { + if x != nil { + return x.Retention + } + return nil +} + +func (x *StreamLifecycle) GetMaxItems() int64 { + if x != nil { + return x.MaxItems + } + return 0 +} + +var File_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto protoreflect.FileDescriptor + +const file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDesc = "" + + "\n" + + "Z temporal.api.common.v1.Payload + 6, // 1: temporal.server.chasm.lib.stream.proto.v1.StreamState.producers:type_name -> temporal.server.chasm.lib.stream.proto.v1.StreamState.ProducersEntry + 7, // 2: temporal.server.chasm.lib.stream.proto.v1.StreamState.consumers:type_name -> temporal.server.chasm.lib.stream.proto.v1.StreamState.ConsumersEntry + 5, // 3: temporal.server.chasm.lib.stream.proto.v1.StreamState.lifecycle:type_name -> temporal.server.chasm.lib.stream.proto.v1.StreamLifecycle + 9, // 4: temporal.server.chasm.lib.stream.proto.v1.StreamState.close_time:type_name -> google.protobuf.Timestamp + 1, // 5: temporal.server.chasm.lib.stream.proto.v1.StreamState.budget:type_name -> temporal.server.chasm.lib.stream.proto.v1.StreamBudget + 10, // 6: temporal.server.chasm.lib.stream.proto.v1.StreamLifecycle.retention:type_name -> google.protobuf.Duration + 2, // 7: temporal.server.chasm.lib.stream.proto.v1.StreamState.ProducersEntry.value:type_name -> temporal.server.chasm.lib.stream.proto.v1.ProducerCursor + 3, // 8: temporal.server.chasm.lib.stream.proto.v1.StreamState.ConsumersEntry.value:type_name -> temporal.server.chasm.lib.stream.proto.v1.ConsumerCursor + 9, // [9:9] is the sub-list for method output_type + 9, // [9:9] is the sub-list for method input_type + 9, // [9:9] is the sub-list for extension type_name + 9, // [9:9] is the sub-list for extension extendee + 0, // [0:9] is the sub-list for field type_name +} + +func init() { file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_init() } +func file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_init() { + if File_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDesc), len(file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDesc)), + NumEnums: 0, + NumMessages: 8, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_goTypes, + DependencyIndexes: file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_depIdxs, + MessageInfos: file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_msgTypes, + }.Build() + File_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto = out.File + file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_goTypes = nil + file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_depIdxs = nil +} diff --git a/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go b/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go new file mode 100644 index 0000000000..9644b7bbe0 --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go @@ -0,0 +1,80 @@ +// Code generated by protoc-gen-go-helpers. DO NOT EDIT. +package streampb + +import ( + "google.golang.org/protobuf/proto" +) + +// Marshal an object of type StreamRetentionTask to the protobuf v3 wire format +func (val *StreamRetentionTask) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamRetentionTask from the protobuf v3 wire format +func (val *StreamRetentionTask) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamRetentionTask) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamRetentionTask values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamRetentionTask) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamRetentionTask + switch t := that.(type) { + case *StreamRetentionTask: + that1 = t + case StreamRetentionTask: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +// Marshal an object of type StreamNotifyConsumersTask to the protobuf v3 wire format +func (val *StreamNotifyConsumersTask) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamNotifyConsumersTask from the protobuf v3 wire format +func (val *StreamNotifyConsumersTask) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamNotifyConsumersTask) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamNotifyConsumersTask values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamNotifyConsumersTask) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamNotifyConsumersTask + switch t := that.(type) { + case *StreamNotifyConsumersTask: + that1 = t + case StreamNotifyConsumersTask: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} diff --git a/chasm/lib/stream/gen/streampb/v1/tasks.pb.go b/chasm/lib/stream/gen/streampb/v1/tasks.pb.go new file mode 100644 index 0000000000..2b944c7d24 --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/tasks.pb.go @@ -0,0 +1,160 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// plugins: +// protoc-gen-go +// protoc +// source: temporal/server/chasm/lib/stream/proto/v1/tasks.proto + +package streampb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Fires at close_time plus retention. A closed stream stays readable until +// then, which is what removes the shutdown handshake the signal-based +// implementation forces on producers and consumers. +type StreamRetentionTask struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamRetentionTask) Reset() { + *x = StreamRetentionTask{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamRetentionTask) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamRetentionTask) ProtoMessage() {} + +func (x *StreamRetentionTask) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamRetentionTask.ProtoReflect.Descriptor instead. +func (*StreamRetentionTask) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescGZIP(), []int{0} +} + +// Fires after an append that leaves a registered consumer behind. Appending +// never schedules a workflow task on its own, because a stream item is data an +// execution produced rather than a decision input to it. A workflow that +// subscribed is the exception: it asked to be told, and it cannot find out any +// other way, since nothing else it does would notice the stream moved. +type StreamNotifyConsumersTask struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamNotifyConsumersTask) Reset() { + *x = StreamNotifyConsumersTask{} + mi := &file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamNotifyConsumersTask) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamNotifyConsumersTask) ProtoMessage() {} + +func (x *StreamNotifyConsumersTask) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamNotifyConsumersTask.ProtoReflect.Descriptor instead. +func (*StreamNotifyConsumersTask) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescGZIP(), []int{1} +} + +var File_temporal_server_chasm_lib_stream_proto_v1_tasks_proto protoreflect.FileDescriptor + +const file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDesc = "" + + "\n" + + "5temporal/server/chasm/lib/stream/proto/v1/tasks.proto\x12)temporal.server.chasm.lib.stream.proto.v1\"\x15\n" + + "\x13StreamRetentionTask\"\x1b\n" + + "\x19StreamNotifyConsumersTaskB>Z= head { + continue + } + if len(out) >= maxMessages { + return out, next, nil + } + next = offset + 1 + if len(wanted) > 0 { + if _, ok := wanted[msg.GetTopic()]; !ok { + continue + } + } + // Set here rather than stored: it is decided by where the + // message sits in the log, not by what the producer wrote. + msg.Offset = offset + out = append(out, msg) + } + } + return out, next, nil +} + +// ToAPIRecords converts stored records to the shape carried on a Workflow +// Task. Every stored field crosses over except the offset, which the slice +// carries as a range, so a reader in any language sees the record the producer +// wrote, kind and identity included. +func ToAPIRecords(in []*streamlib.StreamRecord) []*streampb.StreamRecord { + out := make([]*streampb.StreamRecord, 0, len(in)) + for _, m := range in { + out = append(out, &streampb.StreamRecord{ + Body: m.GetBody(), + Metadata: m.GetMetadata(), + Topic: m.GetTopic(), + Kind: m.GetKind(), + ProducerId: m.GetProducerId(), + Attempt: m.GetAttempt(), + Sequence: m.GetSequence(), + }) + } + return out +} + +// CapByBytes trims a contiguous run of records to a byte budget and returns +// the offset just past the last one kept. +// +// It always keeps the first record, however large. Dropping it would leave the +// cursor unable to advance, and since an unconsumed range schedules a workflow +// task, a stream holding one oversized record would wake the workflow forever +// without ever delivering anything. +func CapByBytes( + records []*streamlib.StreamRecord, + from int64, + maxBytes int, +) ([]*streamlib.StreamRecord, int64) { + if len(records) == 0 { + return records, from + } + + total := 0 + for i, m := range records { + total += proto.Size(m) + if total > maxBytes && i > 0 { + return records[:i], from + int64(i) + } + } + return records, from + int64(len(records)) +} + +// checkBatchBytes rejects an append that is too large to store or too large to +// ever hand back. +// +// The per-record bound matters on its own: a record is never split, so one +// that exceeds a consumer's byte budget can never be delivered, and CapByBytes +// would hand it over alone forever rather than reject it. The batch bound is +// the storage side, since a batch is written as a single node. +func checkBatchBytes(records []*streamlib.StreamRecord, limits Limits) error { + total := 0 + for i, m := range records { + size := proto.Size(m) + if size > limits.MaxMessageBytes { + return serviceerror.NewInvalidArgumentf( + "record %d is %d bytes, over the %d byte limit", i, size, limits.MaxMessageBytes) + } + total += size + } + if total > limits.MaxBatchBytes { + return serviceerror.NewInvalidArgumentf( + "batch is %d bytes, over the %d byte limit", total, limits.MaxBatchBytes) + } + return nil +} diff --git a/chasm/lib/stream/messages_test.go b/chasm/lib/stream/messages_test.go new file mode 100644 index 0000000000..bd18bb05b3 --- /dev/null +++ b/chasm/lib/stream/messages_test.go @@ -0,0 +1,91 @@ +package stream + +import ( + "testing" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + streampb "go.temporal.io/api/stream/v1" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +func sized(n int, bytes int) []*streamlib.StreamRecord { + out := make([]*streamlib.StreamRecord, n) + for i := range out { + out[i] = &streamlib.StreamRecord{ + Body: &commonpb.Payload{Data: make([]byte, bytes)}, + Kind: streampb.STREAM_RECORD_KIND_DATA, + } + } + return out +} + +func TestCapByBytesTrimsToAPrefix(t *testing.T) { + messages, next := CapByBytes(sized(10, 100), 4, 250) + require.Len(t, messages, 2) + require.Equal(t, int64(6), next, "the recorded range must cover exactly what was kept") +} + +func TestCapByBytesKeepsEverythingUnderBudget(t *testing.T) { + messages, next := CapByBytes(sized(3, 10), 0, 1<<20) + require.Len(t, messages, 3) + require.Equal(t, int64(3), next) +} + +// A single oversized message must still go out. Held back it would stall the +// cursor, and an unconsumed range schedules a workflow task, so the workflow +// would wake forever and never receive anything. +func TestCapByBytesAlwaysDeliversTheFirstMessage(t *testing.T) { + messages, next := CapByBytes(sized(3, 5000), 7, 10) + require.Len(t, messages, 1) + require.Equal(t, int64(8), next) +} + +func TestCapByBytesOnAnEmptyRun(t *testing.T) { + messages, next := CapByBytes(nil, 9, 100) + require.Empty(t, messages) + require.Equal(t, int64(9), next) +} + +// A reader in any language decodes the record the producer wrote, so every +// field the store holds has to come back on the Workflow Task, the kind and the +// producer identity included. A FINISH record is a record like any other to the +// consumer that reads it. +func TestToAPIRecordsCarriesTheRecordAsWritten(t *testing.T) { + stored := []*streamlib.StreamRecord{ + { + Body: &commonpb.Payload{Data: []byte("token")}, + Metadata: map[string]*commonpb.Payload{"model": {Data: []byte("m1")}}, + Topic: "tokens", + Kind: streampb.STREAM_RECORD_KIND_DATA, + ProducerId: "model-call", + Attempt: 2, + Sequence: 7, + Offset: 41, + }, + { + Topic: "tokens", + Kind: streampb.STREAM_RECORD_KIND_FINISH, + ProducerId: "model-call", + Attempt: 2, + Sequence: 8, + Offset: 42, + }, + } + + got := ToAPIRecords(stored) + require.Len(t, got, 2) + + require.Equal(t, "token", string(got[0].GetBody().GetData())) + require.Equal(t, "m1", string(got[0].GetMetadata()["model"].GetData())) + require.Equal(t, "tokens", got[0].GetTopic()) + require.Equal(t, streampb.STREAM_RECORD_KIND_DATA, got[0].GetKind()) + require.Equal(t, "model-call", got[0].GetProducerId()) + require.Equal(t, int64(2), got[0].GetAttempt()) + require.Equal(t, int64(7), got[0].GetSequence()) + + require.Equal(t, streampb.STREAM_RECORD_KIND_FINISH, got[1].GetKind()) + require.Nil(t, got[1].GetBody()) + require.Equal(t, "model-call", got[1].GetProducerId()) + require.Equal(t, int64(8), got[1].GetSequence()) +} diff --git a/chasm/lib/stream/proto/v1/message.proto b/chasm/lib/stream/proto/v1/message.proto new file mode 100644 index 0000000000..fc7cdd00f3 --- /dev/null +++ b/chasm/lib/stream/proto/v1/message.proto @@ -0,0 +1,42 @@ +syntax = "proto3"; + +package temporal.server.chasm.lib.stream.proto.v1; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/stream/v1/message.proto"; + +option go_package = "go.temporal.io/server/chasm/lib/stream/gen/streampb;streampb"; + +// The stored shape of temporal.api.stream.v1.StreamRecord plus the offset a +// read assigns. Field for field the public record, so one crosses the frontend +// in either direction without translation. +message StreamRecord { + temporal.api.common.v1.Payload body = 1; + // Producer-supplied provenance, stored as sent. + map metadata = 2; + string topic = 3; + // The producer's position within its attempt, zero when it does not number + // its records. Stored as sent; the global offset is what orders the stream. + int64 sequence = 4; + // Settled to DATA on append when left unspecified, so a retry hashes the + // same bytes and a reader never sees the zero value. + temporal.api.stream.v1.StreamRecordKind kind = 5; + + // Position in the whole stream, set on read and never stored. A consumer + // that resumes at record granularity needs it, and a topic-filtered read + // leaves gaps that make it underivable from the response alone. + int64 offset = 6; + // Who wrote the record. Empty when the owning Workflow did: its publish + // command clears the field, whatever the worker sent. + string producer_id = 7; + // The producer's attempt, stored as sent. Readers treat a later attempt by + // the same producer as superseding what the earlier one wrote. + int64 attempt = 8; +} + +// One append is one batch, and one batch is one data node. The server stores +// this serialized and opaque; it decodes only to trim a partial first page or +// to apply a topic filter. +message StreamRecordBatch { + repeated StreamRecord records = 1; +} diff --git a/chasm/lib/stream/proto/v1/stream_state.proto b/chasm/lib/stream/proto/v1/stream_state.proto new file mode 100644 index 0000000000..70cb9c9f67 --- /dev/null +++ b/chasm/lib/stream/proto/v1/stream_state.proto @@ -0,0 +1,145 @@ +syntax = "proto3"; + +package temporal.server.chasm.lib.stream.proto.v1; + +import "google/protobuf/duration.proto"; +import "google/protobuf/timestamp.proto"; +import "temporal/api/common/v1/message.proto"; + +option go_package = "go.temporal.io/server/chasm/lib/stream/gen/streampb;streampb"; + +// Size is O(producers + consumers), never O(messages). Payload bytes live in +// the component's data nodes, not here, which keeps this off the CHASM +// partial-read path. +message StreamState { + // Field numbers of removed fields, kept out of reuse so an old blob never + // decodes into a field with a different meaning. + reserved 3, 6, 7, 8; + + // Visibility frontier. Readers never observe an offset at or past this. + int64 head_offset = 1; + // Truncation floor. Offsets below this are gone. + int64 base_offset = 2; + + bool closed = 4; + temporal.api.common.v1.Payload close_reason = 5; + + map producers = 9; + map consumers = 10; + + StreamLifecycle lifecycle = 11; + + // Set when a successor run takes ownership, so an in-flight poll can follow + // the chain instead of stalling on a superseded run. + string redirect_run_id = 12; + + // Wall-clock close time, used to schedule retention deletion. + google.protobuf.Timestamp close_time = 13; + + // Set on a stream a workflow owns. Its batches are the owning execution's + // mutable state, so appends past the budget are refused rather than left to + // the execution size limit, which terminates the workflow. + StreamBudget budget = 14; + + // Bytes appended over the stream's life, kept for the budget check. A stream + // with a budget carries no lifecycle cap and is never truncated, so nothing + // reclaims behind it and this is also what it holds. The item side of the + // budget measures head minus base for the same reason: offsets are global + // and a stream can begin above zero. + int64 appended_bytes = 15; + + // A notify task is scheduled and has not run yet. Appends while it is set + // schedule none of their own; the task reads the head when it runs, so it + // carries every append that landed before it. + bool notify_pending = 16; +} + +// Hard bounds on what a stream may hold. Distinct from StreamLifecycle.max_items, +// which reclaims the oldest messages: a budget refuses the newest. +message StreamBudget { + int64 max_items = 1; + int64 max_bytes = 2; +} + +message ProducerCursor { + int64 seq = 1; + int64 first_offset = 2; + int64 count = 3; + // Distinguishes a genuine retry from a client reusing a sequence with + // different content, which must be rejected rather than deduplicated. + bytes content_hash = 4; + // Set by FinishWriting. Ends this producer's writes without closing the + // stream for anyone else. + bool fenced = 5; +} + +message ConsumerCursor { + string workflow_id = 1; + string run_id = 2; + // How far this consumer has read. Used to decide whether it needs waking, + // not to decide what the stream may drop. + int64 offset = 3; + bool active = 4; + // Set when the consumer is a workflow in another execution, which is the + // only case that has to be told the frontier moved. A workflow consuming a + // stream it owns sees that while closing its own transaction. + bool external = 5; + // The oldest offset this consumer's History still depends on. + // + // It is where the subscription started, not where it has read to. A range + // this consumer already consumed is recorded in its History and has to be + // re-readable for the workflow to replay, so bytes below the read position + // are exactly the ones a replay needs most. It stays put while the consumer + // is active and is released when it deregisters. + int64 replay_floor = 6; +} + +// A consuming Workflow's position in a stream. This lives in the consuming +// Workflow's own state rather than on the stream, so advancing it commits in +// the same transaction as the WorkflowTaskCompleted event that records the +// range. Keeping it on the stream would make the advance a cross-execution +// write, and a crash between the two would either redeliver or skip. +message WorkflowStreamCursor { + reserved 2, 3; + + string stream_id = 1; + + // Next offset to deliver. + int64 offset = 4; + + // The stream's frontier as of the last delivery. A workflow consuming a + // stream in another execution cannot read the real frontier while closing its + // own transaction, so this is what tells it a capped slice left more behind. + int64 known_head = 8; + + // Set when the stream lives in another execution, which decides whether the + // frontier is read locally or taken from known_head. + bool external = 9; + + // The range attached to the Workflow Task currently in flight. Recorded on + // the event that closes that task, then folded into offset. An empty range + // is still recorded: a task where the subscription saw nothing is a fact + // replay has to reproduce. + int64 pending_from = 5; + int64 pending_to = 6; + bool has_pending = 7; + + // Where this subscription began reading. Retained separately from offset, + // which advances, because replay has to tell "committed nothing yet" apart + // from "the events recording what was committed are not in this page". + int64 start_offset = 10; +} + +message StreamLifecycle { + // How long a closed stream stays readable before it is deleted. + google.protobuf.Duration retention = 1; + // Cap on readable messages. Whole batches are reclaimed once the floor + // passes them, so a capped stream has bounded storage. + // + // While a workflow consumer is registered the cap stops being a rolling + // window: the consumer's floor sits at the offset it subscribed from, + // because replay re-reads every range its History recorded, so the cap + // behaves as a lifetime quota measured from that subscription and appends + // past it are refused rather than reclaiming behind the consumer. + int64 max_items = 2; +} diff --git a/chasm/lib/stream/proto/v1/tasks.proto b/chasm/lib/stream/proto/v1/tasks.proto new file mode 100644 index 0000000000..c4b8e5e6c6 --- /dev/null +++ b/chasm/lib/stream/proto/v1/tasks.proto @@ -0,0 +1,17 @@ +syntax = "proto3"; + +package temporal.server.chasm.lib.stream.proto.v1; + +option go_package = "go.temporal.io/server/chasm/lib/stream/gen/streampb;streampb"; + +// Fires at close_time plus retention. A closed stream stays readable until +// then, which is what removes the shutdown handshake the signal-based +// implementation forces on producers and consumers. +message StreamRetentionTask {} + +// Fires after an append that leaves a registered consumer behind. Appending +// never schedules a workflow task on its own, because a stream item is data an +// execution produced rather than a decision input to it. A workflow that +// subscribed is the exception: it asked to be told, and it cannot find out any +// other way, since nothing else it does would notice the stream moved. +message StreamNotifyConsumersTask {} diff --git a/chasm/lib/stream/stream.go b/chasm/lib/stream/stream.go new file mode 100644 index 0000000000..dd3bd6b190 --- /dev/null +++ b/chasm/lib/stream/stream.go @@ -0,0 +1,811 @@ +package stream + +import ( + "bytes" + "crypto/sha256" + "maps" + "slices" + "time" + + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/serviceerror" + streampb "go.temporal.io/api/stream/v1" + "go.temporal.io/server/chasm" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + "go.temporal.io/server/common" + "go.temporal.io/server/common/payload" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// Stream is a durable, offset-addressed append-only sequence. State holds the +// frontier and the producer and consumer tables, so it is O(producers + +// consumers) no matter how long the stream gets; the payload lives in Batches. +// +// Appending never schedules a workflow task. A stream item is data produced by +// an execution, not a decision input to it, so nothing in a workflow's state +// machine advances because one arrived. +type Stream struct { + chasm.UnimplementedComponent + + State *streamlib.StreamState + + // Batches holds the payload, each keyed by the offset it starts at. They are + // data nodes, so they replicate with the component and are reclaimed with + // it, and a retry addresses the same key rather than racing it. + // + // They also live in mutable state, so the payload counts against the + // execution size limit of whatever execution holds the stream. + Batches chasm.Map[int64, *commonpb.DataBlob] + + // Present so streams are listable. Operators need to find them the same way + // they find workflows, and without this the only way to reach a stream is + // to already know its ID. + Visibility chasm.Field[*chasm.Visibility] +} + +type NewStreamRequest struct { + Lifecycle *streamlib.StreamLifecycle + + // Budget bounds what the stream may hold. Set for a stream a workflow + // owns, whose batches are the owning execution's mutable state. + Budget *streamlib.StreamBudget + + // Attached means the stream is a subcomponent of another execution rather + // than a root. CHASM requires a visibility component to be an immediate + // child of the root, so an attached stream carries none and is found + // through its owner instead of through ListStreams. + Attached bool +} + +type AddMessagesRequest struct { + Records []*streamlib.StreamRecord + + // Optional idempotency. A producer supplies either an identity and + // sequence, or an expected offset, or neither and accepts at-least-once. + // + // The dedup table keeps one entry per producer id, so a producer may have + // one append in flight at a time: sequences have to arrive in order, and a + // retry of an earlier sequence after a later one committed is refused. + // Pipelining needs a distinct producer id per lane. + ProducerID string + Sequence int64 + ExpectedOffset *int64 + + // The namespace's limits, resolved by the caller. A zero value means the + // defaults. + Limits Limits +} + +type AddMessagesResult struct { + FirstOffset int64 + NextOffset int64 + Count int64 + + // True when a retry matched a recorded producer sequence, so nothing was + // appended and the original offsets are returned. + Deduplicated bool + + // The bytes this append wrote, so a caller can prime a cache without + // reading them back. Nil when deduplicated. + Blob *commonpb.DataBlob +} + +func NewStream(ctx chasm.MutableContext, req NewStreamRequest) (*Stream, error) { + visibility := chasm.NewEmptyField[*chasm.Visibility]() + if !req.Attached { + visibility = chasm.NewComponentField(ctx, chasm.NewVisibility(ctx)) + } + return &Stream{ + Visibility: visibility, + Batches: make(chasm.Map[int64, *commonpb.DataBlob]), + State: &streamlib.StreamState{ + Lifecycle: req.Lifecycle, + Budget: req.Budget, + Producers: make(map[string]*streamlib.ProducerCursor), + Consumers: make(map[string]*streamlib.ConsumerCursor), + }, + }, nil +} + +// ContextMetadata satisfies chasm.RootComponent. A stream carries no metadata +// worth propagating to the request context. +func (s *Stream) ContextMetadata(_ chasm.Context) map[string]string { + return nil +} + +// Terminate seals the stream so a forced shutdown does not leave it accepting +// writes. Data already appended stays readable, because consumers may have read +// it and the stream is append-only. +func (s *Stream) Terminate( + mctx chasm.MutableContext, + req chasm.TerminateComponentRequest, +) (chasm.TerminateComponentResponse, error) { + // Encoded rather than wrapped raw: a payload without encoding metadata is + // not decodable by any SDK data converter, so the reason would reach a + // reader as opaque bytes. + return chasm.TerminateComponentResponse{}, + s.CloseAndSchedule(mctx, payload.EncodeString(req.Reason)) +} + +// Snapshot returns a copy of the frontier for read paths. It is a copy because +// the caller reads it outside the transition that produced it. +func (s *Stream) Snapshot(_ chasm.Context, _ struct{}) (*streamlib.StreamState, error) { + return common.CloneProto(s.State), nil +} + +func (s *Stream) LifecycleState(_ chasm.Context) chasm.LifecycleState { + if s.State.Closed { + return chasm.LifecycleStateCompleted + } + return chasm.LifecycleStateRunning +} + +// AddMessages assigns a contiguous offset range and writes the bytes into the +// component, so the payload and the frontier commit in one transaction. A torn +// append is therefore not a state anyone can observe. +func (s *Stream) AddMessages( + mctx chasm.MutableContext, + req AddMessagesRequest, +) (AddMessagesResult, error) { + if s.State.Closed { + return AddMessagesResult{}, serviceerror.NewFailedPrecondition("stream is closed") + } + if len(req.Records) == 0 { + return AddMessagesResult{}, serviceerror.NewInvalidArgument("no records to append") + } + if len(req.Records) > MaxRecordsPerBatch { + return AddMessagesResult{}, serviceerror.NewInvalidArgumentf( + "batch of %d exceeds the limit of %d records", len(req.Records), MaxRecordsPerBatch) + } + limits := req.Limits.withDefaults() + if err := checkBatchBytes(req.Records, limits); err != nil { + return AddMessagesResult{}, err + } + blob, err := marshalBatch(settleKinds(req.Records)) + if err != nil { + return AddMessagesResult{}, err + } + hash := contentHash(blob.Data) + + if replay, err := s.checkProducer(req, hash); err != nil || replay != nil { + if err != nil { + return AddMessagesResult{}, err + } + return *replay, nil + } + + // After the retry check, so a known producer is never rejected for room. + if err := s.checkProducerRoom(req.ProducerID, limits.MaxProducersPerStream); err != nil { + return AddMessagesResult{}, err + } + if err := s.checkBudget(int64(len(req.Records)), int64(len(blob.Data))); err != nil { + return AddMessagesResult{}, err + } + + // Before anything is written, because the alternative is to write and then + // discover the cap can only be met by deleting bytes a consumer's committed + // History still refers to. Refusing the write is the honest half of that + // choice: capacity may constrain what is admitted, and may not quietly take + // back a workflow's ability to replay a decision it already made. + if err := s.checkCapRoom(int64(len(req.Records))); err != nil { + return AddMessagesResult{}, err + } + + if req.ExpectedOffset != nil && *req.ExpectedOffset != s.State.HeadOffset { + return AddMessagesResult{}, serviceerror.NewAlreadyExistsf( + "expected offset %d but stream head is %d", *req.ExpectedOffset, s.State.HeadOffset) + } + + first := s.State.HeadOffset + count := int64(len(req.Records)) + + if s.Batches == nil { + s.Batches = make(chasm.Map[int64, *commonpb.DataBlob]) + } + s.Batches[first] = chasm.NewDataField(mctx, blob) + + s.State.HeadOffset = first + count + s.State.AppendedBytes += int64(len(blob.Data)) + if req.ProducerID != "" { + if s.State.Producers == nil { + s.State.Producers = make(map[string]*streamlib.ProducerCursor) + } + s.State.Producers[req.ProducerID] = &streamlib.ProducerCursor{ + Seq: req.Sequence, + FirstOffset: first, + Count: count, + ContentHash: hash, + } + } + + s.applyCap() + result := AddMessagesResult{ + FirstOffset: first, + NextOffset: s.State.HeadOffset, + Count: count, + Blob: blob, + } + s.notifyConsumers(mctx) + return result, nil +} + +// notifyConsumers schedules the wake for consumers this append left behind. +// +// Only a workflow in another execution needs it. One consuming a stream it owns +// sees the new frontier while closing its own transaction, so waking it through +// a task would only duplicate a decision already made locally. +func (s *Stream) notifyConsumers(mctx chasm.MutableContext) { + // Without a context there is no transition to attach the task to. The + // component's own unit tests drive the transitions that way. + if mctx == nil { + return + } + // One task outstanding at a time. The task reads the head when it runs and + // clears the flag before it reads, so an append that lands after the clear + // schedules the next one and nothing is missed. + if s.State.NotifyPending { + return + } + for _, consumer := range s.State.Consumers { + if consumer.GetExternal() && consumer.GetActive() && consumer.GetOffset() < s.State.HeadOffset { + s.State.NotifyPending = true + mctx.AddTask(s, chasm.TaskAttributes{ScheduledTime: mctx.Now(s)}, + &streamlib.StreamNotifyConsumersTask{}) + return + } + } +} + +// TakeNotifySnapshot is the notify task's read of the stream. Clearing the +// pending flag in the same transition that reads the head is what makes the +// coalescing safe: any append that commits after this one sees the flag down +// and schedules its own task. +func (s *Stream) TakeNotifySnapshot( + _ chasm.MutableContext, _ struct{}, +) (*streamlib.StreamState, error) { + s.State.NotifyPending = false + return common.CloneProto(s.State), nil +} + +// checkProducerRoom keeps the dedup table bounded. +// +// Entries whose whole batch sits below the floor go first: a retry of a batch +// that truncation already removed cannot be served its recorded offsets +// anyway, so the entry has no use left. +func (s *Stream) checkProducerRoom(producerID string, maxProducers int) error { + if producerID == "" { + return nil + } + if _, known := s.State.Producers[producerID]; known { + return nil + } + if len(s.State.Producers) < maxProducers { + return nil + } + for id, cursor := range s.State.Producers { + if cursor.GetFirstOffset()+cursor.GetCount() <= s.State.GetBaseOffset() { + delete(s.State.Producers, id) + } + } + if len(s.State.Producers) >= maxProducers { + return serviceerror.NewInvalidArgumentf( + "stream already tracks %d producers, which is the limit", maxProducers) + } + return nil +} + +// held is how many records the stream still has. Offsets are global and a +// stream can begin above zero, so the head alone is a position rather than an +// amount. +func (s *Stream) held() int64 { + return s.State.HeadOffset - s.State.BaseOffset +} + +// checkBudget refuses an append a budgeted stream cannot hold. +// +// Refused rather than reclaimed: the budget exists because the batches are +// mutable state of the execution that owns the stream, and the alternative to +// refusing here is the execution size limit terminating that workflow later, +// with nothing naming the stream as the cause. +func (s *Stream) checkBudget(count int64, size int64) error { + budget := s.State.GetBudget() + if budget == nil { + return nil + } + if limit := budget.GetMaxItems(); limit > 0 && s.held()+count > limit { + return serviceerror.NewResourceExhaustedf( + enumspb.RESOURCE_EXHAUSTED_CAUSE_PERSISTENCE_STORAGE_LIMIT, + "stream holds %d of its budget of %d records; the append of %d does not fit", + s.held(), limit, count) + } + if limit := budget.GetMaxBytes(); limit > 0 && s.State.AppendedBytes+size > limit { + return serviceerror.NewResourceExhaustedf( + enumspb.RESOURCE_EXHAUSTED_CAUSE_PERSISTENCE_STORAGE_LIMIT, + "stream holds %d of its budget of %d bytes; the append of %d does not fit", + s.State.AppendedBytes, limit, size) + } + return nil +} + +// checkProducer applies per-producer idempotency. It returns a replay result +// when the request is a genuine retry, and an error when it is not a retry but +// cannot be accepted either. +func (s *Stream) checkProducer(req AddMessagesRequest, hash []byte) (*AddMessagesResult, error) { + if req.ProducerID == "" { + return nil, nil + } + cursor := s.State.Producers[req.ProducerID] + if cursor == nil { + return nil, nil + } + if cursor.Fenced { + return nil, serviceerror.NewFailedPrecondition("producer has finished writing to this stream") + } + if req.Sequence > cursor.Seq { + return nil, nil + } + if req.Sequence < cursor.Seq { + return nil, serviceerror.NewInvalidArgumentf( + "stale producer sequence %d, last accepted for producer %q is %d; a producer id "+ + "carries one append at a time, so use a separate id per concurrent lane", + req.Sequence, req.ProducerID, cursor.Seq) + } + // Same sequence. Identical content is a retry; different content is a + // client bug, and returning the recorded offsets would report success while + // silently dropping the caller's data. + if !bytes.Equal(cursor.ContentHash, hash) { + return nil, serviceerror.NewInvalidArgumentf( + "producer sequence %d already used with different content", req.Sequence) + } + return &AddMessagesResult{ + FirstOffset: cursor.FirstOffset, + NextOffset: cursor.FirstOffset + cursor.Count, + Count: cursor.Count, + Deduplicated: true, + }, nil +} + +// FinishWriting ends one producer's writes without closing the stream, so other +// producers carry on. Weaker than Close on purpose. +func (s *Stream) FinishWriting(_ chasm.MutableContext, producerID string) error { + if producerID == "" { + return serviceerror.NewInvalidArgument("producer id is required") + } + if s.State.Producers == nil { + s.State.Producers = make(map[string]*streamlib.ProducerCursor) + } + cursor := s.State.Producers[producerID] + if cursor == nil { + cursor = &streamlib.ProducerCursor{Seq: -1} + s.State.Producers[producerID] = cursor + } + cursor.Fenced = true + return nil +} + +// Close seals the stream. It does not delete it: a closed stream stays readable +// through retention, which is what removes the shutdown handshake the current +// signal-based implementation forces on users. +// Close returns when retention deletion should be scheduled, or the zero time +// if the stream was already closed or has no retention configured. Scheduling +// is the caller's job, which keeps the component a pure state transition and +// testable without a live context. +func (s *Stream) Close(now time.Time, reason *commonpb.Payload) time.Time { + if s.State.Closed { + return time.Time{} + } + s.State.Closed = true + s.State.CloseReason = reason + s.State.CloseTime = timestamppb.New(now) + + retention := s.State.GetLifecycle().GetRetention().AsDuration() + if retention <= 0 { + return time.Time{} + } + return now.Add(retention) +} + +// CloseAndSchedule closes the stream and arms retention if it asked for it. +func (s *Stream) CloseAndSchedule(mctx chasm.MutableContext, reason *commonpb.Payload) error { + if at := s.Close(mctx.Now(s), reason); !at.IsZero() { + mctx.AddTask(s, chasm.TaskAttributes{ScheduledTime: at}, &streamlib.StreamRetentionTask{}) + } + return nil +} + +// Truncate advances the readable floor. +// +// It stops at an active consumer's replay floor. A workflow that consumed a +// range recorded that range in its History and can be asked to replay from it, +// so those bytes are part of its recovery rather than spare capacity. Dropping +// them succeeds here and fails much later, during a replay nobody is watching, +// which is the worst place to find out. +// +// The floor is released by deregistering the consumer, which is an act someone +// takes deliberately. An operator who means to drop the bytes anyway does that +// first, and then this call goes through. +// +// A consumer that is behind but not active is not protected. Reading from +// below the base is an error naming where the stream now starts, the same +// answer a log with a retention window gives anywhere else. +func (s *Stream) Truncate(_ chasm.MutableContext, newBase int64) error { + if newBase < s.State.BaseOffset { + return serviceerror.NewInvalidArgumentf( + "cannot truncate backwards from %d to %d", s.State.BaseOffset, newBase) + } + if newBase > s.State.HeadOffset { + return serviceerror.NewInvalidArgumentf( + "cannot truncate past head offset %d", s.State.HeadOffset) + } + if floor, holder, pinned := s.replayFloor(); pinned && newBase > floor { + return serviceerror.NewFailedPreconditionf( + "cannot truncate to %d: consumer %q still depends on offset %d and above "+ + "to replay; deregister it first if those records are no longer needed", + newBase, holder, floor) + } + s.State.BaseOffset = newBase + s.reclaim(newBase) + return nil +} + +// replayFloor is the oldest offset any active consumer's History still depends +// on, and who is holding it. Named, because a refusal that does not say which +// consumer to look at leaves the operator with nothing to act on. +func (s *Stream) replayFloor() (int64, string, bool) { + var floor int64 + var holder string + found := false + for id, c := range s.State.Consumers { + if !c.GetActive() { + continue + } + if !found || c.GetReplayFloor() < floor { + floor = c.GetReplayFloor() + holder = id + found = true + } + } + return floor, holder, found +} + +// reclaim drops batches lying entirely below the readable floor. A batch +// straddling the floor stays, because the offsets above it are still readable. +func (s *Stream) reclaim(newBase int64) { + starts := s.batchStarts() + for i, start := range starts { + end := s.State.HeadOffset + if i+1 < len(starts) { + end = starts[i+1] + } + if end > newBase { + return + } + delete(s.Batches, start) + } +} + +// batchStarts returns the batch keys in offset order. Reading and reclaiming +// both need where a batch ends, which is where the next one begins. +func (s *Stream) batchStarts() []int64 { + return slices.Sorted(maps.Keys(s.Batches)) +} + +// WindowRequest asks for whatever a reader can be given from an offset. +type WindowRequest struct { + From int64 + MaxMessages int32 + Topics []string +} + +// Window is one read's worth: the frontier it was served against, the batches +// covering the range, and the range itself. +type Window struct { + State *streamlib.StreamState + Blobs []*commonpb.DataBlob + Starts []int64 + To int64 + Limit int + // The execution holding the stream, so a slice built from this window can + // say which run it came from. + RunID string +} + +// ReadWindow serves a read from the component, so the frontier and the bytes +// come from one view. Read separately they can disagree, because the frontier +// moves while the bytes are being fetched. +func (s *Stream) ReadWindow(ctx chasm.Context, req WindowRequest) (Window, error) { + if req.From < s.State.BaseOffset { + return Window{}, serviceerror.NewFailedPreconditionf( + "offset %d has been truncated, the stream starts at %d", req.From, s.State.BaseOffset) + } + if req.From > s.State.HeadOffset { + return Window{}, serviceerror.NewInvalidArgumentf( + "offset %d is past the stream head %d", req.From, s.State.HeadOffset) + } + + // Clamped at both ends. The caller picks the page size, and an unclamped + // one lets a single poll ask the store to materialise the whole stream. + limit := int(req.MaxMessages) + if limit <= 0 || limit > DefaultMaxMessagesPerPoll { + limit = DefaultMaxMessagesPerPoll + } + w := Window{ + State: common.CloneProto(s.State), + To: req.From, + Limit: limit, + RunID: ctx.ExecutionKey().RunID, + } + if req.From == s.State.HeadOffset { + return w, nil + } + + // Clip to what the caller can actually be given. One offset is one record, + // so the bound is exact. Without it a poll for a single record off a long + // stream materialises every batch to the head before trimming. + w.To = min(s.State.HeadOffset, req.From+int64(limit)) + blobs, starts, err := s.ReadBatches(ctx, req.From, w.To, 0) + if err != nil { + return Window{}, err + } + w.Blobs, w.Starts = blobs, starts + return w, nil +} + +// ReadBatches returns the batches covering [from, to), oldest first, alongside +// the offset each one starts at. A read landing mid-batch gets the batch +// holding it, because a consumer asks for an offset rather than for a batch. +// Blobs come back unparsed: decoding user payloads is the SDK's job. +func (s *Stream) ReadBatches( + ctx chasm.Context, + from int64, + to int64, + maxBatches int, +) ([]*commonpb.DataBlob, []int64, error) { + if from >= to { + return nil, nil, nil + } + var blobs []*commonpb.DataBlob + var starts []int64 + all := s.batchStarts() + for i, start := range all { + end := s.State.HeadOffset + if i+1 < len(all) { + end = all[i+1] + } + if end <= from { + continue + } + if start >= to { + break + } + blobs = append(blobs, s.Batches[start].Get(ctx)) + starts = append(starts, start) + if maxBatches > 0 && len(blobs) >= maxBatches { + break + } + } + return blobs, starts, nil +} + +// applyCap advances the readable floor when the stream is over its record cap. +// Evaluated at the end of a successful append rather than by a sweeper: the +// append transition is already writing, so folding the check into it costs +// nothing and keeps the cap tight instead of eventually true. +func (s *Stream) applyCap() { + maxItems := s.State.GetLifecycle().GetMaxItems() + if maxItems <= 0 { + return + } + readable := s.State.HeadOffset - s.State.BaseOffset + if readable <= maxItems { + return + } + newBase := s.State.HeadOffset - maxItems + // Clamped rather than refused, because refusing belongs to admission and + // has already happened: checkCapRoom turned away the append that would have + // needed this. Reaching the clamp means a consumer registered after the + // records were written, and keeping its bytes is still the right answer. + if floor, _, pinned := s.replayFloor(); pinned && newBase > floor { + newBase = floor + } + if newBase <= s.State.BaseOffset { + return + } + s.State.BaseOffset = newBase + s.reclaim(newBase) +} + +// checkCapRoom refuses an append the cap could only absorb by dropping bytes an +// active consumer still needs. +// +// A capped stream with no consumer behaves as before: the oldest records go. +// The refusal only arrives when honouring the cap and honouring a recorded +// consumption are the same records, and it names the consumer so the operator +// knows what to do about it. +// +// The floor is the subscription's start and does not move while the consumer +// is active, because replay re-reads every range the consumer's History +// recorded, back to that start. So on a stream with a subscriber the cap is a +// lifetime quota rather than a rolling window, and a consumer that has read +// everything still holds the writers. Deregistering the consumer releases it. +// The two promises cannot both hold, and the one kept is that a workflow can +// always replay a decision it already made. +func (s *Stream) checkCapRoom(count int64) error { + maxItems := s.State.GetLifecycle().GetMaxItems() + if maxItems <= 0 { + return nil + } + wantBase := s.State.HeadOffset + count - maxItems + if wantBase <= s.State.BaseOffset { + return nil + } + floor, holder, pinned := s.replayFloor() + if !pinned || wantBase <= floor { + return nil + } + return serviceerror.NewResourceExhaustedf( + enumspb.RESOURCE_EXHAUSTED_CAUSE_UNSPECIFIED, + "stream is at its cap of %d records and consumer %q still depends on "+ + "offset %d and above to replay; the append would have to delete those "+ + "records to make room", + maxItems, holder, floor) +} + +// ConsumerRegistration describes a consumer being registered on a stream. +type ConsumerRegistration struct { + ConsumerID string + WorkflowID string + RunID string + // Negative means the head as of the registering transition. + Offset int64 + External bool + // Zero means the default. + MaxConsumers int +} + +// RegisterConsumer records an in-workflow consumer, so appends know who to wake +// and retention knows what it may not delete. It returns the offset the +// consumer reads from: the resolved start for a new consumer, and the current +// read position for one that is already registered. +// +// A negative offset means the head as of this transition. Resolving it here, +// against the frontier the same transaction sees, is what makes the recorded +// start a fact rather than a reading taken a moment earlier. +// +// The floor it records is where the subscription started, not where it has read +// to. The ranges this consumer already took are written into its History, and a +// replay is asked to reproduce them, so the bytes behind the read position are +// the ones a recovery needs. Deregistering releases the floor. +func (s *Stream) RegisterConsumer(_ chasm.MutableContext, reg ConsumerRegistration) (int64, error) { + if reg.ConsumerID == "" { + return 0, serviceerror.NewInvalidArgument("consumer id is required") + } + maxConsumers := reg.MaxConsumers + if maxConsumers <= 0 { + maxConsumers = MaxConsumersPerStream + } + offset := reg.Offset + if offset < 0 { + offset = s.State.HeadOffset + } + if offset < s.State.BaseOffset { + return 0, serviceerror.NewFailedPreconditionf( + "offset %d is below the stream's floor of %d", offset, s.State.BaseOffset) + } + if _, known := s.State.Consumers[reg.ConsumerID]; !known && + len(s.State.Consumers) >= maxConsumers { + return 0, serviceerror.NewInvalidArgumentf( + "stream already has %d consumers, which is the limit", maxConsumers) + } + if s.State.Consumers == nil { + s.State.Consumers = make(map[string]*streamlib.ConsumerCursor) + } + // One workflow id has one open run, so an entry for another run of this + // workflow belongs to a closed run. Its floor would otherwise hold storage + // for a replay nobody can ask for, and the new run would inherit it. + for id, c := range s.State.Consumers { + if id != reg.ConsumerID && c.GetExternal() && + c.GetWorkflowId() == reg.WorkflowID && c.GetRunId() != reg.RunID { + delete(s.State.Consumers, id) + } + } + if existing, ok := s.State.Consumers[reg.ConsumerID]; ok { + // A consumer coming back after its floor was released can find the + // stream has moved past what its History refers to. Saying so here is + // the only chance to say it before the workflow depends on it again. + if existing.GetReplayFloor() < s.State.BaseOffset { + return 0, serviceerror.NewFailedPreconditionf( + "consumer %q recorded offset %d, and the stream now starts at %d", + reg.ConsumerID, existing.GetReplayFloor(), s.State.BaseOffset) + } + existing.Active = true + return existing.GetOffset(), nil + } + s.State.Consumers[reg.ConsumerID] = &streamlib.ConsumerCursor{ + WorkflowId: reg.WorkflowID, + RunId: reg.RunID, + Offset: offset, + Active: true, + External: reg.External, + ReplayFloor: offset, + } + return offset, nil +} + +// AdvanceConsumer moves a consumer's pin forward as it reads. It never moves +// backwards: the floor is what lets a recorded range still be re-read, so +// lowering it would give back a guarantee already written to History. +func (s *Stream) AdvanceConsumer(_ chasm.MutableContext, consumerID string, offset int64) { + consumer, ok := s.State.Consumers[consumerID] + if !ok || offset <= consumer.Offset { + return + } + consumer.Offset = offset +} + +// DeregisterConsumer releases the floor a consumer was holding, so retention +// and the record cap can reach its records again. The entry stays, inactive, +// so the same consumer coming back is refused if the stream has moved past +// what its History refers to. +func (s *Stream) DeregisterConsumer(_ chasm.MutableContext, consumerID string) { + if consumer, ok := s.State.Consumers[consumerID]; ok { + consumer.Active = false + } +} + +// ForgetConsumer drops a consumer whose run is closed, releasing its floor. +// +// A closed run never takes another workflow task, so nothing will ask for a +// range it has not already been given. It can still be replayed: a query +// against a completed consumer runs on a worker that may never have seen it, +// and a reset can branch from it. Neither is served once truncation has taken +// the bytes, so both are refused with the offset the stream now starts at +// rather than answered with a hole. +func (s *Stream) ForgetConsumer(_ chasm.MutableContext, consumerID string) { + delete(s.State.Consumers, consumerID) +} + +// settleKinds returns the batch with every unspecified kind read as data. A +// producer that never heard of kinds leaves the field at its zero value, and +// every reader would otherwise have to agree on what that means. +// +// The records belong to the caller's request, which on the command path is the +// worker's own proto, so a record that needs settling is copied rather than +// written through. Settled before the batch is marshalled, so a retry hashes +// the same bytes. +func settleKinds(records []*streamlib.StreamRecord) []*streamlib.StreamRecord { + out := make([]*streamlib.StreamRecord, len(records)) + for i, m := range records { + if m.GetKind() != streampb.STREAM_RECORD_KIND_UNSPECIFIED { + out[i] = m + continue + } + settled := common.CloneProto(m) + settled.Kind = streampb.STREAM_RECORD_KIND_DATA + out[i] = settled + } + return out +} + +func marshalBatch(records []*streamlib.StreamRecord) (*commonpb.DataBlob, error) { + // The serialized batch is also the producer's deduplication fingerprint, and + // protobuf map iteration order is not stable. A record carrying payload or + // record metadata would otherwise hash differently on a retry and be + // refused as a conflicting duplicate of itself. + data, err := (proto.MarshalOptions{Deterministic: true}).Marshal( + &streamlib.StreamRecordBatch{Records: records}) + if err != nil { + return nil, err + } + return &commonpb.DataBlob{ + EncodingType: enumspb.ENCODING_TYPE_PROTO3, + Data: data, + }, nil +} + +func contentHash(data []byte) []byte { + sum := sha256.Sum256(data) + return sum[:] +} diff --git a/chasm/lib/stream/stream_test.go b/chasm/lib/stream/stream_test.go new file mode 100644 index 0000000000..d9dc9ca040 --- /dev/null +++ b/chasm/lib/stream/stream_test.go @@ -0,0 +1,714 @@ +package stream + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/api/serviceerror" + streampb "go.temporal.io/api/stream/v1" + "go.temporal.io/server/chasm" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + "go.temporal.io/server/common/payload" + "google.golang.org/protobuf/types/known/durationpb" +) + +// Built directly rather than through NewStream: these exercise state +// transitions, and NewStream also wires a visibility field that needs a live +// context. Construction through the real path is covered end to end in +// tests/stream_test.go. +func newTestStream(t *testing.T) *Stream { + t.Helper() + return &Stream{ + State: &streamlib.StreamState{ + Producers: make(map[string]*streamlib.ProducerCursor), + Consumers: make(map[string]*streamlib.ConsumerCursor), + }, + } +} + +func msgs(bodies ...string) []*streamlib.StreamRecord { + out := make([]*streamlib.StreamRecord, len(bodies)) + for i, b := range bodies { + out[i] = &streamlib.StreamRecord{ + Body: &commonpb.Payload{Data: []byte(b)}, + Kind: streampb.STREAM_RECORD_KIND_DATA, + } + } + return out +} + +func TestAddMessagesAssignsContiguousOffsets(t *testing.T) { + s := newTestStream(t) + + first, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c")}) + require.NoError(t, err) + require.Equal(t, int64(0), first.FirstOffset) + require.Equal(t, int64(3), first.Count) + require.Equal(t, int64(3), first.NextOffset) + + second, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("d", "e")}) + require.NoError(t, err) + require.Equal(t, int64(3), second.FirstOffset) + require.Equal(t, int64(5), second.NextOffset) + require.Equal(t, int64(5), s.State.HeadOffset) +} + +func TestAddMessagesWritesTheBatchIntoTheComponent(t *testing.T) { + s := newTestStream(t) + + res, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + require.NotEmpty(t, res.Blob.Data) + + // Keyed by the offset it starts at, which is the key a retry of this + // append writes under, so the retry replaces rather than races. + require.Len(t, s.Batches, 1) + _, ok := s.Batches[0] + require.True(t, ok, "the batch must be keyed by its first offset") + require.Equal(t, int64(2), s.State.HeadOffset) +} + +func TestDedupReturnsOriginalOffsets(t *testing.T) { + s := newTestStream(t) + req := AddMessagesRequest{Records: msgs("a", "b"), ProducerID: "p1", Sequence: 1} + + first, err := s.AddMessages(nil, req) + require.NoError(t, err) + require.False(t, first.Deduplicated) + + retry := req + again, err := s.AddMessages(nil, retry) + require.NoError(t, err) + require.True(t, again.Deduplicated) + require.Equal(t, first.FirstOffset, again.FirstOffset) + require.Nil(t, again.Blob, "a deduplicated retry writes nothing") + require.Equal(t, int64(2), s.State.HeadOffset, "a retry must not advance the head") +} + +func TestDedupIgnoresPayloadMetadataMapOrder(t *testing.T) { + s := newTestStream(t) + for attempt := range 32 { + body := &commonpb.Payload{ + Data: []byte("encoded"), + Metadata: map[string][]byte{ + "encoding": []byte("test/envelope"), + "key-id": []byte("key-1"), + "nonce": []byte("fixed-for-this-record"), + }, + } + result, err := s.AddMessages(nil, AddMessagesRequest{ + Records: []*streamlib.StreamRecord{{ + Kind: streampb.STREAM_RECORD_KIND_DATA, + Body: body, + Metadata: map[string]*commonpb.Payload{"attempt": body, "checkpoint": body}, + }}, + ProducerID: "encoded-producer", Sequence: 1, + }) + require.NoError(t, err, "identical retry %d must ignore protobuf map iteration order", attempt) + require.Equal(t, attempt > 0, result.Deduplicated) + require.Equal(t, int64(0), result.FirstOffset) + require.Equal(t, int64(1), s.State.HeadOffset) + } +} + +func TestDedupRejectsDifferentContent(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("a"), ProducerID: "p1", Sequence: 1, + }) + require.NoError(t, err) + + // Returning the recorded offsets here would report success while dropping + // the caller's data, which is worse than failing. + _, err = s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("different"), ProducerID: "p1", Sequence: 1, + }) + require.Error(t, err) + require.Contains(t, err.Error(), "different content") +} + +func TestExpectedOffsetMismatchReportsHead(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a")}) + require.NoError(t, err) + + stale := int64(0) + _, err = s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("b"), ExpectedOffset: &stale, + }) + require.Error(t, err) + require.Contains(t, err.Error(), "stream head is 1") +} + +func TestFinishWritingFencesOneProducerOnly(t *testing.T) { + s := newTestStream(t) + require.NoError(t, s.FinishWriting(nil, "p1")) + + _, err := s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("a"), ProducerID: "p1", Sequence: 1, + }) + require.Error(t, err) + + // Another producer is unaffected: finishing is per-producer, not a close. + _, err = s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("a"), ProducerID: "p2", Sequence: 1, + }) + require.NoError(t, err) + require.False(t, s.State.Closed) +} + +func TestCloseRejectsFurtherAppends(t *testing.T) { + s := newTestStream(t) + s.Close(time.Now(), nil) + + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a")}) + require.Error(t, err) + var precondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &precondition) +} + +func TestReadSpansBatchesAndStartsAtTheBatchHoldingTheOffset(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c")}) + require.NoError(t, err) + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("d", "e")}) + require.NoError(t, err) + require.Len(t, s.Batches, 2) + + // A read from offset 1 lands inside the first batch. It gets that batch + // whole, because a consumer asks for an offset and not for a batch, and + // the batch is the smallest thing stored. + blobs, starts, err := s.ReadBatches(nil, 1, 5, 0) + require.NoError(t, err) + require.Len(t, blobs, 2) + require.Equal(t, []int64{0, 3}, starts) + + // A read wholly inside the second batch does not drag the first along. + blobs, starts, err = s.ReadBatches(nil, 3, 5, 0) + require.NoError(t, err) + require.Len(t, blobs, 1) + require.Equal(t, []int64{3}, starts) +} + +func TestReclaimDropsOnlyBatchesFullyBelowTheFloor(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c")}) + require.NoError(t, err) + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("d", "e")}) + require.NoError(t, err) + + // The floor lands mid-batch, so that batch stays: offsets above the floor + // are still readable and they live in it. + require.NoError(t, s.Truncate(nil, 1)) + require.Len(t, s.Batches, 2) + + // Now the whole first batch is below the floor and can go. + require.NoError(t, s.Truncate(nil, 3)) + require.Len(t, s.Batches, 1) + _, ok := s.Batches[3] + require.True(t, ok, "the batch holding readable offsets must survive") +} + +func TestTruncateStopsAtAnActiveConsumersReplayFloor(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "wf-1", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + s.AdvanceConsumer(nil, "wf-1", 2) + + // Reading to 2 is exactly what makes offsets 0 and 1 matter: they are in + // this consumer's History and a replay is asked to reproduce them. + err = s.Truncate(nil, 3) + require.ErrorContains(t, err, "still depends on offset 0") + require.Equal(t, int64(0), s.State.BaseOffset) + + // Whatever is above the floor is still spare capacity. + require.NoError(t, s.Truncate(nil, 0)) +} + +func TestTruncateBounds(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + + err = s.Truncate(nil, 1) + require.NoError(t, err) + err = s.Truncate(nil, 0) + require.Error(t, err, "truncation must not go backwards") + err = s.Truncate(nil, 3) + require.Error(t, err, "truncation must not pass the head") +} + +func TestCapTruncatesInline(t *testing.T) { + s := newTestStream(t) + s.State.Lifecycle = &streamlib.StreamLifecycle{MaxItems: 4} + + for range 4 { + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + } + + // Eight appended, four retained, so the floor sits at 4 and the first two + // batches are entirely below it. + require.Equal(t, int64(8), s.State.HeadOffset) + require.Equal(t, int64(4), s.State.BaseOffset) +} + +func TestCapRefusesAnAppendItCouldOnlyAbsorbByDroppingReadRecords(t *testing.T) { + s := newTestStream(t) + s.State.Lifecycle = &streamlib.StreamLifecycle{MaxItems: 2} + _, err := s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "wf-1", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.ErrorContains(t, err, "still depends on offset 0") + require.Equal(t, int64(0), s.State.HeadOffset, "a refused append writes nothing") + + // Nothing is stuck. The consumer going away is what makes room, and it is + // something someone does rather than something that happens quietly. + s.DeregisterConsumer(nil, "wf-1") + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + require.Equal(t, int64(2), s.State.BaseOffset, "the cap applies once nobody needs the bytes") +} + +func TestCloseSchedulesRetentionOnlyWhenConfigured(t *testing.T) { + now := time.Now() + + plain := newTestStream(t) + require.True(t, plain.Close(now, nil).IsZero(), "no retention configured, nothing to schedule") + + withRetention := newTestStream(t) + withRetention.State.Lifecycle = &streamlib.StreamLifecycle{ + Retention: durationpb.New(time.Hour), + } + at := withRetention.Close(now, nil) + require.Equal(t, now.Add(time.Hour), at) + require.NotNil(t, withRetention.State.CloseTime) + + // Closing twice must not re-arm deletion. + require.True(t, withRetention.Close(now, nil).IsZero()) +} + +func TestRegisterConsumerPinsFromWhereItSubscribed(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + + // Subscribing at 2 says nothing about offsets 0 and 1, so those stay + // droppable and everything from 2 up does not. + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 2, + }) + require.NoError(t, err) + + require.NoError(t, s.Truncate(nil, 2)) + require.Equal(t, int64(2), s.State.BaseOffset) + require.ErrorContains(t, s.Truncate(nil, 3), "still depends on offset 2") +} + +func TestAdvanceConsumerTracksWhereAConsumerHasReached(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + + // The read position decides whether this consumer is worth waking. The + // replay floor, not this, is what retention respects. + s.AdvanceConsumer(nil, "workflow:output", 3) + require.Equal(t, int64(3), s.State.Consumers["workflow:output"].Offset) +} + +// Lowering the pin would hand back a guarantee already written to History: a +// recorded range has to stay re-readable. +func TestAdvanceConsumerNeverRewinds(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + + s.AdvanceConsumer(nil, "workflow:output", 3) + s.AdvanceConsumer(nil, "workflow:output", 1) + + require.Equal(t, int64(3), s.State.Consumers["workflow:output"].GetOffset()) +} + +func TestRegisterConsumerRejectsAnOffsetBelowTheFloor(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + err = s.Truncate(nil, 2) + require.NoError(t, err) + + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 1, + }) + require.ErrorContains(t, err, "below the stream's floor") +} + +// Resubscribing reactivates the existing pin rather than resetting it, so a +// consumer cannot rewind its own floor by subscribing again. +func TestRegisterConsumerTwiceKeepsThePin(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + s.AdvanceConsumer(nil, "workflow:output", 3) + + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + + require.Equal(t, int64(3), s.State.Consumers["workflow:output"].GetOffset()) +} + +func TestDeregisterConsumerReleasesThePin(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 1, + }) + require.NoError(t, err) + + s.DeregisterConsumer(nil, "workflow:output") + + err = s.Truncate(nil, 4) + require.NoError(t, err) +} + +func TestMessageCapStillAppliesWithNoConsumerToProtect(t *testing.T) { + s := newTestStream(t) + s.State.Lifecycle = &streamlib.StreamLifecycle{MaxItems: 2} + + for range 3 { + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + } + + // The refusal is about a consumer's recovery, so a stream with none behaves + // exactly as a capped log should. + require.Equal(t, int64(6), s.State.HeadOffset) + require.Equal(t, int64(4), s.State.BaseOffset) +} + +// A consumer that arrives after the messages were written cannot make the cap +// retroactively wrong, so the clamp keeps its bytes and the stream sits over +// its cap until it goes away. +func TestCapClampsToAConsumerThatRegisteredLate(t *testing.T) { + s := newTestStream(t) + s.State.Lifecycle = &streamlib.StreamLifecycle{MaxItems: 2} + + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + s.State.Lifecycle = &streamlib.StreamLifecycle{MaxItems: 1} + s.applyCap() + + require.Equal(t, int64(0), s.State.BaseOffset, "the clamp keeps what the consumer needs") +} + +// The reason the pin was taken out in the first place. It must not come back: +// a consumer that finished has to stop holding storage. +func TestAConsumerThatDeregisteredHoldsNothing(t *testing.T) { + s := newTestStream(t) + s.State.Lifecycle = &streamlib.StreamLifecycle{MaxItems: 2} + _, err := s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + s.DeregisterConsumer(nil, "workflow:output") + + for range 3 { + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + } + require.Equal(t, int64(4), s.State.BaseOffset) +} + +// Coming back to a stream that moved past what its History refers to has to be +// refused at registration, which is the last moment before the workflow +// depends on those offsets again. +func TestReregisteringBelowTheFloorIsRefused(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 0, + }) + require.NoError(t, err) + s.DeregisterConsumer(nil, "workflow:output") + require.NoError(t, s.Truncate(nil, 2)) + + // Resubscribing further along does not repair the gap. What this consumer + // already recorded starts at 0, and offsets 0 and 1 are gone. + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:output", WorkflowID: "wf-1", RunID: "run-1", Offset: 3, + }) + require.ErrorContains(t, err, "the stream now starts at 2") +} + +// A caller sending a fresh producer id per request would otherwise grow the +// component state until no append fits, which leaves the stream unwritable for +// good rather than failing the call that caused it. +func TestStreamProducerTableIsBounded(t *testing.T) { + s := newTestStream(t) + + for i := range MaxProducersPerStream { + _, err := s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("m"), + ProducerID: fmt.Sprintf("p%d", i), + Sequence: 1, + }) + require.NoError(t, err) + } + + _, err := s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("one too many"), + ProducerID: "p-over", + Sequence: 1, + }) + var invalid *serviceerror.InvalidArgument + require.ErrorAs(t, err, &invalid) + + // A producer already tracked keeps working, so the cap cannot wedge the + // producers that filled it. + _, err = s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("still fine"), + ProducerID: "p0", + Sequence: 2, + }) + require.NoError(t, err) + + // An anonymous append is never blocked by the table. + _, err = s.AddMessages(nil, AddMessagesRequest{ + Records: msgs("anon"), + }) + require.NoError(t, err) +} + +// Each consumer holds a truncation floor, so an unbounded table pins storage as +// well as growing state. +func TestStreamConsumerTableIsBounded(t *testing.T) { + s := newTestStream(t) + + for i := range MaxConsumersPerStream { + _, err := s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: fmt.Sprintf("c%d", i), WorkflowID: "wf", RunID: "run", External: true, + }) + require.NoError(t, err) + } + + _, err := s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "c-over", WorkflowID: "wf", RunID: "run", Offset: 0, External: true, + }) + var invalid *serviceerror.InvalidArgument + require.ErrorAs(t, err, &invalid) + + // Re-registering an existing consumer is an update, not a new entry. + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "c0", WorkflowID: "wf", RunID: "run", Offset: 0, External: true, + }) + require.NoError(t, err) +} + +// A producer that leaves the kind unset means data. Delivery to a workflow +// drops anything that is not data, so without this the message would take an +// offset and never be seen by a subscriber. +func TestAddMessagesTreatsAnUnsetKindAsData(t *testing.T) { + s := newTestStream(t) + unset := []*streamlib.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("a")}}, + {Body: &commonpb.Payload{Data: []byte("b")}}, + } + _, err := s.AddMessages(nil, AddMessagesRequest{ + Records: unset, ProducerID: "p", Sequence: 1, + }) + require.NoError(t, err) + + blobs, starts, err := s.ReadBatches(nil, 0, 2, 0) + require.NoError(t, err) + collected, _, err := CollectRecords(blobs, starts, 0, 2, 10, nil) + require.NoError(t, err) + require.Len(t, ToAPIRecords(collected), 2, "both messages must reach a subscriber") + + // The retry carries the same unset kind and has to hash the same. + retry, err := s.AddMessages(nil, AddMessagesRequest{ + Records: []*streamlib.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("a")}}, + {Body: &commonpb.Payload{Data: []byte("b")}}, + }, + ProducerID: "p", Sequence: 1, + }) + require.NoError(t, err) + require.True(t, retry.Deduplicated) +} + +// A budgeted stream refuses the append that would not fit rather than dropping +// older messages: its batches are someone's mutable state, and the alternative +// to refusing is the execution size limit terminating that workflow later. +func TestBudgetRefusesAnAppendThatDoesNotFit(t *testing.T) { + s := newTestStream(t) + s.State.Budget = &streamlib.StreamBudget{MaxItems: 3} + + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b")}) + require.NoError(t, err) + + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("c", "d")}) + var exhausted *serviceerror.ResourceExhausted + require.ErrorAs(t, err, &exhausted) + require.Equal(t, int64(2), s.State.HeadOffset, "a refused append writes nothing") + + // The last slot is still there for an append that fits. + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("c")}) + require.NoError(t, err) + + bytesOnly := newTestStream(t) + bytesOnly.State.Budget = &streamlib.StreamBudget{MaxBytes: 16} + _, err = bytesOnly.AddMessages(nil, AddMessagesRequest{Records: msgs("small")}) + require.NoError(t, err) + _, err = bytesOnly.AddMessages(nil, AddMessagesRequest{Records: msgs("another one")}) + require.ErrorAs(t, err, &exhausted) + require.Equal(t, int64(1), bytesOnly.State.HeadOffset) +} + +// One workflow id has one open run, so a pin from another run of the same +// workflow belongs to a run that finished. Registering the new run drops it, +// which is what lets the new run start at its own offset instead of inheriting +// a floor that may already be below the stream's base. +func TestRegisterConsumerReplacesAnEntryFromAnotherRun(t *testing.T) { + s := newTestStream(t) + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d")}) + require.NoError(t, err) + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:wf/run-1", WorkflowID: "wf", RunID: "run-1", Offset: 0, External: true, + }) + require.NoError(t, err) + + start, err := s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:wf/run-2", WorkflowID: "wf", RunID: "run-2", Offset: 3, External: true, + }) + require.NoError(t, err) + require.Equal(t, int64(3), start) + require.Len(t, s.State.Consumers, 1) + require.Equal(t, "run-2", s.State.Consumers["workflow:wf/run-2"].GetRunId()) + require.NoError(t, s.Truncate(nil, 3), "only the new run's floor holds") + + // A different workflow id is not the same consumer and keeps its pin. + _, err = s.RegisterConsumer(nil, ConsumerRegistration{ + ConsumerID: "workflow:other/run-9", WorkflowID: "other", RunID: "run-9", + Offset: 3, External: true, + }) + require.NoError(t, err) + require.Len(t, s.State.Consumers, 2) +} + +// Every append that leaves an external consumer behind would otherwise queue a +// task of its own. One outstanding task carries every append that lands before +// it runs, and the task's own read lowers the flag before the next one is owed. +func TestNotifyCoalescesIntoOneOutstandingTask(t *testing.T) { + mctx := &chasm.MockMutableContext{MockContext: chasm.MockContext{ + HandleNow: func(chasm.Component) time.Time { return time.Unix(0, 0) }, + }} + s := newTestStream(t) + s.Batches = make(chasm.Map[int64, *commonpb.DataBlob]) + _, err := s.RegisterConsumer(mctx, ConsumerRegistration{ + ConsumerID: "workflow:wf/run-1", WorkflowID: "wf", RunID: "run-1", Offset: 0, External: true, + }) + require.NoError(t, err) + + _, err = s.AddMessages(mctx, AddMessagesRequest{Records: msgs("a")}) + require.NoError(t, err) + _, err = s.AddMessages(mctx, AddMessagesRequest{Records: msgs("b")}) + require.NoError(t, err) + require.Len(t, mctx.Tasks, 1, "the second append rides the task the first scheduled") + + // The task's read lowers the flag, so the next append owes a new task. + state, err := s.TakeNotifySnapshot(mctx, struct{}{}) + require.NoError(t, err) + require.Equal(t, int64(2), state.GetHeadOffset(), + "the task sees every append that landed before it") + _, err = s.AddMessages(mctx, AddMessagesRequest{Records: msgs("c")}) + require.NoError(t, err) + require.Len(t, mctx.Tasks, 2) +} + +// The budget bounds what the stream holds, and offsets are global: a stream +// whose floor has moved, or one that begins above zero, holds fewer records +// than its head offset names. Measuring the head instead puts such a stream +// over budget the moment it exists. +func TestBudgetCountsHeldRecordsRatherThanTheHeadOffset(t *testing.T) { + s := newTestStream(t) + s.State.Budget = &streamlib.StreamBudget{MaxItems: 5} + + _, err := s.AddMessages(nil, AddMessagesRequest{Records: msgs("a", "b", "c", "d", "e")}) + require.NoError(t, err) + + require.NoError(t, s.Truncate(nil, 5)) + require.Equal(t, int64(5), s.State.BaseOffset) + + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("f", "g", "h")}) + require.NoError(t, err, "a truncated stream holds nothing and has its whole budget free") + require.Equal(t, int64(8), s.State.HeadOffset) + + // Still bounded: three held plus three more is over the budget of five. + _, err = s.AddMessages(nil, AddMessagesRequest{Records: msgs("i", "j", "k")}) + var exhausted *serviceerror.ResourceExhausted + require.ErrorAs(t, err, &exhausted) +} + +// A close reason with no encoding metadata is not decodable by any SDK data +// converter, so it would reach a reader as opaque bytes. +func TestTerminateEncodesTheCloseReason(t *testing.T) { + mctx := &chasm.MockMutableContext{MockContext: chasm.MockContext{ + HandleNow: func(chasm.Component) time.Time { return time.Unix(0, 0) }, + }} + s := newTestStream(t) + _, err := s.Terminate(mctx, chasm.TerminateComponentRequest{Reason: "operator asked"}) + require.NoError(t, err) + + require.True(t, s.State.Closed) + var reason string + require.NoError(t, payload.Decode(s.State.GetCloseReason(), &reason)) + require.Equal(t, "operator asked", reason) +} + +// The records belong to the caller's request, which on the command path is the +// worker's own proto. Settling the kind through them would edit it. +func TestAddMessagesDoesNotSettleTheKindOnTheCallersRecords(t *testing.T) { + s := newTestStream(t) + caller := []*streamlib.StreamRecord{{Body: &commonpb.Payload{Data: []byte("a")}}} + + _, err := s.AddMessages(nil, AddMessagesRequest{Records: caller}) + require.NoError(t, err) + require.Equal(t, streampb.STREAM_RECORD_KIND_UNSPECIFIED, caller[0].GetKind()) + + blobs, starts, err := s.ReadBatches(nil, 0, 1, 0) + require.NoError(t, err) + collected, _, err := CollectRecords(blobs, starts, 0, 1, 10, nil) + require.NoError(t, err) + require.Equal(t, streampb.STREAM_RECORD_KIND_DATA, collected[0].GetKind(), + "the stored record still reads as data") +}