diff --git a/chasm/lib/stream/config.go b/chasm/lib/stream/config.go index ed535d2c1c5..45dfa50beec 100644 --- a/chasm/lib/stream/config.go +++ b/chasm/lib/stream/config.go @@ -30,6 +30,12 @@ const LongPollBuffer = 3 * time.Second // shard would otherwise stretch the lock hold to match it. const RoutedCallTimeout = 5 * time.Second +// RoutedSetBudget bounds every routed call one workflow task makes while the +// execution's lock is held, taken together. RoutedCallTimeout bounds one of +// them, and a task can carry as many as the subscription limit allows, so +// without this the lock hold grows with that count. +const RoutedSetBudget = 15 * time.Second + // MaxListPageSize bounds a visibility page when the caller does not. const MaxListPageSize = 1000 @@ -96,6 +102,12 @@ const ( // that the byte budget alone does not see. OwnedStreamMaxItems = 10_000 + // MaxSubscriptionsPerWorkflow bounds how many streams one execution + // consumes. Each subscription costs a routed call on the completion path + // that made it and another on every task start, both with the execution's + // lock held, so the count is what bounds the lock hold. + MaxSubscriptionsPerWorkflow = 100 + // OwnedStreamsMaxBytesPerWorkflow bounds every stream one execution owns // taken together. The per-stream budget multiplied by the stream count // comes to far more than limit.mutableStateSize.error, so without this an @@ -163,6 +175,11 @@ under limit.mutableStateSize.error, which would otherwise terminate the workflow OwnedStreamMaxItems, `Message budget of a stream a workflow owns. Appends past it are refused.`, ) + MaxSubscriptionsPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting( + "stream.maxSubscriptionsPerWorkflow", + MaxSubscriptionsPerWorkflow, + `Most streams one workflow execution can consume.`, + ) OwnedStreamsMaxBytesPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting( "stream.ownedStreamsMaxBytesPerWorkflow", OwnedStreamsMaxBytesPerWorkflow, @@ -235,6 +252,7 @@ type Config struct { // Bounds every stream one execution owns taken together, which the // per-stream budget cannot do. OwnedStreamsMaxBytesPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter + MaxSubscriptionsPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter AppendRecordsPerSecond dynamicconfig.IntPropertyFnWithNamespaceFilter AppendBytesPerSecond dynamicconfig.IntPropertyFnWithNamespaceFilter @@ -258,6 +276,7 @@ func NewConfig(dc *dynamicconfig.Collection) *Config { OwnedStreamMaxItems: OwnedStreamMaxItemsSetting.Get(dc), OwnedStreamsMaxBytesPerWorkflow: OwnedStreamsMaxBytesPerWorkflowSetting.Get(dc), + MaxSubscriptionsPerWorkflow: MaxSubscriptionsPerWorkflowSetting.Get(dc), AppendRecordsPerSecond: AppendRecordsPerSecondSetting.Get(dc), AppendBytesPerSecond: AppendBytesPerSecondSetting.Get(dc), @@ -279,6 +298,7 @@ type Limits struct { OwnedStreamMaxItems int OwnedStreamsMaxBytesPerWorkflow int + MaxSubscriptionsPerWorkflow int } // LimitsFor resolves the limits for a namespace. A nil Config, which is what @@ -299,6 +319,7 @@ func (c *Config) LimitsFor(namespaceName string) Limits { OwnedStreamMaxItems: c.OwnedStreamMaxItems(namespaceName), OwnedStreamsMaxBytesPerWorkflow: c.OwnedStreamsMaxBytesPerWorkflow(namespaceName), + MaxSubscriptionsPerWorkflow: c.MaxSubscriptionsPerWorkflow(namespaceName), }.withDefaults() } @@ -343,5 +364,6 @@ func (l Limits) withDefaults() Limits { fill(&l.OwnedStreamMaxBytes, OwnedStreamMaxBytes) fill(&l.OwnedStreamMaxItems, OwnedStreamMaxItems) fill(&l.OwnedStreamsMaxBytesPerWorkflow, OwnedStreamsMaxBytesPerWorkflow) + fill(&l.MaxSubscriptionsPerWorkflow, MaxSubscriptionsPerWorkflow) return l } diff --git a/chasm/lib/stream/service/events.go b/chasm/lib/stream/service/events.go new file mode 100644 index 00000000000..d496181ee3a --- /dev/null +++ b/chasm/lib/stream/service/events.go @@ -0,0 +1,78 @@ +package service + +import ( + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + "go.temporal.io/server/service/history/hsm" +) + +// streamSubscribedEventDefinition tells the history service how to treat the +// event a subscribe command writes. +// +// It applies nothing. The cursor lives in CHASM state, which is persisted and +// rebuilt with the execution, so replication and reset have nothing to +// reconstruct from the event. It exists so the command has an event at all: +// every SDK matches issued commands against command-generated events in order, +// and a command producing none puts that matching out of step. +type streamSubscribedEventDefinition struct{} + +func (streamSubscribedEventDefinition) Type() enumspb.EventType { + return enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED +} + +// Subscribing does not itself give the workflow anything to decide on. The +// range that follows does, and that wakes the workflow through its cursor. +func (streamSubscribedEventDefinition) IsWorkflowTaskTrigger() bool { return false } + +func (streamSubscribedEventDefinition) Apply(*hsm.Node, *historypb.HistoryEvent) error { + return nil +} + +// A command event, so never reapplied onto another branch: a workflow that +// still wants the subscription issues the command again on the new one. +func (streamSubscribedEventDefinition) CherryPick( + *hsm.Node, + *historypb.HistoryEvent, + map[enumspb.ResetReapplyExcludeType]struct{}, +) error { + return hsm.ErrNotCherryPickable +} + +// streamRecordsAppendedEventDefinition tells the history service how to treat +// the event a publish command writes. +// +// Like the subscription event it applies nothing: the stream's frontier is +// CHASM state committed with the workflow task, and the bodies are in the +// stream component. What the event carries is the offset range, which is what +// lets anyone reading History find the batch without History having held it. +type streamRecordsAppendedEventDefinition struct{} + +func (streamRecordsAppendedEventDefinition) Type() enumspb.EventType { + return enumspb.EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED +} + +// A workflow publishing to its own stream has nothing to be woken about. +func (streamRecordsAppendedEventDefinition) IsWorkflowTaskTrigger() bool { return false } + +func (streamRecordsAppendedEventDefinition) Apply(*hsm.Node, *historypb.HistoryEvent) error { + return nil +} + +// A command event, so never reapplied onto another branch. Reapplying would +// claim offsets in a log the new branch never wrote to. +func (streamRecordsAppendedEventDefinition) CherryPick( + *hsm.Node, + *historypb.HistoryEvent, + map[enumspb.ResetReapplyExcludeType]struct{}, +) error { + return hsm.ErrNotCherryPickable +} + +// RegisterEventDefinitions makes the stream events known to the history +// service. +func RegisterEventDefinitions(reg *hsm.Registry) error { + if err := reg.RegisterEventDefinition(streamSubscribedEventDefinition{}); err != nil { + return err + } + return reg.RegisterEventDefinition(streamRecordsAppendedEventDefinition{}) +} diff --git a/chasm/lib/stream/service/fx.go b/chasm/lib/stream/service/fx.go index 030be46eeee..756b4820fc4 100644 --- a/chasm/lib/stream/service/fx.go +++ b/chasm/lib/stream/service/fx.go @@ -23,6 +23,7 @@ var HistoryModule = fx.Module( fx.Invoke(func(l *library, registry *chasm.Registry) error { return registry.Register(l) }), + fx.Invoke(RegisterEventDefinitions), ) var FrontendModule = fx.Module( diff --git a/chasm/lib/workflow/fx.go b/chasm/lib/workflow/fx.go index 676ee077e80..8e69ef2f8b8 100644 --- a/chasm/lib/workflow/fx.go +++ b/chasm/lib/workflow/fx.go @@ -20,12 +20,16 @@ var Module = fx.Module( chasmRegistry *chasm.Registry, library *library, config *nexusoperation.Config, + streamConfig *stream.Config, ) error { if err := library.registry.Register( newNexusLibrary(config, chasmRegistry.NexusEndpointProcessor), ); err != nil { return err } + if err := library.registry.Register(newStreamLibrary(streamConfig)); err != nil { + return err + } return chasmRegistry.Register(library) }), ) diff --git a/chasm/lib/workflow/stream_admission_test.go b/chasm/lib/workflow/stream_admission_test.go index 9a5c19209a1..9a07e88b43c 100644 --- a/chasm/lib/workflow/stream_admission_test.go +++ b/chasm/lib/workflow/stream_admission_test.go @@ -4,7 +4,10 @@ import ( "testing" "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" streampb "go.temporal.io/api/stream/v1" "go.temporal.io/server/chasm" @@ -116,6 +119,51 @@ func TestKnownHeadIsOnlyPushedIntoAnExternalCursor(t *testing.T) { require.ErrorAs(t, err, ¬Found) } +// Each subscription costs a routed call on the completion path that made it, +// with this execution's lock held, and another on every task start. Nothing +// else bounds how many a workflow may hold or how many one task may carry. +func TestSubscriptionsPerWorkflowAreBounded(t *testing.T) { + ctx := newStreamBudgetTestContext() + backend := &chasm.MockNodeBackend{ + HandleAddHistoryEvent: func( + eventType enumspb.EventType, set func(*historypb.HistoryEvent), + ) *historypb.HistoryEvent { + e := &historypb.HistoryEvent{EventType: eventType} + set(e) + return e + }, + } + w := &Workflow{MSPointer: chasm.NewMSPointer(backend)} + opts := CommandHandlerOptions{WorkflowTaskCompletedEventID: 10} + limits := stream.Limits{MaxSubscriptionsPerWorkflow: 2} + + subscribe := func(id string) error { + return handleSubscribeStreamCommand(ctx, w, nil, &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SUBSCRIBE_STREAM, + Attributes: &commandpb.Command_SubscribeStreamCommandAttributes{ + SubscribeStreamCommandAttributes: &commandpb.SubscribeStreamCommandAttributes{ + StreamNameOrId: id, + }, + }, + }, opts, limits) + } + + require.NoError(t, subscribe("a")) + require.NoError(t, subscribe("b")) + + // Counted against what this task has already staged, so one task cannot + // carry an unbounded set of them either. + err := subscribe("c") + var failTask FailWorkflowTaskError + require.ErrorAs(t, err, &failTask) + require.Equal(t, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, failTask.Cause) + + // Subscribing again to one already staged registers nothing new, so it is + // not refused for room. + require.NoError(t, subscribe("a")) +} + // An activity id may contain a slash, so the reserved key escapes it. Without // that two different (activity, name) pairs would land on one stream. func TestActivityStreamKeysDoNotCollide(t *testing.T) { @@ -147,3 +195,38 @@ func TestWorkflowCannotNameAnActivityStream(t *testing.T) { w.OwnedStream(ctx, key).State.GetAppendedBytes()) require.Len(t, w.Streams, 1) } + +// An activity's streams end when it reaches a terminal status, while the +// workflow's own streams and another activity's stay open. +func TestCloseActivityStreamsEndsOnlyThatActivity(t *testing.T) { + ctx := newStreamBudgetTestContext() + w := &Workflow{} + limits := stream.Limits{MaxOwnedStreamsPerWorkflow: 10} + for _, key := range []string{ + DefaultStreamName, + ActivityStreamKey("act", DefaultStreamName), + ActivityStreamKey("act", "reasoning"), + ActivityStreamKey("act-2", DefaultStreamName), + } { + _, err := w.AppendToOwnedStream(ctx, key, stream.AddMessagesRequest{ + Records: budgetTestRecords(1), Limits: limits, + }) + require.NoError(t, err) + } + + require.True(t, w.HasOpenActivityStreams(ctx, "act")) + require.False(t, w.HasOpenActivityStreams(ctx, "never-wrote")) + require.NoError(t, w.CloseActivityStreams(ctx, "act")) + require.False(t, w.HasOpenActivityStreams(ctx, "act")) + + closed := func(key string) bool { return w.OwnedStream(ctx, key).State.GetClosed() } + require.True(t, closed(ActivityStreamKey("act", DefaultStreamName))) + require.True(t, closed(ActivityStreamKey("act", "reasoning"))) + require.False(t, closed(ActivityStreamKey("act-2", DefaultStreamName))) + require.False(t, closed(DefaultStreamName)) + + _, err := w.AppendToOwnedStream(ctx, ActivityStreamKey("act", DefaultStreamName), + stream.AddMessagesRequest{Records: budgetTestRecords(1), Limits: limits}) + var precondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &precondition, "an ended activity's stream takes no more records") +} diff --git a/chasm/lib/workflow/stream_commands.go b/chasm/lib/workflow/stream_commands.go index 6d78a1b73cc..9eb8bfa1efb 100644 --- a/chasm/lib/workflow/stream_commands.go +++ b/chasm/lib/workflow/stream_commands.go @@ -1,12 +1,20 @@ package workflow import ( + "errors" + "fmt" "net/url" + "slices" "strings" + commandpb "go.temporal.io/api/command/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" + streampb "go.temporal.io/api/stream/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" ) // DefaultStreamName is the stream a command addresses when it names none. @@ -33,6 +41,45 @@ func IsActivityStreamKey(name string) bool { return strings.HasPrefix(name, activityStreamPrefix) } +// activityStreamKeys returns the keys of the streams one activity owns, in a +// stable order. Only the keys are read, so no stream is loaded. +func (w *Workflow) activityStreamKeys(activityID string) []string { + prefix := ActivityStreamKey(activityID, "") + var keys []string + for key := range w.Streams { + if strings.HasPrefix(key, prefix) { + keys = append(keys, key) + } + } + slices.Sort(keys) + return keys +} + +// HasOpenActivityStreams reports whether the activity owns a stream that is +// not closed yet. +func (w *Workflow) HasOpenActivityStreams(ctx chasm.Context, activityID string) bool { + for _, key := range w.activityStreamKeys(activityID) { + if !w.Streams[key].Get(ctx).State.GetClosed() { + return true + } + } + return false +} + +// CloseActivityStreams ends every stream the activity owns. Called when the +// activity reaches a terminal status, which is when a reader tailing it has to +// be released: the workflow is still running, so the rule that ends a stream +// with its execution does not reach these. A retry is not terminal, so the +// next attempt keeps writing to the same streams. +func (w *Workflow) CloseActivityStreams(mctx chasm.MutableContext, activityID string) error { + for _, key := range w.activityStreamKeys(activityID) { + if err := w.Streams[key].Get(mctx).CloseAndSchedule(mctx, nil); err != nil { + return err + } + } + return nil +} + // CheckWorkflowStreamName refuses a name the workflow itself may not use, // because it is too long or because it would reach an activity's stream. func CheckWorkflowStreamName(name string) error { @@ -48,6 +95,308 @@ func (w *Workflow) ownedStreams() stream.OwnedStreams { return stream.OwnedStreams{Streams: &w.Streams, Kind: "workflow"} } +// StreamAdmissionFailure turns a refusal of a stream command into a workflow +// task failure with the given cause. +// +// A refusal returned as a plain error fails the RespondWorkflowTaskCompleted +// call instead, and the worker then retries the same commands against the same +// limits until the task times out, with nothing in History saying why. Only +// refusals are converted: an internal or storage error is still the request's +// to report, because a retry can succeed. +func StreamAdmissionFailure(cause enumspb.WorkflowTaskFailedCause, err error) error { + var invalid *serviceerror.InvalidArgument + var precondition *serviceerror.FailedPrecondition + var exhausted *serviceerror.ResourceExhausted + var notFound *serviceerror.NotFound + if errors.As(err, &invalid) || errors.As(err, &precondition) || + errors.As(err, &exhausted) || errors.As(err, ¬Found) { + return FailWorkflowTaskError{Cause: cause, Message: err.Error()} + } + return err +} + +// handleAppendStreamRecordsCommand appends to a stream the workflow owns. +// +// The stream is a co-located subcomponent, so the batch and the frontier land +// in the workflow task's own commit: no extra transition, no cross-execution +// write, and a task that fails takes the publish with it. +// +// The offsets are known here, unlike a subscription's, so the event is written +// here too rather than reserved and filled in later. +func handleAppendStreamRecordsCommand( + chasmCtx chasm.MutableContext, + wf *Workflow, + validator Validator, + command *commandpb.Command, + opts CommandHandlerOptions, + limits stream.Limits, +) error { + badAttributes := enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES + attrs := command.GetAppendStreamRecordsCommandAttributes() + if attrs == nil { + return FailWorkflowTaskError{ + Cause: badAttributes, Message: "AppendStreamRecordsCommandAttributes is not set", + } + } + if len(attrs.GetRecords()) == 0 { + return FailWorkflowTaskError{ + Cause: badAttributes, Message: "AppendStreamRecords command carries no records", + } + } + + // The batch becomes one data node, so the whole batch is what has to fit. + // Left unchecked it fails later in the flush, which surfaces as a + // persistence error out of a task the worker will replay and re-issue + // forever, with nothing naming the batch as the cause. + size := 0 + for _, m := range attrs.GetRecords() { + size += m.Size() + } + if !validator.IsValidPayloadSize(size) { + return FailWorkflowTaskError{ + Cause: enumspb.WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE, + Message: "AppendStreamRecordsCommandAttributes.Records exceeds size limit", + TerminateWorkflow: true, + } + } + + name := attrs.GetStreamName() + if name == "" { + name = DefaultStreamName + } + + s, err := wf.streamNamed(chasmCtx, name, limits) + if err != nil { + return StreamAdmissionFailure(badAttributes, err) + } + + result, err := s.AddMessages(chasmCtx, stream.AddMessagesRequest{ + Records: toLibraryRecords(attrs.GetRecords()), + Limits: limits, + // Every stream this execution owns shares one byte budget, because + // their per-stream budgets multiplied out come to more than the + // execution size limit, which terminates the workflow rather than + // refusing an append. + SiblingBytes: wf.siblingStreamBytes(chasmCtx, name), + }) + if err != nil { + return StreamAdmissionFailure(badAttributes, err) + } + wf.RecordStreamRecordsAppended( + name, result.FirstOffset, result.NextOffset, opts.WorkflowTaskCompletedEventID) + return nil +} + +// handleSubscribeStreamCommand registers this workflow as a consumer. +// +// Nothing is registered here. A stream in another execution has to be pinned +// on its own shard, which a command handler cannot reach while holding the +// state lock, so the subscription is staged and resolved after the commands and +// before the commit. +func handleSubscribeStreamCommand( + chasmCtx chasm.MutableContext, + wf *Workflow, + _ Validator, + command *commandpb.Command, + opts CommandHandlerOptions, + limits stream.Limits, +) error { + badAttributes := enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES + attrs := command.GetSubscribeStreamCommandAttributes() + if attrs == nil { + return FailWorkflowTaskError{ + Cause: badAttributes, Message: "SubscribeStreamCommandAttributes is not set", + } + } + nameOrID := attrs.GetStreamNameOrId() + if nameOrID == "" { + return FailWorkflowTaskError{ + Cause: badAttributes, Message: "SubscribeStream command names no stream", + } + } + + // A second subscribe to the same stream registers nothing, but it still + // gets an event. Every SDK matches issued commands against + // command-generated events in order, so a command that produces none puts + // that matching out of step, which is the whole reason this event exists. + _, already := wf.StreamCursors[nameOrID] + + // Each new subscription costs a routed call on the completion path, made + // with this execution's lock held, and each delivery costs another on + // every task start. Bounded here, because nothing else bounds how many a + // workflow may hold or how many one task may carry. + if subscribed, known := wf.subscribedStreams(nameOrID); !known && + subscribed >= limits.MaxSubscriptionsPerWorkflow { + return FailWorkflowTaskError{ + Cause: badAttributes, + Message: fmt.Sprintf("workflow already subscribes to %d streams, the limit", + limits.MaxSubscriptionsPerWorkflow), + } + } + + start, err := stream.RequestedStart( + attrs.GetStartPosition(), "start_offset", attrs.GetStartOffset()) + if err != nil { + return FailWorkflowTaskError{Cause: badAttributes, Message: err.Error()} + } + + // Everything is staged, including a stream this workflow owns, so that the + // resolved start offset and the event recording it are produced in one + // place rather than two. + wf.StagePendingSubscription(PendingStreamSubscription{ + StreamID: nameOrID, + Start: start, + AlreadySubscribed: already, + Event: wf.ReserveStreamSubscribedEvent( + nameOrID, opts.WorkflowTaskCompletedEventID), + }) + return nil +} + +// streamSubscribedEvent is the event a subscription writes. +// +// It is recorded once per subscription, not per record: the offsets a task +// consumed ride WorkflowTaskCompleted, and payloads never enter History. The +// event exists because a command that produces none desynchronises the +// command-to-event matching every SDK's replay depends on, and because without +// it nothing in History explains why a workflow started receiving stream data. +type streamSubscribedEvent struct{} + +func (streamSubscribedEvent) Type() enumspb.EventType { + return enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED +} + +func (streamSubscribedEvent) IsWorkflowTaskTrigger() bool { return false } + +// Apply recreates the cursor for a run rebuilt from its history, which is how a +// reset run comes to exist. The event records the stream and where reading +// began; the ranges consumed since are folded in as the completed events that +// carry them are applied. Whether the stream lives in this execution or in +// another is not in the event, so the rebuilt cursor is owned until the reset +// copies that from the run it was rebuilt from. +func (streamSubscribedEvent) Apply( + mctx chasm.MutableContext, + wf *Workflow, + event *historypb.HistoryEvent, +) error { + attrs := event.GetWorkflowStreamSubscribedEventAttributes() + if _, ok := wf.StreamCursors[attrs.GetStreamId()]; ok { + return nil + } + cursor, err := stream.NewCursor(mctx, stream.NewCursorRequest{ + StreamID: attrs.GetStreamId(), + StartOffset: attrs.GetStartOffset(), + }) + if err != nil { + return err + } + if wf.StreamCursors == nil { + wf.StreamCursors = make(chasm.Map[string, *stream.Cursor]) + } + wf.StreamCursors[attrs.GetStreamId()] = chasm.NewComponentField(mctx, cursor) + return nil +} + +// A command event, so it is never cherry-picked: the workflow reissues the +// subscribe command on the new branch if it still wants one. +func (streamSubscribedEvent) CherryPick( + chasm.MutableContext, + *Workflow, + *historypb.HistoryEvent, + map[enumspb.ResetReapplyExcludeType]struct{}, +) error { + return ErrEventNotCherryPickable +} + +// ReserveStreamSubscribedEvent writes the event a subscribe command owes, +// leaving the start offset for the flush to fill in. +// +// Written here rather than where the offset becomes known, because an SDK +// matches the commands a workflow issued against the events they produced by +// position. The flush runs after every command, so an event written there +// would sit behind the events of commands that were issued later, and the +// first replay of a workflow that subscribed before doing anything else would +// fail on the mismatch. +func (w *Workflow) ReserveStreamSubscribedEvent( + streamID string, + workflowTaskCompletedEventID int64, +) *historypb.HistoryEvent { + eventType := enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED + return w.AddHistoryEvent(eventType, func(e *historypb.HistoryEvent) { + attrs := &historypb.WorkflowStreamSubscribedEventAttributes{ + WorkflowTaskCompletedEventId: workflowTaskCompletedEventID, + StreamId: streamID, + } + e.Attributes = &historypb.HistoryEvent_WorkflowStreamSubscribedEventAttributes{ + WorkflowStreamSubscribedEventAttributes: attrs, + } + }) +} + +// RecordStreamSubscribedOffset completes a reserved event. Safe to do after the +// fact because the builder serializes the batch at commit, which is after the +// flush that resolves the offset. +func RecordStreamSubscribedOffset(event *historypb.HistoryEvent, startOffset int64) { + event.GetWorkflowStreamSubscribedEventAttributes().StartOffset = startOffset +} + +// streamRecordsAppendedEvent is the event a publish writes. +// +// One per batch, holding the offset range and nothing else. That is what makes +// it a fixed cost: a batch of one 20-byte record and a batch of a thousand +// 2KB records write the same event, because the bodies stay in the stream +// component. It exists for the same reason the subscription event does, that a +// command producing no event desynchronises the command-to-event matching +// every SDK's replay depends on, and it doubles as the only record in History +// that the workflow published at all. +type streamRecordsAppendedEvent struct{} + +func (streamRecordsAppendedEvent) Type() enumspb.EventType { + return enumspb.EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED +} + +func (streamRecordsAppendedEvent) IsWorkflowTaskTrigger() bool { return false } + +// The frontier it describes is CHASM state, persisted and rebuilt with the +// execution, so there is nothing here to reconstruct. +func (streamRecordsAppendedEvent) Apply( + chasm.MutableContext, *Workflow, *historypb.HistoryEvent, +) error { + return nil +} + +// A command event, so it is never cherry-picked: the offsets belong to a log +// the new branch did not write. +func (streamRecordsAppendedEvent) CherryPick( + chasm.MutableContext, + *Workflow, + *historypb.HistoryEvent, + map[enumspb.ResetReapplyExcludeType]struct{}, +) error { + return ErrEventNotCherryPickable +} + +// RecordStreamRecordsAppended writes the event for one published batch. +func (w *Workflow) RecordStreamRecordsAppended( + streamID string, + fromOffset int64, + toOffset int64, + workflowTaskCompletedEventID int64, +) { + eventType := enumspb.EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED + w.AddHistoryEvent(eventType, func(e *historypb.HistoryEvent) { + attrs := &historypb.WorkflowStreamRecordsAppendedEventAttributes{ + WorkflowTaskCompletedEventId: workflowTaskCompletedEventID, + StreamId: streamID, + FromOffset: fromOffset, + ToOffset: toOffset, + } + e.Attributes = &historypb.HistoryEvent_WorkflowStreamRecordsAppendedEventAttributes{ + WorkflowStreamRecordsAppendedEventAttributes: attrs, + } + }) +} + // streamNamed returns the workflow's stream of that name, creating it on first // use. Implicit creation is deliberate: a workflow publishing to its own output // should not have to coordinate with anyone about who creates it. @@ -62,6 +411,115 @@ func (w *Workflow) streamNamed( return w.ownedStreams().Named(ctx, name, limits) } +// toLibraryRecords shapes a command's records for the store. +// +// The producer id is cleared rather than copied: an empty id is how a reader +// tells the owning workflow's records from an outside producer's, and only this +// path writes on the workflow's behalf. The kind is left as sent, since the +// store settles an unspecified one on the copy it serializes. +func toLibraryRecords(in []*streampb.StreamRecord) []*streamlib.StreamRecord { + out := make([]*streamlib.StreamRecord, len(in)) + for i, m := range in { + out[i] = &streamlib.StreamRecord{ + Body: m.GetBody(), + Metadata: m.GetMetadata(), + Topic: m.GetTopic(), + Kind: m.GetKind(), + ProducerId: "", + Attempt: m.GetAttempt(), + Sequence: m.GetSequence(), + } + } + return out +} + +// streamLibrary registers the stream commands with the workflow registry. +type streamLibrary struct { + config *stream.Config +} + +func newStreamLibrary(config *stream.Config) *streamLibrary { + return &streamLibrary{config: config} +} + +func (l *streamLibrary) CommandHandlers() map[enumspb.CommandType]CommandHandler { + return map[enumspb.CommandType]CommandHandler{ + enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS: func( + chasmCtx chasm.MutableContext, + wf *Workflow, + validator Validator, + command *commandpb.Command, + opts CommandHandlerOptions, + ) error { + namespaceName := chasmCtx.NamespaceEntry().Name().String() + if err := l.checkEnabled( + namespaceName, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES, + ); err != nil { + return err + } + limits := l.config.LimitsFor(namespaceName) + return handleAppendStreamRecordsCommand(chasmCtx, wf, validator, command, opts, limits) + }, + enumspb.COMMAND_TYPE_SUBSCRIBE_STREAM: func( + chasmCtx chasm.MutableContext, + wf *Workflow, + validator Validator, + command *commandpb.Command, + opts CommandHandlerOptions, + ) error { + namespaceName := chasmCtx.NamespaceEntry().Name().String() + if err := l.checkEnabled( + namespaceName, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, + ); err != nil { + return err + } + limits := l.config.LimitsFor(namespaceName) + return handleSubscribeStreamCommand(chasmCtx, wf, validator, command, opts, limits) + }, + } +} + +// checkEnabled fails the workflow task when streams are off for the namespace. +// +// The command path does not go through the stream service, so the frontend +// gate does not cover it. Failed with a cause rather than returned raw, so the +// worker sees why instead of reissuing the same command until the task times +// out. +func (l *streamLibrary) checkEnabled( + namespaceName string, + cause enumspb.WorkflowTaskFailedCause, +) error { + if l.config.EnabledFor(namespaceName) { + return nil + } + return FailWorkflowTaskError{ + Cause: cause, + Message: "streams are not enabled for namespace: " + namespaceName, + } +} + +func (l *streamLibrary) EventDefinitions() []EventDefinition { + return []EventDefinition{streamSubscribedEvent{}, streamRecordsAppendedEvent{}} +} + +// subscribedStreams counts the distinct streams this workflow consumes or is +// about to, and says whether the given one is among them. Staged subscriptions +// count: they are resolved before the commit, so a single task can otherwise +// stage as many as it likes. +func (w *Workflow) subscribedStreams(streamID string) (int, bool) { + seen := make(map[string]struct{}, len(w.StreamCursors)+len(w.pendingStreamSubscriptions)) + for name := range w.StreamCursors { + seen[name] = struct{}{} + } + for _, pending := range w.pendingStreamSubscriptions { + seen[pending.StreamID] = struct{}{} + } + _, known := seen[streamID] + return len(seen), known +} + // siblingStreamBytes is what every other stream this workflow owns holds, // streams its activities own included, since they live in the same state. func (w *Workflow) siblingStreamBytes(ctx chasm.Context, name string) int64 { diff --git a/chasm/lib/workflow/stream_commands_test.go b/chasm/lib/workflow/stream_commands_test.go new file mode 100644 index 00000000000..66b65df9096 --- /dev/null +++ b/chasm/lib/workflow/stream_commands_test.go @@ -0,0 +1,41 @@ +package workflow + +import ( + "testing" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + streampb "go.temporal.io/api/stream/v1" +) + +// An empty producer id is how a reader tells the owning workflow's records from +// an outside producer's. The command path is the only writer on the workflow's +// behalf, so it decides the field rather than trusting what the worker sent, +// while the attempt and sequence stay the worker's to number. +func TestPublishedRecordsBelongToTheWorkflow(t *testing.T) { + got := toLibraryRecords([]*streampb.StreamRecord{ + { + Body: &commonpb.Payload{Data: []byte("token")}, + Metadata: map[string]*commonpb.Payload{"k": {Data: []byte("v")}}, + Topic: "tokens", + Kind: streampb.STREAM_RECORD_KIND_FINISH, + ProducerId: "impostor", + Attempt: 3, + Sequence: 9, + }, + {Body: &commonpb.Payload{Data: []byte("kindless")}}, + }) + require.Len(t, got, 2) + + require.Equal(t, "token", string(got[0].GetBody().GetData())) + require.Equal(t, "v", string(got[0].GetMetadata()["k"].GetData())) + require.Equal(t, "tokens", got[0].GetTopic()) + require.Equal(t, streampb.STREAM_RECORD_KIND_FINISH, got[0].GetKind()) + require.Empty(t, got[0].GetProducerId(), "the workflow is the producer here") + require.Equal(t, int64(3), got[0].GetAttempt()) + require.Equal(t, int64(9), got[0].GetSequence()) + + require.Equal(t, streampb.STREAM_RECORD_KIND_UNSPECIFIED, got[1].GetKind(), + "the kind is left as sent; the store settles it on the copy it serializes") + require.Empty(t, got[1].GetProducerId()) +} diff --git a/chasm/lib/workflow/stream_cursor_test.go b/chasm/lib/workflow/stream_cursor_test.go new file mode 100644 index 00000000000..ef3dea670ad --- /dev/null +++ b/chasm/lib/workflow/stream_cursor_test.go @@ -0,0 +1,220 @@ +package workflow + +import ( + "testing" + + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + streampb "go.temporal.io/api/stream/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +func newStreamCursorTestContext() chasm.MutableContext { + return &chasm.MockMutableContext{ + MockContext: chasm.MockContext{ + HandleExecutionKey: func() chasm.ExecutionKey { + return chasm.ExecutionKey{ + NamespaceID: "ns-1", + BusinessID: "wf-1", + RunID: "run-1", + } + }, + }, + } +} + +// Built directly rather than through NewStream, which wires a visibility field +// needing a live context. +func newAttachedStream(t *testing.T, ctx chasm.MutableContext, count int) *stream.Stream { + t.Helper() + + s := &stream.Stream{ + State: &streamlib.StreamState{ + Producers: make(map[string]*streamlib.ProducerCursor), + Consumers: make(map[string]*streamlib.ConsumerCursor), + }, + } + + messages := make([]*streamlib.StreamRecord, count) + for i := range messages { + messages[i] = &streamlib.StreamRecord{Kind: streampb.STREAM_RECORD_KIND_DATA} + } + _, err := s.AddMessages(ctx, stream.AddMessagesRequest{Records: messages}) + require.NoError(t, err) + + return s +} + +// Subscribing registers the consumer on the stream, which decides both whether +// an append is worth waking it for and what retention may not take. A range +// this workflow consumes goes into its History, and a replay is asked to +// reproduce it, so the stream holds those messages while the subscription is +// active and releases them when it is not. +func TestSubscribeRegistersTheConsumer(t *testing.T) { + ctx := newStreamCursorTestContext() + w := &Workflow{} + owned := newAttachedStream(t, ctx, 4) + w.Streams = chasm.Map[string, *stream.Stream]{ + DefaultStreamName: chasm.NewComponentField(ctx, owned), + } + + start, err := w.SubscribeToOwnedStream( + ctx, DefaultStreamName, stream.AtOffset(0), stream.DefaultLimits()) + require.NoError(t, err) + require.Equal(t, int64(0), start) + + consumer := owned.State.GetConsumers()[streamConsumerID(DefaultStreamName)] + require.NotNil(t, consumer, "the stream has to know who is reading it") + require.Equal(t, int64(0), consumer.GetOffset()) + require.True(t, consumer.GetActive()) + + require.Equal(t, int64(0), consumer.GetReplayFloor()) + + // And it holds the floor for as long as it is subscribed. + require.ErrorContains(t, owned.Truncate(ctx, 1), "still depends on offset 0") +} + +func TestSubscribeFromTheTailResolvesToHead(t *testing.T) { + ctx := newStreamCursorTestContext() + w := &Workflow{} + owned := newAttachedStream(t, ctx, 4) + w.Streams = chasm.Map[string, *stream.Stream]{ + DefaultStreamName: chasm.NewComponentField(ctx, owned), + } + + start, err := w.SubscribeToOwnedStream( + ctx, DefaultStreamName, stream.Tail(), stream.DefaultLimits()) + require.NoError(t, err) + require.Equal(t, int64(4), start, "tail means from wherever the stream is now") +} + +// A workflow reads a topic by name before anything has been written to it, so +// subscribing has to bring the stream into being the way a first write does. +func TestSubscribeCreatesTheStreamItNames(t *testing.T) { + ctx := newStreamCursorTestContext() + w := &Workflow{} + + start, err := w.SubscribeToOwnedStream(ctx, "inputs", stream.AtOffset(0), stream.DefaultLimits()) + require.NoError(t, err) + require.Equal(t, int64(0), start) + + field, ok := w.Streams["inputs"] + require.True(t, ok, "the subscribed name is now an owned stream") + state, err := field.Get(ctx).Snapshot(ctx, struct{}{}) + require.NoError(t, err) + require.Equal(t, int64(0), state.GetHeadOffset()) + require.Len(t, state.GetConsumers(), 1, "the subscription pinned the new stream") +} + +// Two publishes in one workflow task each write their own batch, keyed by the +// offset it starts at, so neither collides with the other and a retry of +// either addresses the same key it wrote before. +func TestPublishStagesEachBatchAtItsOwnOffset(t *testing.T) { + ctx := newStreamCursorTestContext() + + // A backend, because the handler writes the publish event and that event is + // part of what the command owes. + backend := &chasm.MockNodeBackend{ + HandleAddHistoryEvent: func( + t enumspb.EventType, set func(*historypb.HistoryEvent), + ) *historypb.HistoryEvent { + e := &historypb.HistoryEvent{EventType: t} + set(e) + return e + }, + } + w := &Workflow{MSPointer: chasm.NewMSPointer(backend)} + opts := CommandHandlerOptions{WorkflowTaskCompletedEventID: 10} + + publish := &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: []*streampb.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("x")}}, + {Body: &commonpb.Payload{Data: []byte("y")}}, + }, + }, + }, + } + + limits := stream.DefaultLimits() + require.NoError(t, handleAppendStreamRecordsCommand(ctx, w, allowAnySize{}, publish, opts, limits)) + require.NoError(t, handleAppendStreamRecordsCommand(ctx, w, allowAnySize{}, publish, opts, limits)) + + // Both publishes committed with the workflow task, so the batches are on + // the component keyed by the offsets they start at. + owned := w.Streams[DefaultStreamName].Get(ctx) + require.Len(t, owned.Batches, 2) + _, ok := owned.Batches[0] + require.True(t, ok) + _, ok = owned.Batches[2] + require.True(t, ok) + require.Equal(t, int64(4), owned.State.GetHeadOffset()) +} + +type allowAnySize struct{} + +func (allowAnySize) IsValidPayloadSize(int) bool { return true } + +// A consumer that falls behind a truncating stream must be told, not handed +// what is left with a hole in it. +// +// Being outrun is possible again once the subscription is released, which is +// deliberate: a floor that nothing ever gave up would keep every message for +// as long as a consumer had ever existed. The trade is that being outrun has +// to be loud. +func TestConsumerOutrunByTruncationIsToldSo(t *testing.T) { + ctx := newStreamCursorTestContext() + w := &Workflow{} + owned := newAttachedStream(t, ctx, 4) + w.Streams = chasm.Map[string, *stream.Stream]{ + DefaultStreamName: chasm.NewComponentField(ctx, owned), + } + + _, err := w.SubscribeToOwnedStream( + ctx, DefaultStreamName, stream.AtOffset(0), stream.DefaultLimits()) + require.NoError(t, err) + + // Someone decides these messages are no longer needed, and only then can + // the stream move past where this consumer is sitting. + owned.DeregisterConsumer(ctx, streamConsumerID(DefaultStreamName)) + err = owned.Truncate(ctx, 3) + require.NoError(t, err, "a released floor must not keep holding") + + cursor := w.StreamCursors[DefaultStreamName].Get(ctx) + require.Less(t, cursor.Offset(), owned.State.GetBaseOffset()) + + state, err := owned.Snapshot(ctx, struct{}{}) + require.NoError(t, err) + require.Equal(t, int64(3), state.GetBaseOffset(), + "the floor moved, which is what the consumer has to find out about") +} + +// A publish command naming the part of the map activities use fails the task +// with a cause, instead of writing into a stream an activity owns. +func TestPublishCommandCannotReachAnActivityStream(t *testing.T) { + ctx := newStreamBudgetTestContext() + w := &Workflow{} + publish := &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + StreamName: ActivityStreamKey("act", DefaultStreamName), + Records: []*streampb.StreamRecord{{Body: &commonpb.Payload{Data: []byte("x")}}}, + }, + }, + } + err := handleAppendStreamRecordsCommand(ctx, w, allowAnySize{}, publish, + CommandHandlerOptions{}, stream.DefaultLimits()) + var failTask FailWorkflowTaskError + require.ErrorAs(t, err, &failTask) + require.Equal(t, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES, failTask.Cause) + require.Empty(t, w.Streams) +} diff --git a/chasm/lib/workflow/workflow.go b/chasm/lib/workflow/workflow.go index 747148698a0..439e9297769 100644 --- a/chasm/lib/workflow/workflow.go +++ b/chasm/lib/workflow/workflow.go @@ -2,6 +2,8 @@ package workflow import ( "fmt" + "slices" + "strings" commonpb "go.temporal.io/api/common/v1" failurepb "go.temporal.io/api/failure/v1" @@ -55,6 +57,40 @@ type Workflow struct { // would collide with the other origin under the same key is refused rather // than silently handed the wrong cursor. StreamCursors chasm.Map[string, *stream.Cursor] + + // Subscribe commands whose stream is in another execution, so the addressing + // has to be looked up before a cursor can be made. In memory only, drained + // by the flush before commit. + pendingStreamSubscriptions []PendingStreamSubscription +} + +// PendingStreamSubscription is a subscribe command whose stream lives in +// another execution, waiting for the flush to look up its addressing. +type PendingStreamSubscription struct { + StreamID string + // Where the command asked to start, already checked. The flush resolves it + // against the stream. + Start *streampb.StreamStartPosition + // The workflow already holds a cursor for this stream. The subscription + // itself is done, but the command still needs its event, because that is + // what a replaying worker matches the re-issued command against. + AlreadySubscribed bool + + // The event this command already wrote, waiting on its start offset. In + // memory only, like the rest of this struct. + Event *historypb.HistoryEvent +} + +// StagePendingSubscription records a subscription for the flush to resolve. +func (w *Workflow) StagePendingSubscription(sub PendingStreamSubscription) { + w.pendingStreamSubscriptions = append(w.pendingStreamSubscriptions, sub) +} + +// DrainStreamSubscriptions returns and clears the staged subscriptions. +func (w *Workflow) DrainStreamSubscriptions() []PendingStreamSubscription { + out := w.pendingStreamSubscriptions + w.pendingStreamSubscriptions = nil + return out } // streamConsumerID names this workflow's pin on a stream it owns. An attached @@ -171,6 +207,58 @@ func (w *Workflow) SubscribeToExternalStream( return req.StartOffset, nil } +// ExportStreamSubscriptions returns the subscriptions a successor run has to +// inherit. +// +// Only subscriptions to streams in other executions are exported. A stream this +// workflow owns lives in this execution and does not survive the run +// transition, so carrying a cursor for one would leave the successor pointing +// at a stream it cannot reach. +// +// So a subscription to a stream the workflow owns ends at a continue-as-new. +// The successor's stream of the same name is a new one starting at offset +// zero, an outside producer addressing the workflow with no run id reaches +// that one, and records written to the predecessor's stream between the +// commit and the successor's re-subscribe are not carried. A workflow that +// wants its own output to continue across runs has to republish; a stream in +// another execution, which is what the offsets of a long-lived reader belong +// to, does carry. +func (w *Workflow) ExportStreamSubscriptions(ctx chasm.Context) []ExternalStreamSubscription { + var out []ExternalStreamSubscription + for _, field := range w.StreamCursors { + cursor := field.Get(ctx) + if !cursor.IsExternal() { + continue + } + out = append(out, ExternalStreamSubscription{ + StreamID: cursor.StreamID(), + StartOffset: cursor.Offset(), + KnownHead: cursor.KnownHead(), + }) + } + // Stable order, so a successor's state does not depend on map iteration. + slices.SortFunc(out, func(a, b ExternalStreamSubscription) int { + return strings.Compare(a.StreamID, b.StreamID) + }) + return out +} + +// ImportStreamSubscriptions installs subscriptions inherited from the run this +// one continues. The offset carries over unchanged: offsets are global to the +// stream, so the successor resumes exactly where its predecessor stopped and +// the stream itself is untouched. +func (w *Workflow) ImportStreamSubscriptions( + mctx chasm.MutableContext, + subscriptions []ExternalStreamSubscription, +) error { + for _, sub := range subscriptions { + if _, err := w.SubscribeToExternalStream(mctx, sub); err != nil { + return err + } + } + return nil +} + // AdvanceKnownHead records how far a stream in another execution has moved. // // A workflow cannot read that frontier itself while closing its own diff --git a/service/history/api/command_attr_validator.go b/service/history/api/command_attr_validator.go index 8c48b0f99f1..257135405d9 100644 --- a/service/history/api/command_attr_validator.go +++ b/service/history/api/command_attr_validator.go @@ -661,7 +661,13 @@ func (v *CommandAttrValidator) ValidateCommandSequence( enumspb.COMMAND_TYPE_MODIFY_WORKFLOW_PROPERTIES, enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE, enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, - enumspb.COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION: + enumspb.COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION, + // Publishing to a stream the workflow owns. Not a close command: + // it appends and returns, scheduling nothing further. + enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + // Subscribing to a stream. Also not closing: it records a cursor + // and the workflow carries on. + enumspb.COMMAND_TYPE_SUBSCRIBE_STREAM: // noop case enumspb.COMMAND_TYPE_CONTINUE_AS_NEW_WORKFLOW_EXECUTION, enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, diff --git a/service/history/api/recordworkflowtaskstarted/stream_routing.go b/service/history/api/recordworkflowtaskstarted/stream_routing.go new file mode 100644 index 00000000000..64a3447462b --- /dev/null +++ b/service/history/api/recordworkflowtaskstarted/stream_routing.go @@ -0,0 +1,42 @@ +package recordworkflowtaskstarted + +import ( + "context" + + "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +type streamClientContextKey struct{} + +// WithStreamClient supplies shard routing for calls that reach a stream in +// another execution from inside a workflow task's transaction. +func WithStreamClient(ctx context.Context, client streamlib.StreamServiceClient) context.Context { + if client == nil { + return ctx + } + return context.WithValue(ctx, streamClientContextKey{}, client) +} + +// StreamClientFromContext returns the routed client installed by the engine. +func StreamClientFromContext(ctx context.Context) (streamlib.StreamServiceClient, bool) { + client, ok := ctx.Value(streamClientContextKey{}).(streamlib.StreamServiceClient) + return client, ok +} + +// WithRoutedDeadline bounds a routed call made while the workflow lock is held. +func WithRoutedDeadline(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(ctx, stream.RoutedCallTimeout) +} + +// WithRoutedBudget bounds a whole set of routed calls made under one workflow +// lock. +// +// Each call has a deadline of its own, but a task can carry as many of them as +// the subscription limit allows, and per-call deadlines multiplied by that +// count is how long the lock could be held. One budget over the set is what +// actually bounds it: a call that runs out of it fails, and the workflow task +// fails with it rather than the lock being held for minutes. +func WithRoutedBudget(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(ctx, stream.RoutedSetBudget) +} diff --git a/service/history/api/respondworkflowtaskcompleted/api.go b/service/history/api/respondworkflowtaskcompleted/api.go index d9b91c6e39f..b0db29dcf2a 100644 --- a/service/history/api/respondworkflowtaskcompleted/api.go +++ b/service/history/api/respondworkflowtaskcompleted/api.go @@ -453,6 +453,34 @@ func (handler *WorkflowTaskCompletedHandler) Invoke( return nil, err } + // Subscriptions to streams in other executions, resolved here for the + // same reason: the command handler has nowhere to look the addressing + // up from, and by delivery time the cursor has to already exist. + // + // Skipped once the task has failed. Registering a consumer is a durable + // write on the stream's own execution, and this workflow's side of it + // is about to be rolled back, which would leave a pin on someone else's + // stream with no cursor behind it and nothing to release it. A failed + // command does not return an error, so this has to be checked here + // rather than inferred from err. + if workflowTaskHandler.workflowTaskFailedCause == nil && !workflowTaskHandler.stopProcessing { + err = resolveStagedStreamSubscriptions( + ctx, + ms, + ms.GetWorkflowKey().NamespaceID, + handler.config.Stream.LimitsFor(namespaceEntry.Name().String()), + workflowTaskHandler.stagedStreamSubscriptions, + ) + if failWFTErr, ok := errors.AsType[chasmworkflow.FailWorkflowTaskError](err); ok { + // A refused subscription fails the task the way a refused + // command does, so the cause lands in History. + err = workflowTaskHandler.failWorkflowTask(failWFTErr.Cause, failWFTErr) + } + if err != nil { + return nil, err + } + } + // Worker must respond with Update Accepted or Update Rejected message on every Update Requested // message that were delivered on specific WT, when completing this WT. // If worker ignored the update request (old SDK or SDK bug), then server rejects this update. diff --git a/service/history/api/respondworkflowtaskcompleted/stream_appends.go b/service/history/api/respondworkflowtaskcompleted/stream_appends.go new file mode 100644 index 00000000000..d9c9c6a8733 --- /dev/null +++ b/service/history/api/respondworkflowtaskcompleted/stream_appends.go @@ -0,0 +1,168 @@ +package respondworkflowtaskcompleted + +import ( + "context" + + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" + "go.temporal.io/server/service/history/api/recordworkflowtaskstarted" + historyi "go.temporal.io/server/service/history/interfaces" +) + +// resolveStagedStreamSubscriptions turns subscribe commands for streams in +// other executions into cursors on this workflow. +// +// The lookup cannot happen in the command handler, which runs under the state +// lock with nowhere to do I/O from, and it cannot happen at delivery either, +// because by then the cursor has to already exist. So it happens here, between +// the commands and the commit. +// +// The pin goes on the stream before the cursor goes on the workflow, and that +// order is the guarantee. Interrupted after the first write there is a pin +// holding storage nothing reads, which costs space. The other order would leave +// a cursor no truncation floor protects, and truncation would be free to take a +// range it still points at. +// +// The pin is taken through the routed service client. The engine on this +// request context resolves shards through the local controller, which refuses +// any shard this host does not own, so a stream living elsewhere in the +// cluster can only be reached by going back out through the service. +// +// A refusal, such as an offset below the stream's floor, comes back as a +// workflow task failure so the worker sees a cause rather than retrying the +// same command. A pin already taken for an earlier subscription in the same +// task stays on its stream; the notify task releases it when it finds this +// workflow does not consume that stream. +func resolveStagedStreamSubscriptions( + ctx context.Context, + ms historyi.MutableState, + namespaceID string, + limits stream.Limits, + staged []chasmworkflow.PendingStreamSubscription, +) error { + if len(staged) == 0 { + return nil + } + + wf, chasmCtx, err := ms.ChasmWorkflowComponent(ctx) + if err != nil { + return err + } + + // One budget over the whole set. Each pin is a routed call with a deadline + // of its own, nothing bounds how many a task carries beyond the + // subscription limit, and the execution's lock is held for all of them. + ctx, cancelBudget := recordworkflowtaskstarted.WithRoutedBudget(ctx) + defer cancelBudget() + + for _, pending := range staged { + // Already subscribed, so only the event is owed. Re-registering would + // re-run the pin write with the original start offset, which would drag + // the stream's truncation floor back to where this consumer began. + if pending.AlreadySubscribed { + cursor, ok := wf.StreamCursors[pending.StreamID] + if !ok { + return serviceerror.NewInternalf( + "stream %q was marked already subscribed but has no cursor", pending.StreamID) + } + chasmworkflow.RecordStreamSubscribedOffset( + pending.Event, cursor.Get(chasmCtx).Offset()) + continue + } + + // A stream this workflow owns needs no lookup and no pin registration: + // it is in this execution, and its cursor commits with everything else. + if _, owned := wf.Streams[pending.StreamID]; owned { + startOffset, err := wf.SubscribeToOwnedStream( + chasmCtx, pending.StreamID, pending.Start, limits) + if err != nil { + return chasmworkflow.StreamAdmissionFailure( + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, err) + } + chasmworkflow.RecordStreamSubscribedOffset(pending.Event, startOffset) + continue + } + + pin, err := registerExternalConsumer(ctx, ms, namespaceID, pending) + if err != nil { + return chasmworkflow.StreamAdmissionFailure( + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, err) + } + if pin.GetStreamAbsent() { + // The stream's own shard says no execution holds a stream by this + // id, so the name is one of this workflow's own that nothing has + // written to yet. A reader has to be able to subscribe before the + // first record arrives, so the stream is created here and the + // subscription lands on it. + // + // This is an answer and not a NotFound on purpose. A NotFound can + // also come from a registry miss or a shard that has moved, and + // treating one of those as "create a stream of your own" would bind + // the workflow to different, empty data with a History event that + // looks exactly like the intended subscription. + startOffset, err := wf.SubscribeToOwnedStream( + chasmCtx, pending.StreamID, pending.Start, limits) + if err != nil { + return chasmworkflow.StreamAdmissionFailure( + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, err) + } + chasmworkflow.RecordStreamSubscribedOffset(pending.Event, startOffset) + continue + } + + if _, err := wf.SubscribeToExternalStream(chasmCtx, chasmworkflow.ExternalStreamSubscription{ + StreamID: pending.StreamID, + StartOffset: pin.GetStartOffset(), + KnownHead: pin.GetKnownHead(), + }); err != nil { + // Failed like every other refusal in this loop: with a cause on the + // workflow task, rather than failing the completion call and + // leaving the worker to reissue the same command. + return chasmworkflow.StreamAdmissionFailure( + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, err) + } + + // Completed after the cursor exists. The event was written where the + // command was, but nothing outside this transaction sees either until + // the commit below, so a crash in between leaves no event claiming a + // subscription that was never made. + chasmworkflow.RecordStreamSubscribedOffset(pending.Event, pin.GetStartOffset()) + } + return nil +} + +// registerExternalConsumer takes the pin on the stream's own shard and returns +// the resolved start offset and the frontier as of registration. +func registerExternalConsumer( + ctx context.Context, + ms historyi.MutableState, + namespaceID string, + pending chasmworkflow.PendingStreamSubscription, +) (*streamlib.RegisterStreamConsumerOutput, error) { + client, ok := recordworkflowtaskstarted.StreamClientFromContext(ctx) + if !ok { + return nil, serviceerror.NewInternal( + "stream service client is not available on the workflow task completion path") + } + + key := ms.GetWorkflowKey() + callCtx, cancel := recordworkflowtaskstarted.WithRoutedDeadline(ctx) + defer cancel() + registered, err := client.RegisterStreamConsumer(callCtx, &streamlib.RegisterStreamConsumerRequest{ + NamespaceId: namespaceID, + FrontendRequest: &streamlib.RegisterStreamConsumerInput{ + Namespace: ms.GetNamespaceEntry().Name().String(), + StreamId: pending.StreamID, + ConsumerWorkflowId: key.WorkflowID, + ConsumerRunId: key.RunID, + StartPosition: pending.Start, + }, + }) + if err != nil { + return nil, err + } + return registered.GetFrontendResponse(), nil +} diff --git a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go index 4e8da517eae..8baf1728e6d 100644 --- a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go +++ b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go @@ -61,6 +61,9 @@ type ( workflowTaskDeployment *deploymentpb.Deployment // internal state + // Subscribe commands for streams in other executions, resolved before + // this workflow task commits. + stagedStreamSubscriptions []chasmworkflow.PendingStreamSubscription hasBufferedEventsOrMessages bool workflowTaskFailedCause *workflowTaskFailedCause activityNotStartedCancelled bool @@ -354,6 +357,12 @@ func (handler *workflowTaskCompletedHandler) handleCommand( return nil, chasmErr } err = chasmHandler(chasmCtx, chasmWorkflow, validator, command, handlerOpts) + // A subscribe to a stream in another execution stages itself, + // since a command handler holds the state lock and has no + // context for the lookup it needs. Collect them for the + // resolution that has to precede this task's commit. + handler.stagedStreamSubscriptions = append( + handler.stagedStreamSubscriptions, chasmWorkflow.DrainStreamSubscriptions()...) // Fall back to the HSM handler either when the command type is not supported by CHASM (disabled // feature flag) or when the targeted entity is not owned by the CHASM tree (e.g. an operation // scheduled in HSM before the flag was flipped on). diff --git a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go index 501cf4dc51e..d0293e1a2dc 100644 --- a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go +++ b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go @@ -81,6 +81,9 @@ func TestCommandProtocolMessage(t *testing.T) { out.ms.EXPECT().VisitUpdates(gomock.Any()).AnyTimes() out.ms.EXPECT().GetNamespaceEntry().Return(tests.LocalNamespaceEntry).AnyTimes() out.ms.EXPECT().GetCurrentVersion().Return(tests.LocalNamespaceEntry.FailoverVersion(tests.WorkflowID)).AnyTimes() + out.ms.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{ + WorkflowTaskAttempt: 1, + }).AnyTimes() dcClient := dynamicconfig.StaticClient(nil) if opts.chasmEnabled { diff --git a/service/history/configs/config.go b/service/history/configs/config.go index fffcaf2149d..6b1d704accc 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -3,6 +3,7 @@ package configs import ( "go.temporal.io/server/chasm/lib/callback" "go.temporal.io/server/chasm/lib/nexusoperation" + "go.temporal.io/server/chasm/lib/stream" "go.temporal.io/server/common" "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/health" @@ -256,6 +257,7 @@ type Config struct { HistoryCountLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter HistoryCountSuggestContinueAsNew dynamicconfig.IntPropertyFnWithNamespaceFilter HistoryMaxPageSize dynamicconfig.IntPropertyFnWithNamespaceFilter + Stream *stream.Config MutableStateActivityFailureSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter MutableStateActivityFailureSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter MutableStateSizeLimitError dynamicconfig.IntPropertyFn @@ -726,6 +728,7 @@ func NewConfig( HistoryCountLimitWarn: dynamicconfig.HistoryCountLimitWarn.Get(dc), HistoryCountSuggestContinueAsNew: dynamicconfig.HistoryCountSuggestContinueAsNew.Get(dc), HistoryMaxPageSize: dynamicconfig.HistoryMaxPageSize.Get(dc), + Stream: stream.NewConfig(dc), MutableStateActivityFailureSizeLimitError: dynamicconfig.MutableStateActivityFailureSizeLimitError.Get(dc), MutableStateActivityFailureSizeLimitWarn: dynamicconfig.MutableStateActivityFailureSizeLimitWarn.Get(dc), MutableStateSizeLimitError: dynamicconfig.MutableStateSizeLimitError.Get(dc), diff --git a/service/history/history_engine.go b/service/history/history_engine.go index f1d4774efa7..de5fb6ae0a6 100644 --- a/service/history/history_engine.go +++ b/service/history/history_engine.go @@ -14,6 +14,7 @@ import ( replicationspb "go.temporal.io/server/api/replication/v1" workflowspb "go.temporal.io/server/api/workflow/v1" "go.temporal.io/server/chasm" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/client" "go.temporal.io/server/common" @@ -106,6 +107,7 @@ import ( type ( engineOptions struct { workflowResendScheduler workflowresend.Scheduler + streamClient streampb.StreamServiceClient } // EngineOption adds optional history engine dependencies without adding required parameters. @@ -147,6 +149,7 @@ type ( workflowConsistencyChecker api.WorkflowConsistencyChecker workflowResendScheduler workflowresend.Scheduler chasmEngine chasm.Engine + streamClient streampb.StreamServiceClient versionChecker headers.VersionChecker versionCache worker_versioning.VersionMembershipAndReactivationStatusCache workerDeploymentClient workerdeployment.Client @@ -170,6 +173,13 @@ func WithWorkflowResendScheduler(scheduler workflowresend.Scheduler) EngineOptio } } +// WithStreamClient routes standalone stream reads to their history shard owner. +func WithStreamClient(streamClient streampb.StreamServiceClient) EngineOption { + return func(options *engineOptions) { + options.streamClient = streamClient + } +} + func applyEngineOptions(options []EngineOption) engineOptions { var result engineOptions for _, option := range options { @@ -264,6 +274,7 @@ func NewEngineWithShardContext( outboundQueueCBPool: outboundQueueCBPool, testHooks: testHooks, chasmEngine: chasmEngine, + streamClient: engineOptions.streamClient, versionCache: versionCache, workerDeploymentClient: workerDeploymentClient, routingInfoCache: routingInfoCache, @@ -602,7 +613,7 @@ func (e *historyEngineImpl) RecordWorkflowTaskStarted( request *historyservice.RecordWorkflowTaskStartedRequest, ) (*historyservice.RecordWorkflowTaskStartedResponseWithRawHistory, error) { return recordworkflowtaskstarted.Invoke( - ctx, + recordworkflowtaskstarted.WithStreamClient(ctx, e.streamClient), request, e.shardContext, e.config, @@ -629,7 +640,7 @@ func (e *historyEngineImpl) RespondWorkflowTaskCompleted( e.matchingClient, e.versionCache, ) - return h.Invoke(ctx, req) + return h.Invoke(recordworkflowtaskstarted.WithStreamClient(ctx, e.streamClient), req) } // RespondWorkflowTaskFailed fails a workflow task diff --git a/service/history/history_engine_factory.go b/service/history/history_engine_factory.go index a2dd26c2324..fb75ae56779 100644 --- a/service/history/history_engine_factory.go +++ b/service/history/history_engine_factory.go @@ -3,6 +3,7 @@ package history import ( "go.opentelemetry.io/otel/trace" "go.temporal.io/server/chasm" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/client" "go.temporal.io/server/common/persistence" @@ -53,6 +54,7 @@ type ( PersistenceRateLimiter replication.PersistenceRateLimiter TestHooks testhooks.TestHooks ChasmEngine chasm.Engine + StreamClient streampb.StreamServiceClient VersionMembershipCache worker_versioning.VersionMembershipAndReactivationStatusCache WorkerDeploymentClient workerdeployment.Client RoutingInfoCache worker_versioning.RoutingInfoCache @@ -97,5 +99,6 @@ func (f *historyEngineFactory) CreateEngine( f.TestHooks, f.ChasmEngine, WithWorkflowResendScheduler(f.WorkflowResendScheduler), + WithStreamClient(f.StreamClient), ) } diff --git a/service/history/historybuilder/history_builder_test.go b/service/history/historybuilder/history_builder_test.go index 3b5835e18a5..9e21c34901a 100644 --- a/service/history/historybuilder/history_builder_test.go +++ b/service/history/historybuilder/history_builder_test.go @@ -2325,7 +2325,8 @@ func (s *historyBuilderSuite) TestBufferEvent() { commandType := enumspb.CommandType(ct) // Unspecified is not counted. // ProtocolMessage command doesn't have corresponding event. - if commandType == enumspb.COMMAND_TYPE_UNSPECIFIED || commandType == enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE { + if commandType == enumspb.COMMAND_TYPE_UNSPECIFIED || + commandType == enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE { continue } commandsWithEventsCount++ diff --git a/service/history/stream_routing_test.go b/service/history/stream_routing_test.go new file mode 100644 index 00000000000..397829bc3ce --- /dev/null +++ b/service/history/stream_routing_test.go @@ -0,0 +1,18 @@ +package history + +import ( + "testing" + + "github.com/stretchr/testify/require" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +type testStreamRoutingClient struct { + streampb.StreamServiceClient +} + +func TestWithStreamClient(t *testing.T) { + client := &testStreamRoutingClient{} + options := applyEngineOptions([]EngineOption{WithStreamClient(client)}) + require.Same(t, client, options.streamClient) +} diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go index 0be52cd7f98..9e5301f850e 100644 --- a/service/history/workflow/mutable_state_impl.go +++ b/service/history/workflow/mutable_state_impl.go @@ -670,6 +670,79 @@ func (ms *MutableStateImpl) mustInitHSM() { ms.stateMachineNode = stateMachineNode } +// carryStreamSubscriptionsTo hands this run's subscriptions to the run that +// continues it. +// +// A cursor is workflow state, so without this a continue-as-new would silently +// end a subscription the workflow never cancelled: the stream would keep its +// consumer pin and keep pushing to the workflow id, and the successor would +// have nowhere to put it. +func (ms *MutableStateImpl) carryStreamSubscriptionsTo(newMutableState *MutableStateImpl) error { + if !ms.HasChasmWorkflowComponent() { + return nil + } + wf, chasmCtx, err := ms.ChasmWorkflowComponentReadOnly(context.Background()) + if err != nil { + return err + } + subscriptions := wf.ExportStreamSubscriptions(chasmCtx) + if len(subscriptions) == 0 { + return nil + } + + newMutableState.EnsureChasmWorkflowComponent(context.Background()) + newWorkflow, newChasmCtx, err := newMutableState.ChasmWorkflowComponent(context.Background()) + if err != nil { + return err + } + return newWorkflow.ImportStreamSubscriptions(newChasmCtx, subscriptions) +} + +// closeActivityStreams ends the streams an activity owns once it reaches a +// terminal status. +// +// An activity a workflow scheduled keeps its streams in the workflow's map, and +// the workflow is still running, so nothing else tells a reader tailing the +// activity that it is over. The read-only check comes first because taking the +// component for writing persists it, and most activities own no streams. +func (ms *MutableStateImpl) closeActivityStreams(activityID string) error { + if !ms.HasChasmWorkflowComponent() { + return nil + } + wf, chasmCtx, err := ms.ChasmWorkflowComponentReadOnly(context.Background()) + if err != nil { + return err + } + if !wf.HasOpenActivityStreams(chasmCtx, activityID) { + return nil + } + wf, mutableCtx, err := ms.ChasmWorkflowComponent(context.Background()) + if err != nil { + return err + } + return wf.CloseActivityStreams(mutableCtx, activityID) +} + +// deleteTerminalActivity deletes an activity that reached a terminal status, +// ending the streams it owns first. +func (ms *MutableStateImpl) deleteTerminalActivity(scheduledEventID int64) error { + if ai, ok := ms.pendingActivityInfoIDs[scheduledEventID]; ok { + if err := ms.closeActivityStreams(ai.GetActivityId()); err != nil { + return err + } + } + return ms.DeleteActivity(scheduledEventID) +} + +func (ms *MutableStateImpl) HasChasmWorkflowComponent() bool { + node, ok := ms.chasmTree.(*chasm.Node) + if !ok { + return false + } + _, err := node.ComponentByPath(chasm.NewContext(context.Background(), node), nil) + return err == nil +} + func (ms *MutableStateImpl) IsWorkflow() bool { return ms.chasmTree.ArchetypeID() == chasm.WorkflowArchetypeID } @@ -4630,7 +4703,7 @@ func (ms *MutableStateImpl) ApplyActivityTaskCompletedEvent( attributes := event.GetActivityTaskCompletedEventAttributes() scheduledEventID := attributes.GetScheduledEventId() - return ms.DeleteActivity(scheduledEventID) + return ms.deleteTerminalActivity(scheduledEventID) } func (ms *MutableStateImpl) AddActivityTaskFailedEvent( @@ -4680,7 +4753,7 @@ func (ms *MutableStateImpl) ApplyActivityTaskFailedEvent( attributes := event.GetActivityTaskFailedEventAttributes() scheduledEventID := attributes.GetScheduledEventId() - return ms.DeleteActivity(scheduledEventID) + return ms.deleteTerminalActivity(scheduledEventID) } func (ms *MutableStateImpl) AddActivityTaskTimedOutEvent( @@ -4732,7 +4805,7 @@ func (ms *MutableStateImpl) ApplyActivityTaskTimedOutEvent( attributes := event.GetActivityTaskTimedOutEventAttributes() scheduledEventID := attributes.GetScheduledEventId() - return ms.DeleteActivity(scheduledEventID) + return ms.deleteTerminalActivity(scheduledEventID) } func (ms *MutableStateImpl) AddActivityTaskCancelRequestedEvent( @@ -4944,7 +5017,7 @@ func (ms *MutableStateImpl) ApplyActivityTaskCanceledEvent( attributes := event.GetActivityTaskCanceledEventAttributes() scheduledEventID := attributes.GetScheduledEventId() - return ms.DeleteActivity(scheduledEventID) + return ms.deleteTerminalActivity(scheduledEventID) } func (ms *MutableStateImpl) AddCompletedWorkflowEvent( @@ -6400,6 +6473,10 @@ func (ms *MutableStateImpl) AddContinueAsNewEvent( return nil, nil, err } + if err = ms.carryStreamSubscriptionsTo(newMutableState); err != nil { + return nil, nil, err + } + if err = ms.ApplyWorkflowExecutionContinuedAsNewEvent( batchID, continueAsNewEvent, diff --git a/tests/stream_consume_test.go b/tests/stream_consume_test.go new file mode 100644 index 00000000000..13db3915c18 --- /dev/null +++ b/tests/stream_consume_test.go @@ -0,0 +1,25 @@ +package tests + +import ( + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/api/workflowservice/v1" +) + +func signalWorkflow(t *testing.T, s *streamTestEnv, workflowID, runID string) { + t.Helper() + _, err := s.env.FrontendClient().SignalWorkflowExecution( + s.ctx(), + &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: s.ns, + WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, + SignalName: "wake", + Identity: "tester", + RequestId: uuid.NewString(), + }, + ) + require.NoError(t, err) +} diff --git a/tests/stream_consumer_lifecycle_test.go b/tests/stream_consumer_lifecycle_test.go new file mode 100644 index 00000000000..9f14115b45c --- /dev/null +++ b/tests/stream_consumer_lifecycle_test.go @@ -0,0 +1,16 @@ +package tests + +import ( + commandpb "go.temporal.io/api/command/v1" + enumspb "go.temporal.io/api/enums/v1" +) + +func completeWorkflowCommand() []*commandpb.Command { + attrs := &commandpb.CompleteWorkflowExecutionCommandAttributes{} + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{ + CompleteWorkflowExecutionCommandAttributes: attrs, + }, + }} +} diff --git a/tests/stream_owner_test.go b/tests/stream_owner_test.go index 574b49f9cc6..1cf5e8394ef 100644 --- a/tests/stream_owner_test.go +++ b/tests/stream_owner_test.go @@ -399,3 +399,75 @@ func TestStreamOwnerDefaultsToTheNamedWorkflow(t *testing.T) { owner := &streamlib.StreamOwner{Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW, Id: workflowID} require.Equal(t, []string{"by id"}, bodies(s.pollOwned(t, owner, "", 0, false).GetRecords())) } + +// A workflow's activity keeps its streams across a retry and they end when the +// activity reaches a terminal status, while the workflow is still running. A +// reader tailing the activity is released then, not at workflow close. +func TestStreamWorkflowActivityStreamsEndWithTheActivity(t *testing.T) { + env, s := newStreamActivityEnv(t) + + workflowID := "stream-wfa-end-" + uuid.NewString() + scheduleStreamingActivity(t, env, s, workflowID, "model-call", streamActivityRetryPolicy) + owner := &streamlib.StreamOwner{ + Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW_ACTIVITY, + Id: workflowID, + ActivityId: "model-call", + } + + first := pollActivityTask(t, env, s, workflowID+"-tq") + _, err := s.addOwned(t, owner, "", &streamlib.AddWorkflowMessagesInput{ + Records: attemptRecords(1, "partial answer"), ProducerId: "worker#1", Sequence: 1, + }) + require.NoError(t, err) + failActivityAttempt(t, env, s, first) + require.False(t, s.describeOwned(t, owner, "").GetClosed(), + "a failed attempt with a retry to come does not end the activity's stream") + + second := pollActivityTask(t, env, s, workflowID+"-tq") + require.EqualValues(t, 2, second.GetAttempt()) + _, err = s.addOwned(t, owner, "", &streamlib.AddWorkflowMessagesInput{ + Records: attemptRecords(2, "full answer"), ProducerId: "worker#2", Sequence: 1, + }) + require.NoError(t, err) + read := s.pollOwned(t, owner, "", 0, false) + require.Equal(t, []string{"partial answer", "full answer"}, bodies(read.GetRecords()), + "the retry inherits the stream") + + type result struct { + out *streamlib.PollMessagesOutput + err error + } + done := make(chan result, 1) + go func() { + resp, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, Owner: owner, FromOffset: read.GetNextOffset(), WaitNewMessages: true, + }, + }) + done <- result{resp.GetFrontendResponse(), err} + }() + // Give the reader time to park, so the wake is what releases it. Nothing + // observable marks a parked reader, so a wait is the only option. + time.Sleep(500 * time.Millisecond) //nolint:forbidigo + completeActivityAttempt(t, env, s, second) + + select { + case r := <-done: + require.NoError(t, r.err) + require.True(t, r.out.GetClosed(), "the parked reader learns the activity ended") + case <-time.After(25 * time.Second): + t.Fatal("the parked reader was not released when the activity completed") + } + + require.True(t, s.describeOwned(t, owner, "").GetClosed()) + _, err = s.addOwned(t, owner, "", &streamlib.AddWorkflowMessagesInput{ + Records: attemptRecords(2, "too late"), + }) + require.Equal(t, codes.FailedPrecondition, status.Code(err), "%v", err) + + // The workflow is still running, so its own stream is not ended. + workflowOwner := &streamlib.StreamOwner{ + Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW, Id: workflowID, + } + require.False(t, s.describeOwned(t, workflowOwner, "").GetClosed()) +} diff --git a/tests/stream_rejection_test.go b/tests/stream_rejection_test.go new file mode 100644 index 00000000000..e4bc60fab74 --- /dev/null +++ b/tests/stream_rejection_test.go @@ -0,0 +1,156 @@ +package tests + +import ( + "testing" + + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/workflowservice/v1" + chasmstream "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + "go.temporal.io/server/tests/testcore" +) + +// A refused publish has to fail the workflow task with a cause, not the +// completion call. Failing the call leaves the worker retrying the same command +// against the same limit until the task times out, with nothing in History +// saying why. +func TestOverLimitPublishFailsTheWorkflowTask(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + execution, tq := startConsumer(t, s, "stream-wf-reject-publish-") + + tooMany := make([]string, chasmstream.MaxRecordsPerBatch+1) + for i := range tooMany { + tooMany[i] = "x" + } + task := 0 + //nolint:staticcheck // SA1019: only the deprecated poller can emit this command type. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + task++ + if task == 1 { + return publishCommand(tooMany...), nil + } + return publishCommand("fits"), nil + }, + Logger: env.Logger, + T: t, + } + + _, err := poller.PollAndProcessWorkflowTask() + require.Error(t, err, "the completion is rejected") + require.ErrorContains(t, err, "exceeds the limit") + + events := env.GetHistory(s.ns, execution) + failed := workflowTaskFailedWith(events, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES) + require.NotNil(t, failed, "the refusal is a workflow task failure with its own cause") + require.ErrorContains(t, err, failed.GetFailure().GetMessage()) + + desc, err := s.client.DescribeWorkflowStream(s.ctx(), &streamlib.DescribeWorkflowStreamRequest{ + FrontendRequest: &streamlib.DescribeWorkflowStreamInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + }, + }) + require.NoError(t, err) + require.Equal(t, int64(0), desc.GetFrontendResponse().GetState().GetHeadOffset(), + "a refused publish appends nothing") + + // The next attempt is a fresh task, and a publish that fits lands at the + // start of the stream. + _, err = poller.PollAndProcessWorkflowTask(testcore.WithExpectedAttemptCount(2)) + require.NoError(t, err) + poll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"fits"}, bodies(poll.GetFrontendResponse().GetRecords())) + require.Equal(t, []int64{0}, offsets(poll.GetFrontendResponse().GetRecords())) +} + +// The guarantee the design is sold on: a publish commits with the workflow +// task, so a task that fails after publishing publishes nothing. +func TestPublishOnAFailedTaskLeavesNoTrace(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + execution, tq := startConsumer(t, s, "stream-wf-reject-atomic-") + + // A ScheduleActivityTask with no task queue fails command validation, so the + // whole task is failed after the publish command already ran. + invalidTrailer := &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK, + Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{ + ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{ + ActivityId: "no-queue", + ActivityType: &commonpb.ActivityType{Name: "noop"}, + }, + }, + } + task := 0 + //nolint:staticcheck // SA1019: only the deprecated poller can emit this command type. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + task++ + if task == 1 { + return append(publishCommand("lost-with-the-task"), invalidTrailer), nil + } + return publishCommand("first-to-land"), nil + }, + Logger: env.Logger, + T: t, + } + + _, err := poller.PollAndProcessWorkflowTask() + require.Error(t, err) + + events := env.GetHistory(s.ns, execution) + require.NotNil(t, workflowTaskFailedWith(events, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_ACTIVITY_ATTRIBUTES)) + for _, e := range events { + require.Nil(t, e.GetWorkflowStreamRecordsAppendedEventAttributes(), + "a publish on a failed task leaves no event") + } + desc, err := s.client.DescribeWorkflowStream(s.ctx(), &streamlib.DescribeWorkflowStreamRequest{ + FrontendRequest: &streamlib.DescribeWorkflowStreamInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + }, + }) + require.NoError(t, err) + require.Equal(t, int64(0), desc.GetFrontendResponse().GetState().GetHeadOffset()) + poll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Empty(t, poll.GetFrontendResponse().GetRecords()) + + // The retry's publish is the first thing the stream ever holds. + _, err = poller.PollAndProcessWorkflowTask(testcore.WithExpectedAttemptCount(2)) + require.NoError(t, err) + poll, err = s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"first-to-land"}, bodies(poll.GetFrontendResponse().GetRecords())) + require.Equal(t, []int64{0}, offsets(poll.GetFrontendResponse().GetRecords())) +} diff --git a/tests/stream_replay_test.go b/tests/stream_replay_test.go new file mode 100644 index 00000000000..442b75a9164 --- /dev/null +++ b/tests/stream_replay_test.go @@ -0,0 +1,69 @@ +package tests + +import ( + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + streampb "go.temporal.io/api/stream/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/protobuf/types/known/durationpb" +) + +// startConsumer starts a workflow on its own task queue and returns the +// execution and the queue. +func startConsumer( + t *testing.T, s *streamTestEnv, prefix string, +) (*commonpb.WorkflowExecution, *taskqueuepb.TaskQueue) { + t.Helper() + id := prefix + uuid.NewString() + tq := &taskqueuepb.TaskQueue{Name: id + "-tq", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + we, err := s.env.FrontendClient().StartWorkflowExecution( + s.ctx(), + &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: s.ns, + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: "stream-consumer"}, + TaskQueue: tq, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: "tester", + }, + ) + require.NoError(t, err) + return &commonpb.WorkflowExecution{WorkflowId: id, RunId: we.GetRunId()}, tq +} + +func publishCommand(bodies ...string) []*commandpb.Command { + messages := make([]*streampb.StreamRecord, len(bodies)) + for i, b := range bodies { + messages[i] = &streampb.StreamRecord{Body: &commonpb.Payload{Data: []byte(b)}, Topic: "tokens"} + } + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: messages, + }, + }, + }} +} + +func workflowTaskFailedWith( + events []*historypb.HistoryEvent, + cause enumspb.WorkflowTaskFailedCause, +) *historypb.WorkflowTaskFailedEventAttributes { + for _, e := range events { + if attrs := e.GetWorkflowTaskFailedEventAttributes(); attrs != nil && attrs.GetCause() == cause { + return attrs + } + } + return nil +} diff --git a/tests/stream_subscribe_start_test.go b/tests/stream_subscribe_start_test.go new file mode 100644 index 00000000000..cfa7e37d9ca --- /dev/null +++ b/tests/stream_subscribe_start_test.go @@ -0,0 +1,135 @@ +package tests + +import ( + "testing" + + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + streampb "go.temporal.io/api/stream/v1" + "go.temporal.io/api/workflowservice/v1" + chasmstream "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + "go.temporal.io/server/tests/testcore" +) + +// A subscribe command names where it starts. The server resolves the position +// when it registers the subscription and records the absolute offset on the +// event, so a replay never resolves it again. + +func subscribeFromCommand( + nameOrID string, offset int64, start *streampb.StreamStartPosition, +) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SUBSCRIBE_STREAM, + Attributes: &commandpb.Command_SubscribeStreamCommandAttributes{ + SubscribeStreamCommandAttributes: &commandpb.SubscribeStreamCommandAttributes{ + StreamNameOrId: nameOrID, StartOffset: offset, StartPosition: start, + }, + }, + } +} + +func subscribedStarts(events []*historypb.HistoryEvent) map[string]int64 { + out := map[string]int64{} + for _, e := range events { + if a := e.GetWorkflowStreamSubscribedEventAttributes(); a != nil { + out[a.GetStreamId()] = a.GetStartOffset() + } + } + return out +} + +func TestStreamSubscribeCommandRecordsTheResolvedStart(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + ctx := streamCtx(t) + execution, tq := startConsumer(t, s, "stream-sub-start-") + + // A standalone stream whose first two records were truncated away. Offset + // zero names nothing it still holds, which earliest has to get right. + standalone := execution.GetWorkflowId() + "-standalone" + s.create(ctx, t, standalone) + _, err := s.add(ctx, t, standalone, &streamlib.AddMessagesInput{ + Records: streamMsgs("", "a", "b", "c", "d", "e"), + }) + require.NoError(t, err) + _, err = s.client.TruncateStream(ctx, &streamlib.TruncateStreamRequest{ + FrontendRequest: &streamlib.TruncateStreamInput{ + Namespace: s.ns, StreamId: standalone, NewBaseOffset: 2, + }, + }) + require.NoError(t, err) + + owner := &streamlib.StreamOwner{ + Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW, Id: execution.GetWorkflowId(), + } + _, err = s.addOwned(t, owner, "inputs", &streamlib.AddWorkflowMessagesInput{ + Records: streamMsgs("", "x", "y", "z"), + }) + require.NoError(t, err) + + //nolint:staticcheck // SA1019: only the deprecated poller can emit this command type. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + return []*commandpb.Command{ + subscribeFromCommand(standalone, 0, chasmstream.Earliest()), + subscribeFromCommand("inputs", 0, chasmstream.LastN(1)), + subscribeFromCommand("not-yet-written", 0, chasmstream.Tail()), + }, nil + }, + Logger: env.Logger, + T: t, + } + _, err = poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + + require.Equal(t, map[string]int64{ + standalone: 2, + "inputs": 2, + "not-yet-written": 0, + }, subscribedStarts(env.GetHistory(s.ns, execution))) +} + +func TestStreamSubscribeCommandRefusesAStartItCannotResolve(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + + for _, command := range []*commandpb.Command{ + subscribeFromCommand("inputs", 1, chasmstream.Earliest()), + subscribeFromCommand("inputs", 0, chasmstream.LastN(0)), + subscribeFromCommand("inputs", 0, &streampb.StreamStartPosition{}), + // The old sentinel for the head is refused rather than reinterpreted. + subscribeFromCommand("inputs", -1, nil), + } { + execution, tq := startConsumer(t, s, "stream-sub-start-refused-") + //nolint:staticcheck // SA1019: only the deprecated poller can emit this command type. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + return []*commandpb.Command{command}, nil + }, + Logger: env.Logger, + T: t, + } + _, err := poller.PollAndProcessWorkflowTask() + require.Error(t, err) + + events := env.GetHistory(s.ns, execution) + require.NotNil(t, workflowTaskFailedWith(events, + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES)) + require.Empty(t, subscribedStarts(events), "a refused subscribe registers nothing") + } +} diff --git a/tests/stream_workflow_test.go b/tests/stream_workflow_test.go new file mode 100644 index 00000000000..9c829551064 --- /dev/null +++ b/tests/stream_workflow_test.go @@ -0,0 +1,553 @@ +package tests + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + streampb "go.temporal.io/api/stream/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + chasmstream "go.temporal.io/server/chasm/lib/stream" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" + "go.temporal.io/server/tests/testcore" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" +) + +// Path A: a workflow publishing to a stream it owns. The stream is co-located +// with the workflow, so the frontier advances in the workflow task's own +// commit, and History gets one fixed-size event naming the offset range rather +// than anything that was published. +// +// Driven through the raw task poller rather than an SDK, because emitting a new +// command type does not need one. +func TestStreamWorkflowPublishesWithARangeEvent(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + + id := "stream-wf-publish-" + uuid.NewString() + tq := &taskqueuepb.TaskQueue{Name: id + "-tq", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + + we, err := env.FrontendClient().StartWorkflowExecution( + s.ctx(), + &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: s.ns, + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: "stream-publisher"}, + TaskQueue: tq, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: "tester", + }, + ) + require.NoError(t, err) + + published := false + // The newer taskpoller cannot emit an arbitrary command type, which is + // the whole point here. + //nolint:staticcheck // SA1019: deprecated poller is the only one that can. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + if published { + return completeWorkflowCommand(), nil + } + published = true + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: []*streampb.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("planning")}, Topic: "progress"}, + {Body: &commonpb.Payload{Data: []byte("calling tool")}, Topic: "progress"}, + }, + }, + }, + }}, nil + }, + Logger: env.Logger, + T: t, + } + + _, err = poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + + // One event for the batch, holding the range and none of the payload. Two + // messages were published, so it has to name both of them and stop there. + events := env.GetHistory(s.ns, &commonpb.WorkflowExecution{WorkflowId: id, RunId: we.GetRunId()}) + var added []*historypb.WorkflowStreamRecordsAppendedEventAttributes + for _, e := range events { + if a := e.GetWorkflowStreamRecordsAppendedEventAttributes(); a != nil { + added = append(added, a) + } + } + require.Len(t, added, 1, "one publish command writes one event") + require.Equal(t, int64(0), added[0].GetFromOffset()) + require.Equal(t, int64(2), added[0].GetToOffset()) + require.Equal(t, chasmworkflow.DefaultStreamName, added[0].GetStreamId(), + "an unnamed stream resolves to the default before it is recorded") + + // The bodies stay out of History. Asserted on the serialized event rather + // than on its fields, because a field this test forgot to check would still + // be carrying them. + raw, err := added[0].Marshal() + require.NoError(t, err) + require.NotContains(t, string(raw), "calling tool", + "the event must name the range, never carry the payload") + + // The bodies are readable from outside the workflow, which is the point of + // publishing them: an attached stream has no id, so it is addressed by its + // owner and its name. + poll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"planning", "calling tool"}, + bodies(poll.GetFrontendResponse().GetRecords())) + require.Equal(t, []int64{0, 1}, offsets(poll.GetFrontendResponse().GetRecords()), + "each message carries where it sits in the log, so a reader can resume between them") + require.Equal(t, int64(2), poll.GetFrontendResponse().GetNextOffset()) + require.Equal(t, int64(2), poll.GetFrontendResponse().GetHeadOffset()) + + // The topic filter applies to an attached stream the same as to any other, + // and offsets stay assigned over the unfiltered stream. + filtered, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, FromOffset: 0, Topics: []string{"nothing-here"}, + }, + }) + require.NoError(t, err) + require.Empty(t, filtered.GetFrontendResponse().GetRecords()) + require.Equal(t, int64(2), filtered.GetFrontendResponse().GetNextOffset(), + "a filtered page still advances the reader") + + // A stream the workflow has not published to reads as an empty one, which + // is what lets a reader attach before the first event. + unwritten, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, StreamName: "not-published-to-yet", FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Empty(t, unwritten.GetFrontendResponse().GetRecords()) + require.Equal(t, int64(0), unwritten.GetFrontendResponse().GetHeadOffset()) + require.False(t, unwritten.GetFrontendResponse().GetClosed()) + + // A workflow that does not exist is still an error, so a mistyped owner is + // not mistaken for a stream with nothing in it. + _, err = s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id + "-does-not-exist", FromOffset: 0, + }, + }) + require.ErrorContains(t, err, "not found") + + // An attached stream still has no standalone id, so the id-addressed read + // must not reach it. + streamID := we.GetRunId() + "/" + chasmworkflow.DefaultStreamName + _, err = s.client.PollMessages(s.ctx(), &streamlib.PollMessagesRequest{ + FrontendRequest: &streamlib.PollMessagesInput{ + Namespace: s.ns, StreamId: streamID, FromOffset: 0, + }, + }) + require.ErrorContains(t, err, "stream not found", + "an attached stream is only reachable through its workflow") +} + +// A reader attached to a workflow's stream before the workflow published +// anything. This is the ordinary case for a UI that opens on a session and +// waits for the agent's first token, so the poll has to park on a stream that +// does not exist yet and wake when the workflow creates it. +func TestStreamWorkflowLongPollWakesOnPublish(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + + id := "stream-wf-longpoll-" + uuid.NewString() + tq := &taskqueuepb.TaskQueue{Name: id + "-tq", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + + _, err := env.FrontendClient().StartWorkflowExecution( + s.ctx(), + &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: s.ns, + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: "stream-publisher"}, + TaskQueue: tq, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: "tester", + }, + ) + require.NoError(t, err) + + type result struct { + out *streamlib.PollMessagesOutput + err error + } + done := make(chan result, 1) + go func() { + resp, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, FromOffset: 0, WaitNewMessages: true, + }, + }) + done <- result{resp.GetFrontendResponse(), err} + }() + + //nolint:staticcheck // SA1019: deprecated poller is the only one that can emit the command. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: []*streampb.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("first token")}}, + }, + }, + }, + }}, nil + }, + Logger: env.Logger, + T: t, + } + _, err = poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + + select { + case r := <-done: + require.NoError(t, r.err) + require.Equal(t, []string{"first token"}, bodies(r.out.GetRecords())) + require.Equal(t, int64(1), r.out.GetNextOffset()) + case <-time.After(25 * time.Second): + t.Fatal("the parked reader did not wake when the workflow published") + } +} + +// Two producers on one stream: the workflow, whose publishes ride its Workflow +// Task, and something outside it. This is the shape of an agent session, where +// a model activity produces the tokens and the workflow only brackets them. +func TestStreamWorkflowTakesAppendsFromOutsideToo(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + + id := "stream-wf-mixed-" + uuid.NewString() + tq := &taskqueuepb.TaskQueue{Name: id + "-tq", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + + _, err := env.FrontendClient().StartWorkflowExecution( + s.ctx(), + &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: s.ns, + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: "stream-publisher"}, + TaskQueue: tq, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: "tester", + }, + ) + require.NoError(t, err) + + appendOutside := func(body string) *streamlib.AddMessagesOutput { + t.Helper() + resp, err := s.client.AddWorkflowMessages(s.ctx(), &streamlib.AddWorkflowMessagesRequest{ + FrontendRequest: &streamlib.AddWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, + Records: []*streamlib.StreamRecord{{ + Body: &commonpb.Payload{Data: []byte(body)}, + Kind: streampb.STREAM_RECORD_KIND_DATA, + }}, + }, + }) + require.NoError(t, err) + return resp.GetFrontendResponse() + } + + // The first writer creates the stream, and here that is not the workflow. + first := appendOutside("token one") + require.Equal(t, int64(0), first.GetFirstOffset()) + + //nolint:staticcheck // SA1019: deprecated poller is the only one that can emit the command. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: []*streampb.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("turn ended")}}, + }, + }, + }, + }}, nil + }, + Logger: env.Logger, + T: t, + } + _, err = poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + + // The workflow appended after the outside writer, so it has to continue the + // same log rather than start its own. + require.Equal(t, int64(2), appendOutside("token two").GetFirstOffset()) + + poll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"token one", "turn ended", "token two"}, + bodies(poll.GetFrontendResponse().GetRecords()), + "both producers write one ordered log") + + // The event names the offset the workflow's own publish landed at, which + // only holds if the workflow saw the outside writer's message first. + events := env.GetHistory(s.ns, &commonpb.WorkflowExecution{WorkflowId: id}) + var added []*historypb.WorkflowStreamRecordsAppendedEventAttributes + for _, e := range events { + if a := e.GetWorkflowStreamRecordsAppendedEventAttributes(); a != nil { + added = append(added, a) + } + } + require.Len(t, added, 1) + require.Equal(t, int64(1), added[0].GetFromOffset()) +} + +// A reader tailing a workflow that ends has to be released. Nothing can be +// added to a stream inside a closed execution, so a reader parked on it would +// otherwise wait for a message that can never come. +func TestStreamWorkflowStreamClosesWithItsWorkflow(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + + id := "stream-wf-close-" + uuid.NewString() + tq := &taskqueuepb.TaskQueue{Name: id + "-tq", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + + _, err := env.FrontendClient().StartWorkflowExecution( + s.ctx(), + &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: s.ns, + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: "stream-publisher"}, + TaskQueue: tq, + WorkflowRunTimeout: durationpb.New(100 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: "tester", + }, + ) + require.NoError(t, err) + + describe := func() *streamlib.StreamState { + t.Helper() + resp, err := s.client.DescribeWorkflowStream(s.ctx(), &streamlib.DescribeWorkflowStreamRequest{ + FrontendRequest: &streamlib.DescribeWorkflowStreamInput{ + Namespace: s.ns, WorkflowId: id, + }, + }) + require.NoError(t, err) + return resp.GetFrontendResponse().GetState() + } + + //nolint:staticcheck // SA1019: deprecated poller is the only one that can emit the command. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + return []*commandpb.Command{ + { + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: []*streampb.StreamRecord{ + {Body: &commonpb.Payload{Data: []byte("last word")}}, + }, + }, + }, + }, + completeWorkflowCommand()[0], + }, nil + }, + Logger: env.Logger, + T: t, + } + _, err = poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + + require.True(t, describe().GetClosed(), "the stream of a completed workflow reads as closed") + + // Closed does not mean gone. Everything published before the workflow + // ended is still there to be read. + poll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: id, FromOffset: 0, WaitNewMessages: true, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"last word"}, bodies(poll.GetFrontendResponse().GetRecords())) + require.True(t, poll.GetFrontendResponse().GetClosed()) +} + +// An outside producer and the workflow's own publishes land in one log, with +// neither pinning the head against the other. Run concurrently, every append +// has to succeed and every message has to appear once, in a contiguous range. +func TestOutsideAppendsRaceTheWorkflowPublishWithoutFailing(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + execution, tq := startConsumer(t, s, "stream-wf-race-") + + const producers = 8 + const perProducer = 10 + const workflowTasks = 5 + + //nolint:staticcheck // SA1019: deprecated poller is the only one that can emit the command. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + return publishCommand("from-workflow"), nil + }, + Logger: env.Logger, + T: t, + } + + errs := make(chan error, producers*perProducer) + var wg sync.WaitGroup + for p := range producers { + wg.Go(func() { + for i := range perProducer { + _, err := s.client.AddWorkflowMessages(s.ctx(), &streamlib.AddWorkflowMessagesRequest{ + FrontendRequest: &streamlib.AddWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + Records: []*streamlib.StreamRecord{{ + Body: &commonpb.Payload{Data: fmt.Appendf(nil, "outside-%d-%d", p, i)}, + }}, + }, + }) + errs <- err + } + }) + } + for range workflowTasks { + _, err := poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + signalWorkflow(t, s, execution.GetWorkflowId(), execution.GetRunId()) + } + wg.Wait() + close(errs) + for err := range errs { + require.NoError(t, err, "an outside append must not fail because the workflow published") + } + + total := producers*perProducer + workflowTasks + poll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Equal(t, int64(total), poll.GetFrontendResponse().GetHeadOffset()) + seen := make(map[string]int, total) + for _, body := range bodies(poll.GetFrontendResponse().GetRecords()) { + seen[body]++ + } + require.Len(t, seen, producers*perProducer+1, "every outside message appears exactly once") + require.Equal(t, workflowTasks, seen["from-workflow"]) +} + +// A stream a workflow owns has a budget, because its batches are the workflow's +// own mutable state. Past it the workflow's publish fails its task with a cause +// and an outside producer is told the stream is full, instead of the execution +// size limit terminating the workflow later. +func TestOwnedStreamRefusesAppendsPastItsBudget(t *testing.T) { + env := testcore.NewEnv(t, testcore.WithDynamicConfig(chasmstream.OwnedStreamMaxItemsSetting, 3)) + s := newStreamTestEnvFrom(t, env) + execution, tq := startConsumer(t, s, "stream-wf-budget-") + + task := 0 + //nolint:staticcheck // SA1019: deprecated poller is the only one that can emit the command. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + task++ + if task == 1 { + return publishCommand("one", "two"), nil + } + return publishCommand("three", "four"), nil + }, + Logger: env.Logger, + T: t, + } + _, err := poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + + signalWorkflow(t, s, execution.GetWorkflowId(), execution.GetRunId()) + _, err = poller.PollAndProcessWorkflowTask() + require.Error(t, err, "two more do not fit in a budget of three") + failed := workflowTaskFailedWith(env.GetHistory(s.ns, execution), + enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES) + require.NotNil(t, failed) + require.Contains(t, failed.GetFailure().GetMessage(), "budget") + + appendOutside := func(bodies ...string) error { + messages := make([]*streamlib.StreamRecord, len(bodies)) + for i, b := range bodies { + messages[i] = &streamlib.StreamRecord{Body: &commonpb.Payload{Data: []byte(b)}} + } + _, err := s.client.AddWorkflowMessages(s.ctx(), &streamlib.AddWorkflowMessagesRequest{ + FrontendRequest: &streamlib.AddWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), Records: messages, + }, + }) + return err + } + // The test client is a bare gRPC connection, so the code is what it sees. + require.Equal(t, codes.ResourceExhausted, status.Code(appendOutside("three", "four"))) + require.NoError(t, appendOutside("three"), "the last slot is still there for one that fits") + require.Equal(t, codes.ResourceExhausted, status.Code(appendOutside("four"))) +}