From 569502f47de2ab3b093647ba19fbb7d071c4399e Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 08:06:48 -0700 Subject: [PATCH 1/3] Added the CHASM stream component with floor-aware retention. Records live in the component next to the workflow that owns them, so a publish rides the same commit. Retention, truncation and the consumer floor are the component's own state, and the caps that bound resource use come from namespace-scoped dynamic config. --- chasm/lib/stream/config.go | 244 ++++++ chasm/lib/stream/cursor.go | 141 ++++ chasm/lib/stream/cursor_test.go | 100 +++ .../gen/streampb/v1/message.go-helpers.pb.go | 80 ++ .../lib/stream/gen/streampb/v1/message.pb.go | 265 ++++++ .../streampb/v1/stream_state.go-helpers.pb.go | 228 ++++++ .../stream/gen/streampb/v1/stream_state.pb.go | 706 ++++++++++++++++ .../gen/streampb/v1/tasks.go-helpers.pb.go | 80 ++ chasm/lib/stream/gen/streampb/v1/tasks.pb.go | 160 ++++ chasm/lib/stream/messages.go | 126 +++ chasm/lib/stream/messages_test.go | 91 +++ chasm/lib/stream/proto/v1/message.proto | 42 + chasm/lib/stream/proto/v1/stream_state.proto | 136 ++++ chasm/lib/stream/proto/v1/tasks.proto | 17 + chasm/lib/stream/stream.go | 768 ++++++++++++++++++ chasm/lib/stream/stream_test.go | 655 +++++++++++++++ 16 files changed, 3839 insertions(+) create mode 100644 chasm/lib/stream/config.go create mode 100644 chasm/lib/stream/cursor.go create mode 100644 chasm/lib/stream/cursor_test.go create mode 100644 chasm/lib/stream/gen/streampb/v1/message.go-helpers.pb.go create mode 100644 chasm/lib/stream/gen/streampb/v1/message.pb.go create mode 100644 chasm/lib/stream/gen/streampb/v1/stream_state.go-helpers.pb.go create mode 100644 chasm/lib/stream/gen/streampb/v1/stream_state.pb.go create mode 100644 chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go create mode 100644 chasm/lib/stream/gen/streampb/v1/tasks.pb.go create mode 100644 chasm/lib/stream/messages.go create mode 100644 chasm/lib/stream/messages_test.go create mode 100644 chasm/lib/stream/proto/v1/message.proto create mode 100644 chasm/lib/stream/proto/v1/stream_state.proto create mode 100644 chasm/lib/stream/proto/v1/tasks.proto create mode 100644 chasm/lib/stream/stream.go create mode 100644 chasm/lib/stream/stream_test.go diff --git a/chasm/lib/stream/config.go b/chasm/lib/stream/config.go new file mode 100644 index 00000000000..cb7ac2cd6d6 --- /dev/null +++ b/chasm/lib/stream/config.go @@ -0,0 +1,244 @@ +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. +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 00000000000..adb433ce709 --- /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 00000000000..deb2b95212c --- /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 00000000000..ef697410810 --- /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 00000000000..400dc3c16b4 --- /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, or -1 when unnumbered. 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 00000000000..c97d64015d9 --- /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 00000000000..ac663fb790d --- /dev/null +++ b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go @@ -0,0 +1,706 @@ +// 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 never reclaims, so this is also what it holds. + 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. + 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 00000000000..9644b7bbe02 --- /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 00000000000..2b944c7d24d --- /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 00000000000..ee7fd463a8c --- /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: -1, + 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(-1), 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 00000000000..d940f769789 --- /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, or -1 when unnumbered. 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 00000000000..2ea9beaec59 --- /dev/null +++ b/chasm/lib/stream/proto/v1/stream_state.proto @@ -0,0 +1,136 @@ +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 never reclaims, so this is also what it holds. + 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. + 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 00000000000..c4b8e5e6c6a --- /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 00000000000..f54a41f5fcd --- /dev/null +++ b/chasm/lib/stream/stream.go @@ -0,0 +1,768 @@ +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" + "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. + 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) { + reason := &commonpb.Payload{Data: []byte(req.Reason)} + return chasm.TerminateComponentResponse{}, s.CloseAndSchedule(mctx, 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 + } + // A record with no kind is 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. Settled before the batch is marshalled, so a + // retry hashes the same bytes. + for _, m := range req.Records { + if m.GetKind() == streampb.STREAM_RECORD_KIND_UNSPECIFIED { + m.Kind = streampb.STREAM_RECORD_KIND_DATA + } + } + + blob, err := marshalBatch(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 +} + +// 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.State.HeadOffset+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.State.HeadOffset, 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 is %d", req.Sequence, 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. +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. A closed run never +// replays and never comes back, so unlike a deregistration there is nothing +// left to refuse later. +func (s *Stream) ForgetConsumer(_ chasm.MutableContext, consumerID string) { + delete(s.State.Consumers, consumerID) +} + +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 00000000000..9438b8bc435 --- /dev/null +++ b/chasm/lib/stream/stream_test.go @@ -0,0 +1,655 @@ +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" + "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) +} From 36fb344bc64ef9cbd9c8abb8730c359c88cd8d6e Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 12:57:48 -0700 Subject: [PATCH 2/3] Measured the stream budget against what a stream holds. The item budget compared an absolute head offset against a count, so a stream whose floor had moved, or one created above zero, was over budget the moment it existed. The close reason now carries its encoding, and the record kind is settled on a copy rather than through the caller's protos. --- chasm/lib/stream/config.go | 4 + .../stream/gen/streampb/v1/stream_state.pb.go | 11 ++- chasm/lib/stream/proto/v1/stream_state.proto | 11 ++- chasm/lib/stream/stream.go | 81 ++++++++++++++----- chasm/lib/stream/stream_test.go | 59 ++++++++++++++ 5 files changed, 145 insertions(+), 21 deletions(-) diff --git a/chasm/lib/stream/config.go b/chasm/lib/stream/config.go index cb7ac2cd6d6..f4d10083133 100644 --- a/chasm/lib/stream/config.go +++ b/chasm/lib/stream/config.go @@ -225,6 +225,10 @@ func DefaultLimits() Limits { // 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 { diff --git a/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go index ac663fb790d..223672a1192 100644 --- a/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go +++ b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go @@ -49,7 +49,10 @@ type StreamState struct { // 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 never reclaims, so this is also what it holds. + // 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 @@ -527,6 +530,12 @@ type StreamLifecycle struct { 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 diff --git a/chasm/lib/stream/proto/v1/stream_state.proto b/chasm/lib/stream/proto/v1/stream_state.proto index 2ea9beaec59..70cb9c9f671 100644 --- a/chasm/lib/stream/proto/v1/stream_state.proto +++ b/chasm/lib/stream/proto/v1/stream_state.proto @@ -42,7 +42,10 @@ message StreamState { StreamBudget budget = 14; // Bytes appended over the stream's life, kept for the budget check. A stream - // with a budget never reclaims, so this is also what it holds. + // 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 @@ -132,5 +135,11 @@ message StreamLifecycle { 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/stream.go b/chasm/lib/stream/stream.go index f54a41f5fcd..dd3bd6b1905 100644 --- a/chasm/lib/stream/stream.go +++ b/chasm/lib/stream/stream.go @@ -14,6 +14,7 @@ import ( "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" ) @@ -63,6 +64,11 @@ type AddMessagesRequest struct { // 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 @@ -116,8 +122,11 @@ func (s *Stream) Terminate( mctx chasm.MutableContext, req chasm.TerminateComponentRequest, ) (chasm.TerminateComponentResponse, error) { - reason := &commonpb.Payload{Data: []byte(req.Reason)} - return chasm.TerminateComponentResponse{}, s.CloseAndSchedule(mctx, reason) + // 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 @@ -154,17 +163,7 @@ func (s *Stream) AddMessages( if err := checkBatchBytes(req.Records, limits); err != nil { return AddMessagesResult{}, err } - // A record with no kind is 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. Settled before the batch is marshalled, so a - // retry hashes the same bytes. - for _, m := range req.Records { - if m.GetKind() == streampb.STREAM_RECORD_KIND_UNSPECIFIED { - m.Kind = streampb.STREAM_RECORD_KIND_DATA - } - } - - blob, err := marshalBatch(req.Records) + blob, err := marshalBatch(settleKinds(req.Records)) if err != nil { return AddMessagesResult{}, err } @@ -297,6 +296,13 @@ func (s *Stream) checkProducerRoom(producerID string, maxProducers int) error { 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 @@ -308,11 +314,11 @@ func (s *Stream) checkBudget(count int64, size int64) error { if budget == nil { return nil } - if limit := budget.GetMaxItems(); limit > 0 && s.State.HeadOffset+count > limit { + 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.State.HeadOffset, limit, count) + s.held(), limit, count) } if limit := budget.GetMaxBytes(); limit > 0 && s.State.AppendedBytes+size > limit { return serviceerror.NewResourceExhaustedf( @@ -342,7 +348,9 @@ func (s *Stream) checkProducer(req AddMessagesRequest, hash []byte) (*AddMessage } if req.Sequence < cursor.Seq { return nil, serviceerror.NewInvalidArgumentf( - "stale producer sequence %d, last accepted is %d", req.Sequence, cursor.Seq) + "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 @@ -616,6 +624,14 @@ func (s *Stream) applyCap() { // 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 { @@ -739,13 +755,40 @@ func (s *Stream) DeregisterConsumer(_ chasm.MutableContext, consumerID string) { } } -// ForgetConsumer drops a consumer whose run is closed. A closed run never -// replays and never comes back, so unlike a deregistration there is nothing -// left to refuse later. +// 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 diff --git a/chasm/lib/stream/stream_test.go b/chasm/lib/stream/stream_test.go index 9438b8bc435..d9dc9ca040b 100644 --- a/chasm/lib/stream/stream_test.go +++ b/chasm/lib/stream/stream_test.go @@ -11,6 +11,7 @@ import ( 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" ) @@ -653,3 +654,61 @@ func TestNotifyCoalescesIntoOneOutstandingTask(t *testing.T) { 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") +} From 7f4da7c41bb33330d4afb3229ab6e6ee73b755b3 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 16:18:48 -0700 Subject: [PATCH 3/3] Freed zero to mean a producer that does not number. The stored record mirrors the public one field for field, and the public one no longer reserves -1 for an unnumbered record. --- chasm/lib/stream/gen/streampb/v1/message.pb.go | 4 ++-- chasm/lib/stream/messages_test.go | 4 ++-- chasm/lib/stream/proto/v1/message.proto | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/chasm/lib/stream/gen/streampb/v1/message.pb.go b/chasm/lib/stream/gen/streampb/v1/message.pb.go index 400dc3c16b4..bce3b3730a2 100644 --- a/chasm/lib/stream/gen/streampb/v1/message.pb.go +++ b/chasm/lib/stream/gen/streampb/v1/message.pb.go @@ -33,8 +33,8 @@ type StreamRecord struct { // 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, or -1 when unnumbered. Stored - // as sent; the global offset is what orders the stream. + // 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. diff --git a/chasm/lib/stream/messages_test.go b/chasm/lib/stream/messages_test.go index ee7fd463a8c..bd18bb05b35 100644 --- a/chasm/lib/stream/messages_test.go +++ b/chasm/lib/stream/messages_test.go @@ -68,7 +68,7 @@ func TestToAPIRecordsCarriesTheRecordAsWritten(t *testing.T) { Kind: streampb.STREAM_RECORD_KIND_FINISH, ProducerId: "model-call", Attempt: 2, - Sequence: -1, + Sequence: 8, Offset: 42, }, } @@ -87,5 +87,5 @@ func TestToAPIRecordsCarriesTheRecordAsWritten(t *testing.T) { 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(-1), got[1].GetSequence()) + 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 index d940f769789..fc7cdd00f3b 100644 --- a/chasm/lib/stream/proto/v1/message.proto +++ b/chasm/lib/stream/proto/v1/message.proto @@ -15,8 +15,8 @@ message StreamRecord { // Producer-supplied provenance, stored as sent. map metadata = 2; string topic = 3; - // The producer's position within its attempt, or -1 when unnumbered. Stored - // as sent; the global offset is what orders the stream. + // 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.