diff --git a/chasm/lib/stream/cursor.go b/chasm/lib/stream/cursor.go index adb433ce709..143fc23f0b8 100644 --- a/chasm/lib/stream/cursor.go +++ b/chasm/lib/stream/cursor.go @@ -139,3 +139,23 @@ func (c *Cursor) AdvanceKnownHead(_ chasm.MutableContext, head int64) { func (c *Cursor) StartOffset() int64 { return c.State.StartOffset } + +// Restore puts the cursor where a completed task's event says it stood, for a +// run being rebuilt from its history. It only moves forward: the events are +// applied in order, and a range folded in earlier is never taken back. +func (c *Cursor) Restore(_ chasm.MutableContext, offset int64) { + if offset > c.State.Offset { + c.State.Offset = offset + } + if offset > c.State.KnownHead { + c.State.KnownHead = offset + } +} + +// MarkExternal says the stream lives in another execution, which the subscribe +// event alone cannot tell a rebuilt run, and carries over the frontier that +// stream last pushed at the run this one was rebuilt from. +func (c *Cursor) MarkExternal(mctx chasm.MutableContext, knownHead int64) { + c.State.External = true + c.AdvanceKnownHead(mctx, knownHead) +} diff --git a/chasm/lib/stream/stream.go b/chasm/lib/stream/stream.go index 03ff5fd04db..bac893fbe58 100644 --- a/chasm/lib/stream/stream.go +++ b/chasm/lib/stream/stream.go @@ -57,6 +57,12 @@ type NewStreamRequest struct { // child of the root, so an attached stream carries none and is found // through its owner instead of through ListStreams. Attached bool + + // StartOffset is where the stream's offsets begin. Zero for a new stream. + // A run created by a reset inherits a cursor whose position is written in + // its History, so the stream it goes on reading has to continue that + // offset space rather than start over at zero. + StartOffset int64 } type AddMessagesRequest struct { @@ -98,6 +104,9 @@ type AddMessagesResult struct { } func NewStream(ctx chasm.MutableContext, req NewStreamRequest) (*Stream, error) { + if req.StartOffset < 0 { + return nil, serviceerror.NewInvalidArgument("start offset cannot be negative") + } visibility := chasm.NewEmptyField[*chasm.Visibility]() if !req.Attached { visibility = chasm.NewComponentField(ctx, chasm.NewVisibility(ctx)) @@ -106,10 +115,12 @@ func NewStream(ctx chasm.MutableContext, req NewStreamRequest) (*Stream, error) Visibility: visibility, Batches: make(chasm.Map[int64, *commonpb.DataBlob]), State: &streamlib.StreamState{ - Lifecycle: req.Lifecycle, - Budget: req.Budget, - Producers: make(map[string]*streamlib.ProducerCursor), - Consumers: make(map[string]*streamlib.ConsumerCursor), + HeadOffset: req.StartOffset, + BaseOffset: req.StartOffset, + Lifecycle: req.Lifecycle, + Budget: req.Budget, + Producers: make(map[string]*streamlib.ProducerCursor), + Consumers: make(map[string]*streamlib.ConsumerCursor), }, }, nil } diff --git a/chasm/lib/workflow/stream_admission_test.go b/chasm/lib/workflow/stream_admission_test.go index f05e3db9ab3..433072887d9 100644 --- a/chasm/lib/workflow/stream_admission_test.go +++ b/chasm/lib/workflow/stream_admission_test.go @@ -162,3 +162,45 @@ func TestSubscriptionsPerWorkflowAreBounded(t *testing.T) { // not refused for room. require.NoError(t, subscribe("a")) } + +// A reset run's stream continues the base run's offset space, so it is born +// with a head offset well above zero while holding nothing. Measuring the +// budget against that head would put it over the moment it exists, and every +// publish on the reset run would be refused. +func TestAnInheritedStreamIsNotBornOverItsItemBudget(t *testing.T) { + ctx := newStreamBudgetTestContext() + w := &Workflow{} + limits := stream.Limits{ + MaxOwnedStreamsPerWorkflow: 10, + MaxConsumersPerStream: 10, + OwnedStreamMaxItems: 3, + OwnedStreamMaxBytes: 1 << 20, + } + + // The inherited cursor stands far past the item budget. + require.NoError(t, w.ownStreamFrom(ctx, DefaultStreamName, 100, limits)) + + _, err := w.AppendToOwnedStream(ctx, DefaultStreamName, stream.AddMessagesRequest{ + Records: budgetTestRecords(8), + Limits: limits, + }) + require.NoError(t, err, "the stream holds nothing, so its whole budget is free") + + owned := w.Streams[DefaultStreamName].Get(ctx) + require.Equal(t, int64(101), owned.State.GetHeadOffset()) + + // Still bounded against what it holds. + for range 2 { + _, err = w.AppendToOwnedStream(ctx, DefaultStreamName, stream.AddMessagesRequest{ + Records: budgetTestRecords(8), + Limits: limits, + }) + require.NoError(t, err) + } + _, err = w.AppendToOwnedStream(ctx, DefaultStreamName, stream.AddMessagesRequest{ + Records: budgetTestRecords(8), + Limits: limits, + }) + var exhausted *serviceerror.ResourceExhausted + require.ErrorAs(t, err, &exhausted) +} diff --git a/chasm/lib/workflow/stream_commands.go b/chasm/lib/workflow/stream_commands.go index 511baa242a9..7362623ab5f 100644 --- a/chasm/lib/workflow/stream_commands.go +++ b/chasm/lib/workflow/stream_commands.go @@ -316,6 +316,13 @@ func (w *Workflow) RecordStreamRecordsAppended( // 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. +// +// The stream belongs to this run. After a reset the new run publishes to and +// subscribes on streams of its own, created here on first use, and the run it +// was reset from keeps the records its own history refers to. A stream the +// reset run inherited a subscription to is created by the reset itself, at the +// offset that subscription stood at, so it is already here by the time a +// command names it. func (w *Workflow) streamNamed( ctx chasm.MutableContext, name string, diff --git a/chasm/lib/workflow/stream_cursor_test.go b/chasm/lib/workflow/stream_cursor_test.go index db65d0cc2fc..dfe88da5ae0 100644 --- a/chasm/lib/workflow/stream_cursor_test.go +++ b/chasm/lib/workflow/stream_cursor_test.go @@ -242,3 +242,159 @@ func TestConsumerOutrunByTruncationIsToldSo(t *testing.T) { require.Equal(t, int64(3), state.GetBaseOffset(), "the floor moved, which is what the consumer has to find out about") } + +func newStreamCursorTestContextForRun(runID string) chasm.MutableContext { + return &chasm.MockMutableContext{ + MockContext: chasm.MockContext{ + HandleExecutionKey: func() chasm.ExecutionKey { + return chasm.ExecutionKey{NamespaceID: "ns-1", BusinessID: "wf-1", RunID: runID} + }, + }, + } +} + +func subscribedEvent(streamID string, startOffset int64) *historypb.HistoryEvent { + return &historypb.HistoryEvent{ + EventType: enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED, + Attributes: &historypb.HistoryEvent_WorkflowStreamSubscribedEventAttributes{ + WorkflowStreamSubscribedEventAttributes: &historypb.WorkflowStreamSubscribedEventAttributes{ + StreamId: streamID, + StartOffset: startOffset, + }, + }, + } +} + +// A run rebuilt from its history, which is what a reset produces, gets its +// cursors back from the events alone: the subscribe event places the cursor +// and each completed task's recorded range moves it. +func TestRebuildRecreatesTheCursorFromItsEvents(t *testing.T) { + ctx := newStreamCursorTestContext() + w := &Workflow{} + + require.NoError(t, streamSubscribedEvent{}.Apply(ctx, w, subscribedEvent("inputs", 1))) + cursor := w.StreamCursors["inputs"].Get(ctx) + require.Equal(t, int64(1), cursor.StartOffset()) + require.Equal(t, int64(1), cursor.Offset()) + require.False(t, cursor.IsExternal(), "the event cannot say where the stream lives") + + require.NoError(t, w.ApplyConsumedStreamRanges(ctx, []*streampb.StreamRange{ + {StreamId: "inputs", FromOffset: 1, ToOffset: 4}, + {StreamId: "out-of-band", FromOffset: 3, ToOffset: 9}, + })) + require.Equal(t, int64(4), cursor.Offset()) + require.Equal(t, int64(1), cursor.StartOffset(), "where reading began does not move") + outOfBand, ok := w.StreamCursors["out-of-band"] + require.True(t, ok, "a recorded range proves a subscription the events never mentioned") + require.Equal(t, int64(3), outOfBand.Get(ctx).StartOffset()) + require.Equal(t, int64(9), outOfBand.Get(ctx).Offset()) + + // The same subscribe event applied twice, as a resubscribe would leave in + // History, must not rewind the cursor. + require.NoError(t, streamSubscribedEvent{}.Apply(ctx, w, subscribedEvent("inputs", 1))) + require.Equal(t, int64(4), w.StreamCursors["inputs"].Get(ctx).Offset()) +} + +// A reset run's cursor on a stream the base run owned gets a stream of the +// reset run's own, starting where the cursor stands, so the ranges below it +// stay in the base run and everything from here on is the reset run's. +func TestResetRunInheritsAnOwnedStreamAtItsCursor(t *testing.T) { + baseCtx := newStreamCursorTestContextForRun("base-run") + base := &Workflow{} + base.Streams = chasm.Map[string, *stream.Stream]{ + DefaultStreamName: chasm.NewComponentField(baseCtx, newAttachedStream(t, baseCtx, 4)), + } + _, err := base.SubscribeToOwnedStream(baseCtx, DefaultStreamName, 0, stream.DefaultLimits()) + require.NoError(t, err) + + resetCtx := newStreamCursorTestContextForRun("reset-run") + reset := &Workflow{} + require.NoError(t, + streamSubscribedEvent{}.Apply(resetCtx, reset, subscribedEvent(DefaultStreamName, 0))) + require.NoError(t, reset.ApplyConsumedStreamRanges(resetCtx, []*streampb.StreamRange{ + {StreamId: DefaultStreamName, FromOffset: 0, ToOffset: 2}, + })) + + require.NoError(t, reset.InheritStreamsOnReset(resetCtx, base, baseCtx, stream.DefaultLimits())) + + cursor := reset.StreamCursors[DefaultStreamName].Get(resetCtx) + require.False(t, cursor.IsExternal()) + require.Equal(t, int64(2), cursor.Offset()) + + own := reset.OwnedStream(resetCtx, DefaultStreamName) + require.NotNil(t, own, "the reset run reads and writes a stream of its own") + state, err := own.Snapshot(resetCtx, struct{}{}) + require.NoError(t, err) + require.Equal(t, int64(2), state.GetBaseOffset(), "the stream continues the offset space") + require.Equal(t, int64(2), state.GetHeadOffset()) + require.NotNil(t, state.GetBudget(), "an owned stream is budgeted like one created by a publish") + pin := state.GetConsumers()[streamConsumerID(DefaultStreamName)] + require.NotNil(t, pin, "the reset run pins its own stream") + require.Equal(t, "reset-run", pin.GetRunId()) + require.Equal(t, int64(2), pin.GetReplayFloor()) + + // The base run's stream is untouched: it still holds what the reset run's + // history refers to. + baseState, err := base.OwnedStream(baseCtx, DefaultStreamName).Snapshot(baseCtx, struct{}{}) + require.NoError(t, err) + require.Equal(t, int64(0), baseState.GetBaseOffset()) + require.Equal(t, int64(4), baseState.GetHeadOffset()) + + // A publish on the reset run lands after the inherited position. + result, err := own.AddMessages(resetCtx, stream.AddMessagesRequest{ + Records: []*streamlib.StreamRecord{{Kind: streampb.STREAM_RECORD_KIND_DATA}}, + Limits: stream.DefaultLimits(), + }) + require.NoError(t, err) + require.Equal(t, int64(2), result.FirstOffset) +} + +// A reset run's cursor on a stream in another execution stays on that stream +// and learns what the base run knew of its frontier. +func TestResetRunInheritsAnExternalCursorAsExternal(t *testing.T) { + baseCtx := newStreamCursorTestContextForRun("base-run") + base := &Workflow{} + _, err := base.SubscribeToExternalStream(baseCtx, ExternalStreamSubscription{ + StreamID: "shared", StartOffset: 0, KnownHead: 7, + }) + require.NoError(t, err) + + resetCtx := newStreamCursorTestContextForRun("reset-run") + reset := &Workflow{} + require.NoError(t, streamSubscribedEvent{}.Apply(resetCtx, reset, subscribedEvent("shared", 0))) + require.NoError(t, reset.ApplyConsumedStreamRanges(resetCtx, []*streampb.StreamRange{ + {StreamId: "shared", FromOffset: 0, ToOffset: 3}, + })) + + require.NoError(t, reset.InheritStreamsOnReset(resetCtx, base, baseCtx, stream.DefaultLimits())) + + cursor := reset.StreamCursors["shared"].Get(resetCtx) + require.True(t, cursor.IsExternal()) + require.Equal(t, int64(3), cursor.Offset()) + require.Equal(t, int64(7), cursor.KnownHead(), "the frontier the base run last knew carries over") + require.Nil(t, reset.OwnedStream(resetCtx, "shared"), "no stream of its own for an external one") +} + +// A subscription made through the service leaves no event, so a reset run whose +// history records nothing for it is given the cursor from the base run, where +// that subscription began. +func TestResetRunCarriesASubscriptionTheEventsNeverMentioned(t *testing.T) { + baseCtx := newStreamCursorTestContextForRun("base-run") + base := &Workflow{} + base.Streams = chasm.Map[string, *stream.Stream]{ + "inputs": chasm.NewComponentField(baseCtx, newAttachedStream(t, baseCtx, 4)), + } + _, err := base.SubscribeToOwnedStream(baseCtx, "inputs", 1, stream.DefaultLimits()) + require.NoError(t, err) + + resetCtx := newStreamCursorTestContextForRun("reset-run") + reset := &Workflow{} + require.NoError(t, reset.InheritStreamsOnReset(resetCtx, base, baseCtx, stream.DefaultLimits())) + + cursor := reset.StreamCursors["inputs"].Get(resetCtx) + require.Equal(t, int64(1), cursor.StartOffset()) + require.Equal(t, int64(1), cursor.Offset()) + state, err := reset.OwnedStream(resetCtx, "inputs").Snapshot(resetCtx, struct{}{}) + require.NoError(t, err) + require.Equal(t, int64(1), state.GetBaseOffset()) +} diff --git a/chasm/lib/workflow/workflow.go b/chasm/lib/workflow/workflow.go index 67b53df269a..0a43ef2d8d3 100644 --- a/chasm/lib/workflow/workflow.go +++ b/chasm/lib/workflow/workflow.go @@ -258,6 +258,171 @@ func (w *Workflow) ImportStreamSubscriptions( return nil } +// ApplyConsumedStreamRanges puts the cursors where the completed event says +// they stood, for a run being rebuilt from its history. +// +// A range for a stream with no cursor belongs to a subscription made out of +// band, through the service rather than by a command, which leaves no event of +// its own. The range is proof the subscription was live, so the cursor is +// created from it: the first range a subscription records begins where it +// started reading. +func (w *Workflow) ApplyConsumedStreamRanges( + mctx chasm.MutableContext, + ranges []*streampb.StreamRange, +) error { + for _, recorded := range ranges { + field, ok := w.StreamCursors[recorded.GetStreamId()] + if !ok { + cursor, err := stream.NewCursor(mctx, stream.NewCursorRequest{ + StreamID: recorded.GetStreamId(), + StartOffset: recorded.GetFromOffset(), + }) + if err != nil { + return err + } + if w.StreamCursors == nil { + w.StreamCursors = make(chasm.Map[string, *stream.Cursor]) + } + field = chasm.NewComponentField(mctx, cursor) + w.StreamCursors[recorded.GetStreamId()] = field + } + field.Get(mctx).Restore(mctx, recorded.GetToOffset()) + } + return nil +} + +// InheritStreamsOnReset finishes the cursors a reset run rebuilt from its +// history with what the events could not say, read from the run it was reset +// from. +// +// Every subscription the base run held is carried. One the rebuild recreated +// keeps the position its events gave it; one the events never mentioned, made +// out of band and never delivered to before the reset point, starts where it +// started in the base run. A cursor on a stream in another execution keeps +// reading that stream; it is marked so and given the frontier the base run +// last knew. A cursor on a stream the base run owned gets a stream of this +// run's own, starting at the offset the cursor stands at: the ranges below it +// are in the base run's stream, which is where replay re-reads them, and +// everything from here on is this run's. Continuing the offset space is what +// keeps a range in this run's History unambiguous about which run holds it. +// +// Two things the reset run does not get. Records the base run's stream held +// but had not delivered, everything between the cursor and the base head, are +// not carried and are invisible to it: the reset run starts past them. And the +// ranges below the cursor stay in the base run's mutable state, so nothing +// here pins that run. Once namespace retention deletes it, the reset run's +// cold replay has nowhere to read those ranges from. +func (w *Workflow) InheritStreamsOnReset( + mctx chasm.MutableContext, + base *Workflow, + baseCtx chasm.Context, + limits stream.Limits, +) error { + if w.StreamCursors == nil { + w.StreamCursors = make(chasm.Map[string, *stream.Cursor]) + } + if err := w.carryBaseCursors(mctx, base, baseCtx); err != nil { + return err + } + + names := make([]string, 0, len(w.StreamCursors)) + for name := range w.StreamCursors { + names = append(names, name) + } + slices.Sort(names) + + for _, name := range names { + cursor := w.StreamCursors[name].Get(mctx) + if baseCursor := cursorNamed(base, baseCtx, name); baseCursor != nil && baseCursor.IsExternal() { + cursor.MarkExternal(mctx, baseCursor.KnownHead()) + continue + } + if err := w.ownStreamFrom(mctx, name, cursor.Offset(), limits); err != nil { + return err + } + } + return nil +} + +// cursorNamed returns a workflow's cursor by stream name, or nil when the +// workflow is absent or holds none by that name. +func cursorNamed(w *Workflow, ctx chasm.Context, name string) *stream.Cursor { + if w == nil { + return nil + } + field, ok := w.StreamCursors[name] + if !ok { + return nil + } + return field.Get(ctx) +} + +// carryBaseCursors creates a cursor for every subscription the base run held +// that the rebuild did not recreate, starting where it started in the base run. +func (w *Workflow) carryBaseCursors( + mctx chasm.MutableContext, + base *Workflow, + baseCtx chasm.Context, +) error { + if base == nil { + return nil + } + for name, field := range base.StreamCursors { + if _, ok := w.StreamCursors[name]; ok { + continue + } + baseCursor := field.Get(baseCtx) + cursor, err := stream.NewCursor(mctx, stream.NewCursorRequest{ + StreamID: baseCursor.StreamID(), + StartOffset: baseCursor.StartOffset(), + }) + if err != nil { + return err + } + w.StreamCursors[name] = chasm.NewComponentField(mctx, cursor) + } + return nil +} + +// ownStreamFrom gives this run a stream of its own by that name, beginning at +// the given offset and pinned by this run's consumer there, unless it has one. +func (w *Workflow) ownStreamFrom( + mctx chasm.MutableContext, + name string, + offset int64, + limits stream.Limits, +) error { + if _, ok := w.Streams[name]; ok { + return nil + } + if w.Streams == nil { + w.Streams = make(chasm.Map[string, *stream.Stream]) + } + created, err := stream.NewStream(mctx, stream.NewStreamRequest{ + Attached: true, + StartOffset: offset, + Budget: &streamlib.StreamBudget{ + MaxItems: int64(limits.OwnedStreamMaxItems), + MaxBytes: int64(limits.OwnedStreamMaxBytes), + }, + }) + if err != nil { + return err + } + key := mctx.ExecutionKey() + if _, err := created.RegisterConsumer(mctx, stream.ConsumerRegistration{ + ConsumerID: streamConsumerID(name), + WorkflowID: key.BusinessID, + RunID: key.RunID, + Offset: offset, + MaxConsumers: limits.MaxConsumersPerStream, + }); err != nil { + return err + } + w.Streams[name] = chasm.NewComponentField(mctx, created) + 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/recordworkflowtaskstarted/stream_routing_test.go b/service/history/api/recordworkflowtaskstarted/stream_routing_test.go index 28589f344a5..4d91e339a6b 100644 --- a/service/history/api/recordworkflowtaskstarted/stream_routing_test.go +++ b/service/history/api/recordworkflowtaskstarted/stream_routing_test.go @@ -67,7 +67,7 @@ func TestExternalStreamLiveAndReplayUseRoutedPayloadRead(t *testing.T) { if replay { window, err = readRecordedRange(ctx, definition.NewWorkflowKey("namespace-id", "consumer", "consumer-run"), - streamOrigin{external: true}, "remote-source", 4, 6) + streamOrigin{external: true}, "consumer-run", "remote-source", 4, 6) } else { window, err = readWindowFor(ctx, nil, nil, "namespace-id", "input", true, "remote-source", 4, 6) } @@ -158,6 +158,6 @@ func TestOwnedReplayPinsConsumerRun(t *testing.T) { }) _, err := readRecordedRange(chasm.NewEngineContext(context.Background(), engine), definition.NewWorkflowKey("ns", "consumer", "consumer-run"), - streamOrigin{name: "owned"}, "owned", 4, 6) + streamOrigin{name: "owned"}, "consumer-run", "owned", 4, 6) require.NoError(t, err) } diff --git a/service/history/api/recordworkflowtaskstarted/stream_slices.go b/service/history/api/recordworkflowtaskstarted/stream_slices.go index 5adc57dcbb2..08bfd23a533 100644 --- a/service/history/api/recordworkflowtaskstarted/stream_slices.go +++ b/service/history/api/recordworkflowtaskstarted/stream_slices.go @@ -402,13 +402,16 @@ type ownedRange struct { // readRecordedRange re-supplies a range a completed task recorded. // -// It runs after the execution lock is released, so the consumer's own component -// is read back through the engine. Standalone sources use the routed service -// client, since their shards can belong to a different history host. +// It runs after the execution lock is released, so an owned stream is read +// back through the engine, from the run that holds it: the consumer itself, or +// the run the consumer was reset from for a range recorded before the reset. +// Standalone sources use the routed service client, since their shards can +// belong to a different history host. func readRecordedRange( ctx context.Context, consumer definition.WorkflowKey, origin streamOrigin, + ownerRunID string, streamID string, from, to int64, ) (stream.Window, error) { @@ -420,7 +423,7 @@ func readRecordedRange( chasm.NewComponentRef[*chasmworkflow.Workflow](chasm.ExecutionKey{ NamespaceID: consumer.NamespaceID, BusinessID: consumer.WorkflowID, - RunID: consumer.RunID, + RunID: ownerRunID, }), func(wf *chasmworkflow.Workflow, cctx chasm.Context, r ownedRange) (stream.Window, error) { s := wf.OwnedStream(cctx, r.name) @@ -447,6 +450,20 @@ type replaySupply struct { // can tell a complete re-supply from a short one. reached map[string]int64 slices []*streampb.StreamSlice + + // Ranges recorded since the last reset point, waiting to learn which run's + // stream holds them. A reset copies the base run's history into the new + // run, so the ranges before the reset point were consumed from the base + // run's owned streams, and the run they belong to is only named by the + // event that marks the reset point, after them in the history. Ranges + // after the last reset point belong to the consumer itself. + pending []recordedRange +} + +// recordedRange is one consumed range and the completion that recorded it. +type recordedRange struct { + eventID int64 + recorded *streampb.StreamRange } // attachReplaySlices re-supplies the payloads for ranges that earlier workflow @@ -553,6 +570,9 @@ func attachReplaySlices( } } + if err := supply.finish(); err != nil { + return err + } if err := supply.checkCoverage(); err != nil { return err } @@ -640,6 +660,9 @@ func ReplaySlicesForQuery( } token = next } + if err := supply.finish(); err != nil { + return nil, err + } if err := supply.checkCoverage(); err != nil { return nil, err } @@ -655,49 +678,75 @@ func AsRefusal(err error) error { return err } -// collect re-reads every range the events on one page recorded. +// collect gathers every range the events on one page recorded and re-reads the +// ones whose run it can name. A reset point names the run for everything +// gathered before it. func (s *replaySupply) collect(events []*historypb.HistoryEvent) error { for _, event := range events { + if attrs := event.GetWorkflowTaskFailedEventAttributes(); attrs != nil && + attrs.GetCause() == enumspb.WORKFLOW_TASK_FAILED_CAUSE_RESET_WORKFLOW && + attrs.GetBaseRunId() != "" { + if err := s.flush(attrs.GetBaseRunId()); err != nil { + return err + } + continue + } for _, recorded := range event.GetWorkflowTaskCompletedEventAttributes().GetConsumedStreamRanges() { - address, ok := s.addresses[recorded.GetStreamId()] - if !ok { + if _, ok := s.addresses[recorded.GetStreamId()]; !ok { // A subscription the workflow has since dropped. The range is // still part of its history, but nothing is consuming it now. continue } + s.pending = append(s.pending, recordedRange{eventID: event.GetEventId(), recorded: recorded}) + } + } + return nil +} - records, ownerRunID, err := s.recordsFor(address, recorded) - if err != nil { - return err - } +// finish re-reads the ranges recorded after the last reset point, which the +// consumer's own streams hold. Called once every page has been walked. +func (s *replaySupply) finish() error { + return s.flush(s.consumer.RunID) +} - if to := recorded.GetToOffset(); to > s.reached[recorded.GetStreamId()] { - s.reached[recorded.GetStreamId()] = to - } +// flush re-reads every range gathered since the last reset point from the run +// named for it, in the order the tasks recorded them. +func (s *replaySupply) flush(ownerRunID string) error { + for _, r := range s.pending { + address := s.addresses[r.recorded.GetStreamId()] + records, runID, err := s.recordsFor(address, ownerRunID, r.recorded) + if err != nil { + return err + } - // Attached even when empty: the task observed nothing, and replay - // has to reproduce that rather than infer it from an absence. - s.slices = append(s.slices, &streampb.StreamSlice{ - StreamId: recorded.GetStreamId(), - RunId: ownerRunID, - FromOffset: recorded.GetFromOffset(), - ToOffset: recorded.GetToOffset(), - Records: records, - WorkflowTaskCompletedEventId: event.GetEventId(), - }) + if to := r.recorded.GetToOffset(); to > s.reached[r.recorded.GetStreamId()] { + s.reached[r.recorded.GetStreamId()] = to } + + // Attached even when empty: the task observed nothing, and replay + // has to reproduce that rather than infer it from an absence. + s.slices = append(s.slices, &streampb.StreamSlice{ + StreamId: r.recorded.GetStreamId(), + RunId: runID, + FromOffset: r.recorded.GetFromOffset(), + ToOffset: r.recorded.GetToOffset(), + Records: records, + WorkflowTaskCompletedEventId: r.eventID, + }) } + s.pending = nil return nil } // recordsFor re-reads one recorded range, or returns nothing for a range that // recorded an empty observation. The run id is the execution holding the -// stream, which for an owned stream is the consumer itself. +// stream: for an owned stream the run named for the range, for a standalone +// one whatever run the read reports. func (s *replaySupply) recordsFor( address streamOrigin, + ownerRunID string, recorded *streampb.StreamRange, ) ([]*streampb.StreamRecord, string, error) { - ownerRunID := s.consumer.RunID if recorded.GetToOffset() <= recorded.GetFromOffset() { if address.external { ownerRunID = "" @@ -705,7 +754,7 @@ func (s *replaySupply) recordsFor( return nil, ownerRunID, nil } - w, err := readRecordedRange(s.ctx, s.consumer, address, + w, err := readRecordedRange(s.ctx, s.consumer, address, ownerRunID, recorded.GetStreamId(), recorded.GetFromOffset(), recorded.GetToOffset()) if err != nil { // Truncation and deletion are the reachable causes, and neither is diff --git a/service/history/api/recordworkflowtaskstarted/stream_slices_test.go b/service/history/api/recordworkflowtaskstarted/stream_slices_test.go new file mode 100644 index 00000000000..f0e63b2a107 --- /dev/null +++ b/service/history/api/recordworkflowtaskstarted/stream_slices_test.go @@ -0,0 +1,125 @@ +package recordworkflowtaskstarted + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + 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/common/definition" + "go.uber.org/mock/gomock" +) + +func completedWithRange(eventID int64, from, to int64) *historypb.HistoryEvent { + return &historypb.HistoryEvent{ + EventId: eventID, + EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED, + Attributes: &historypb.HistoryEvent_WorkflowTaskCompletedEventAttributes{ + WorkflowTaskCompletedEventAttributes: &historypb.WorkflowTaskCompletedEventAttributes{ + ConsumedStreamRanges: []*streampb.StreamRange{ + {StreamId: "output", FromOffset: from, ToOffset: to}, + }, + }, + }, + } +} + +func resetPoint(eventID int64, baseRunID string) *historypb.HistoryEvent { + return &historypb.HistoryEvent{ + EventId: eventID, + EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED, + Attributes: &historypb.HistoryEvent_WorkflowTaskFailedEventAttributes{ + WorkflowTaskFailedEventAttributes: &historypb.WorkflowTaskFailedEventAttributes{ + Cause: enumspb.WORKFLOW_TASK_FAILED_CAUSE_RESET_WORKFLOW, + BaseRunId: baseRunID, + NewRunId: "reset-run", + }, + }, + } +} + +// A reset copies the base run's history into the new run. The ranges recorded +// before the reset point were consumed from the base run's own stream, so +// replay has to read them from that run and say so on the slice, while the +// ranges recorded after it are the reset run's own. A reset of a reset run +// leaves two such points, one per run in the chain. +func TestReplayReadsRangesBeforeAResetFromTheRunThatHoldsThem(t *testing.T) { + var readFrom []string + engine := chasm.NewMockEngine(gomock.NewController(t)) + engine.EXPECT().ReadComponent(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func( + _ context.Context, + ref chasm.ComponentRef, + _ func(chasm.Context, chasm.Component) error, + _ ...chasm.TransitionOption, + ) error { + require.Equal(t, "consumer", ref.BusinessID) + readFrom = append(readFrom, ref.RunID) + return nil + }).AnyTimes() + + supply := &replaySupply{ + ctx: chasm.NewEngineContext(context.Background(), engine), + consumer: definition.NewWorkflowKey("ns", "consumer", "reset-run"), + addresses: map[string]streamOrigin{"output": {name: "output"}}, + reached: map[string]int64{}, + } + + // First page: the oldest run's era ends with the first reset point. + require.NoError(t, supply.collect([]*historypb.HistoryEvent{ + completedWithRange(4, 0, 2), + completedWithRange(8, 2, 2), + resetPoint(10, "first-run"), + })) + // Second page: the middle run's era, then the reset that made this run. + require.NoError(t, supply.collect([]*historypb.HistoryEvent{ + completedWithRange(14, 2, 5), + resetPoint(16, "second-run"), + completedWithRange(20, 5, 5), + completedWithRange(24, 5, 7), + })) + require.NoError(t, supply.finish()) + require.NoError(t, supply.checkCoverage()) + + require.Equal(t, []string{"first-run", "second-run", "reset-run"}, readFrom, + "every non-empty range is read from the run whose era recorded it") + + runs := make([]string, 0, len(supply.slices)) + events := make([]int64, 0, len(supply.slices)) + for _, slice := range supply.slices { + runs = append(runs, slice.GetRunId()) + events = append(events, slice.GetWorkflowTaskCompletedEventId()) + } + require.Equal(t, []int64{4, 8, 14, 20, 24}, events, "slices keep the order the tasks ran in") + require.Equal(t, []string{"first-run", "first-run", "second-run", "reset-run", "reset-run"}, runs, + "an empty range names its era's run too") +} + +// Without a reset point every owned range is the consumer's own. +func TestReplayWithoutAResetReadsFromTheConsumer(t *testing.T) { + engine := chasm.NewMockEngine(gomock.NewController(t)) + engine.EXPECT().ReadComponent(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func( + _ context.Context, + ref chasm.ComponentRef, + _ func(chasm.Context, chasm.Component) error, + _ ...chasm.TransitionOption, + ) error { + require.Equal(t, "consumer-run", ref.RunID) + return nil + }).Times(1) + + supply := &replaySupply{ + ctx: chasm.NewEngineContext(context.Background(), engine), + consumer: definition.NewWorkflowKey("ns", "consumer", "consumer-run"), + addresses: map[string]streamOrigin{"output": {name: "output"}}, + reached: map[string]int64{}, + } + require.NoError(t, supply.collect([]*historypb.HistoryEvent{completedWithRange(4, 0, 3)})) + require.NoError(t, supply.finish()) + require.Len(t, supply.slices, 1) + require.Equal(t, "consumer-run", supply.slices[0].GetRunId()) +} diff --git a/service/history/ndc/workflow_resetter.go b/service/history/ndc/workflow_resetter.go index 1b95952a46b..bc5ea58dcc8 100644 --- a/service/history/ndc/workflow_resetter.go +++ b/service/history/ndc/workflow_resetter.go @@ -255,6 +255,10 @@ func (r *workflowResetterImpl) ResetWorkflow( defer func() { resetWorkflow.GetReleaseFn()(retError) }() resetMS := resetWorkflow.GetMutableState() + err = r.inheritStreams(ctx, namespaceEntry, baseWorkflow.GetMutableState(), resetMS) + if err != nil { + return err + } if err := reapplyEventsFn(ctx, resetMS); err != nil { return err } @@ -288,6 +292,45 @@ func (r *workflowResetterImpl) ResetWorkflow( return nil } +// inheritStreams completes the stream cursors the rebuild recreated from the +// reset run's events with what only the base run's state can say: which +// streams live in other executions, and the frontier those last pushed. A +// cursor on a stream the base run owned gets a stream of the reset run's own, +// continuing the offset space from where the cursor stands. +func (r *workflowResetterImpl) inheritStreams( + ctx context.Context, + namespaceEntry *namespace.Namespace, + baseMS historyi.MutableState, + resetMS historyi.MutableState, +) error { + if !resetMS.HasChasmWorkflowComponent() { + return nil + } + // Read-only first: a reset run with no subscription should not pay a node + // in its first transaction for a check that finds nothing. + readOnly, _, err := resetMS.ChasmWorkflowComponentReadOnly(ctx) + if err != nil { + return err + } + if len(readOnly.StreamCursors) == 0 { + return nil + } + + var base *chasmworkflow.Workflow + var baseCtx chasm.Context + if baseMS.HasChasmWorkflowComponent() { + if base, baseCtx, err = baseMS.ChasmWorkflowComponentReadOnly(ctx); err != nil { + return err + } + } + reset, resetCtx, err := resetMS.ChasmWorkflowComponent(ctx) + if err != nil { + return err + } + limits := r.shardContext.GetConfig().Stream.LimitsFor(namespaceEntry.Name().String()) + return reset.InheritStreamsOnReset(resetCtx, base, baseCtx, limits) +} + func (r *workflowResetterImpl) prepareResetWorkflow( ctx context.Context, namespaceID namespace.ID, diff --git a/service/history/workflow/mutable_state_rebuilder.go b/service/history/workflow/mutable_state_rebuilder.go index 6e1fb1cc892..55edf9745c4 100644 --- a/service/history/workflow/mutable_state_rebuilder.go +++ b/service/history/workflow/mutable_state_rebuilder.go @@ -261,6 +261,9 @@ func (b *MutableStateRebuilderImpl) applyEvents( ); err != nil { return nil, err } + if err := b.applyConsumedStreamRanges(ctx, event); err != nil { + return nil, err + } case enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT: if err := b.mutableState.ApplyWorkflowTaskTimedOutEvent( @@ -732,6 +735,26 @@ func (b *MutableStateRebuilderImpl) applyStateMachineEvent( return b.applyHSMEvent(event) } +// applyConsumedStreamRanges moves the stream cursors to where the completed +// event says they stood. The cursors are CHASM state and the ranges ride the +// event, so a run rebuilt from its history, a reset run above all, ends up with +// its cursors exactly where the events put them. +func (b *MutableStateRebuilderImpl) applyConsumedStreamRanges( + ctx context.Context, + event *historypb.HistoryEvent, +) error { + ranges := event.GetWorkflowTaskCompletedEventAttributes().GetConsumedStreamRanges() + if len(ranges) == 0 || !b.mutableState.ChasmEnabled() { + return nil + } + b.mutableState.EnsureChasmWorkflowComponent(ctx) + wf, chasmCtx, err := b.mutableState.ChasmWorkflowComponent(ctx) + if err != nil { + return err + } + return wf.ApplyConsumedStreamRanges(chasmCtx, ranges) +} + // applyHSMEvent applies an event to the HSM tree func (b *MutableStateRebuilderImpl) applyHSMEvent(event *historypb.HistoryEvent) error { def, ok := b.shard.StateMachineRegistry().EventDefinition(event.GetEventType()) diff --git a/tests/stream_reset_test.go b/tests/stream_reset_test.go new file mode 100644 index 00000000000..fcb770cafa9 --- /dev/null +++ b/tests/stream_reset_test.go @@ -0,0 +1,394 @@ +package tests + +import ( + "testing" + + "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" + "go.temporal.io/api/workflowservice/v1" + 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" +) + +// nthCompletedEvent returns the id of the n-th WorkflowTaskCompleted event, +// counting from one. +func nthCompletedEvent(t *testing.T, events []*historypb.HistoryEvent, n int) int64 { + t.Helper() + seen := 0 + for _, e := range events { + if e.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED { + seen++ + if seen == n { + return e.GetEventId() + } + } + } + t.Fatalf("history has %d completed tasks, wanted %d", seen, n) + return 0 +} + +func resetTo( + t *testing.T, s *streamTestEnv, execution *commonpb.WorkflowExecution, eventID int64, +) string { + t.Helper() + resp, err := s.env.FrontendClient().ResetWorkflowExecution(s.ctx(), + &workflowservice.ResetWorkflowExecutionRequest{ + Namespace: s.ns, + WorkflowExecution: execution, + Reason: "stream reset test", + WorkflowTaskFinishEventId: eventID, + RequestId: uuid.NewString(), + }) + require.NoError(t, err) + return resp.GetRunId() +} + +// A reset copies the base run's history into a new run, and that history +// records ranges the base run consumed from a stream it owned. The reset run +// replays them from the base run's stream, with each slice naming that run, +// and from the reset point on it consumes and publishes on a stream of its own +// that continues the offset space where the inherited cursor stood. +func TestResetReplaysTheBaseRunsRangesAndGoesOnWithItsOwnStream(t *testing.T) { + // Dedicated, because the cold replay at the end evicts the cached + // workflow context through CloseShard. + env := testcore.NewEnv(t, testcore.WithDedicatedCluster()) + s := newStreamTestEnvFrom(t, env) + execution, tq := startConsumer(t, s, "stream-wf-reset-") + + var delivered [][]*streampb.StreamSlice + var commands [][]*commandpb.Command + //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( + resp *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + delivered = append(delivered, resp.GetStreamSlices()) + next := commands[0] + commands = commands[1:] + return next, nil + }, + Logger: env.Logger, + T: t, + } + runTask := func(cmds []*commandpb.Command) []*streampb.StreamSlice { + t.Helper() + commands = append(commands, cmds) + _, err := poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + return delivered[len(delivered)-1] + } + + // Base run: publish two, subscribe, consume them, publish a third, consume it. + runTask(publishCommand("before-1", "before-2")) + _, err := s.client.SubscribeWorkflow(s.ctx(), &streamlib.SubscribeWorkflowRequest{ + FrontendRequest: &streamlib.SubscribeWorkflowInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + StreamName: chasmworkflow.DefaultStreamName, StartOffset: 0, + }, + }) + require.NoError(t, err) + consumed := currentSlice(t, runTask(nil)) + require.Equal(t, []string{"before-1", "before-2"}, apiBodies(consumed.GetRecords())) + + signalWorkflow(t, s, execution.GetWorkflowId(), execution.GetRunId()) + runTask(publishCommand("before-3")) + third := currentSlice(t, runTask(nil)) + require.Equal(t, []string{"before-3"}, apiBodies(third.GetRecords())) + + baseEvents := env.GetHistory(s.ns, execution) + consumedAt := completedEventWithCursors(t, baseEvents) + + // Reset to the third task: the reset run keeps the first two completions, + // so its cursor stands at offset 2 and nothing of the third publish is its. + resetRunID := resetTo(t, s, execution, nthCompletedEvent(t, baseEvents, 3)) + resetRun := &commonpb.WorkflowExecution{WorkflowId: execution.GetWorkflowId(), RunId: resetRunID} + + // The reset run's first task replays the range recorded before the reset + // point from the base run's stream, and consumes nothing new of its own. + first := runTask(nil) + replayed := sliceForEvent(first, consumedAt) + require.NotNil(t, replayed, "the copied completion at event %d is re-supplied", consumedAt) + require.Equal(t, execution.GetRunId(), replayed.GetRunId(), + "a range recorded before the reset point is read from the base run") + require.Equal(t, int64(0), replayed.GetFromOffset()) + require.Equal(t, int64(2), replayed.GetToOffset()) + require.Equal(t, []string{"before-1", "before-2"}, apiBodies(replayed.GetRecords())) + live := currentSlice(t, first) + require.Equal(t, resetRunID, live.GetRunId(), + "from the reset point on the stream is the reset run's") + require.Equal(t, int64(2), live.GetFromOffset(), "the inherited cursor keeps its position") + require.Equal(t, int64(2), live.GetToOffset()) + + // The reset run publishes and consumes on its own stream, whose offsets + // continue from where the cursor stood. + signalWorkflow(t, s, execution.GetWorkflowId(), resetRunID) + runTask(publishCommand("after-1")) + own := currentSlice(t, runTask(nil)) + require.Equal(t, resetRunID, own.GetRunId()) + require.Equal(t, int64(2), own.GetFromOffset()) + require.Equal(t, int64(3), own.GetToOffset()) + require.Equal(t, []string{"after-1"}, apiBodies(own.GetRecords())) + + // Seen from outside: the base run still holds every record its history + // refers to, and the reset run's stream starts at the inherited offset. + basePoll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + OwnerRunId: execution.GetRunId(), FromOffset: 0, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"before-1", "before-2", "before-3"}, + bodies(basePoll.GetFrontendResponse().GetRecords())) + + resetPoll, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + OwnerRunId: resetRunID, FromOffset: 2, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"after-1"}, bodies(resetPoll.GetFrontendResponse().GetRecords())) + require.Equal(t, []int64{2}, offsets(resetPoll.GetFrontendResponse().GetRecords())) + + _, err = s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + OwnerRunId: resetRunID, FromOffset: 0, + }, + }) + require.Equal(t, codes.FailedPrecondition, status.Code(err), + "a reader following the chain into the reset run has to start at the offset it begins at") + + // A cold replay of the reset run re-supplies both eras from the runs that + // hold them, tagged with the completions that recorded them. + env.CloseShard(env.NamespaceID().String(), execution.GetWorkflowId()) + signalWorkflow(t, s, execution.GetWorkflowId(), resetRunID) + cold := runTask(nil) + fromBase := sliceForEvent(cold, consumedAt) + require.NotNil(t, fromBase) + require.Equal(t, execution.GetRunId(), fromBase.GetRunId()) + require.Equal(t, []string{"before-1", "before-2"}, apiBodies(fromBase.GetRecords())) + ownAt := completedEventWithCursorsAfter(t, env.GetHistory(s.ns, resetRun), consumedAt) + fromReset := sliceForEvent(cold, ownAt) + require.NotNil(t, fromReset) + require.Equal(t, resetRunID, fromReset.GetRunId()) + require.Equal(t, int64(2), fromReset.GetFromOffset()) + require.Equal(t, int64(3), fromReset.GetToOffset()) + require.Equal(t, []string{"after-1"}, apiBodies(fromReset.GetRecords())) + + // The base run names its successor, which is how a client following the + // chain finds the reset run. + described, err := env.FrontendClient().DescribeWorkflowExecution(s.ctx(), + &workflowservice.DescribeWorkflowExecutionRequest{Namespace: s.ns, Execution: execution}) + require.NoError(t, err) + require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, + described.GetWorkflowExecutionInfo().GetStatus()) + require.Equal(t, resetRunID, described.GetWorkflowExtendedInfo().GetResetRunId()) +} + +// completedEventWithCursorsAfter returns the id of the first WorkflowTaskCompleted +// event after the given one that recorded a non-empty consumed range. +func completedEventWithCursorsAfter( + t *testing.T, events []*historypb.HistoryEvent, after int64, +) int64 { + t.Helper() + for _, e := range events { + if e.GetEventId() <= after { + continue + } + for _, c := range e.GetWorkflowTaskCompletedEventAttributes().GetConsumedStreamRanges() { + if c.GetToOffset() > c.GetFromOffset() { + return e.GetEventId() + } + } + } + t.Fatal("no completed event after the given one recorded a consumed range") + return 0 +} + +// A reset terminates the base run. Its pin on a stream in another execution is +// released the way a closed run's is, when the stream next tells its consumers +// the frontier moved and finds the run over, and the reset run, which carries +// the subscription on, takes the pin over with the floor its cursor began at. +func TestResetHandsTheExternalPinToTheResetRun(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + + streamID := "reset-shared-stream-" + uuid.NewString() + s.create(s.ctx(), t, streamID) + execution, tq := startConsumer(t, s, "stream-wf-reset-ext-") + + var delivered [][]*streampb.StreamSlice + //nolint:staticcheck // SA1019: consistent with the other stream tests. + poller := &testcore.TaskPoller{ + Client: env.FrontendClient(), + Namespace: s.ns, + TaskQueue: tq, + Identity: "tester", + WorkflowTaskHandler: func( + resp *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + delivered = append(delivered, resp.GetStreamSlices()) + return nil, nil + }, + Logger: env.Logger, + T: t, + } + runTask := func() []*streampb.StreamSlice { + t.Helper() + _, err := poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + return delivered[len(delivered)-1] + } + + runTask() + _, err := s.client.SubscribeWorkflow(s.ctx(), &streamlib.SubscribeWorkflowRequest{ + FrontendRequest: &streamlib.SubscribeWorkflowInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), StreamId: streamID, StartOffset: 0, + }, + }) + require.NoError(t, err) + _, err = s.add(s.ctx(), t, streamID, + &streamlib.AddMessagesInput{Records: streamMsgs("t", "one", "two")}) + require.NoError(t, err) + consumed := currentSlice(t, runTask()) + require.Equal(t, int64(2), consumed.GetToOffset()) + + signalWorkflow(t, s, execution.GetWorkflowId(), execution.GetRunId()) + runTask() + baseEvents := env.GetHistory(s.ns, execution) + consumedAt := completedEventWithCursors(t, baseEvents) + + resetRunID := resetTo(t, s, execution, nthCompletedEvent(t, baseEvents, 3)) + + // The reset run replays the base run's range from the standalone stream + // and stands at the same offset. + first := runTask() + replayed := sliceForEvent(first, consumedAt) + require.NotNil(t, replayed) + require.Equal(t, []string{"one", "two"}, apiBodies(replayed.GetRecords())) + require.Equal(t, int64(2), currentSlice(t, first).GetFromOffset()) + + // The pin is still the base run's until the stream next notifies. + before := describeStream(t, s, streamID) + require.Len(t, before.GetConsumers(), 1) + for _, consumer := range before.GetConsumers() { + require.Equal(t, execution.GetRunId(), consumer.GetRunId()) + } + + // An append pushes the frontier at the pinned run, finds it terminated, and + // hands the pin to the run that now carries the subscription. + _, err = s.add(s.ctx(), t, streamID, + &streamlib.AddMessagesInput{Records: streamMsgs("t", "three")}) + require.NoError(t, err) + afterReset := currentSlice(t, runTask()) + require.Equal(t, int64(2), afterReset.GetFromOffset(), + "the reset run resumes where the base stopped") + require.Equal(t, int64(3), afterReset.GetToOffset()) + require.Equal(t, []string{"three"}, apiBodies(afterReset.GetRecords())) + + after := describeStream(t, s, streamID) + require.Len(t, after.GetConsumers(), 1, "one pin, keyed to the run that consumes") + for _, consumer := range after.GetConsumers() { + require.Equal(t, resetRunID, consumer.GetRunId()) + require.Equal(t, int64(0), consumer.GetReplayFloor(), + "the floor is where the inherited subscription began, which its replay depends on") + } +} + +// A workflow that only publishes gets no cursor, so a reset gives its run a +// stream that starts over at offset zero while the copied history still +// carries the base run's appended events at the offsets it wrote. +// +// Characterized rather than changed: whether the publish side should inherit +// the offset space the way the consume side does is a design decision, and +// until it is made this is what a reader following the workflow id sees. +func TestResetOfAPublisherRestartsItsStreamAtZero(t *testing.T) { + env := testcore.NewEnv(t) + s := newStreamTestEnvFrom(t, env) + execution, tq := startConsumer(t, s, "stream-wf-reset-publisher-") + + var commands [][]*commandpb.Command + //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) { + next := commands[0] + commands = commands[1:] + return next, nil + }, + Logger: env.Logger, + T: t, + } + runTask := func(cmds []*commandpb.Command) { + t.Helper() + commands = append(commands, cmds) + _, err := poller.PollAndProcessWorkflowTask() + require.NoError(t, err) + } + + // Publishes only, so nothing subscribes and no cursor is ever created. + runTask(publishCommand("a", "b")) + signalWorkflow(t, s, execution.GetWorkflowId(), execution.GetRunId()) + runTask(publishCommand("c")) + + baseEvents := env.GetHistory(s.ns, execution) + var appendedOffsets []int64 + for _, e := range baseEvents { + if attrs := e.GetWorkflowStreamRecordsAppendedEventAttributes(); attrs != nil { + appendedOffsets = append(appendedOffsets, attrs.GetFromOffset()) + } + } + require.Equal(t, []int64{0, 2}, appendedOffsets) + + // Reset to the second completion, so the copied history keeps the first + // publish's event at offset 0 and drops the second's. + resetRunID := resetTo(t, s, execution, nthCompletedEvent(t, baseEvents, 2)) + resetRun := &commonpb.WorkflowExecution{ + WorkflowId: execution.GetWorkflowId(), RunId: resetRunID, + } + + // The reset run publishes again. Its stream is a new one, so the offsets + // restart even though the copied history already names offset 0. + runTask(publishCommand("d")) + + read, err := s.client.PollWorkflowMessages(s.ctx(), &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: s.ns, WorkflowId: execution.GetWorkflowId(), + OwnerRunId: resetRunID, FromOffset: 0, MaxMessages: 10, + }, + }) + require.NoError(t, err) + records := read.GetFrontendResponse().GetRecords() + require.Equal(t, []string{"d"}, bodies(records)) + require.Equal(t, int64(0), records[0].GetOffset(), + "the publish side restarts its offset space, unlike the consume side") + + resetEvents := env.GetHistory(s.ns, resetRun) + var resetOffsets []int64 + for _, e := range resetEvents { + if attrs := e.GetWorkflowStreamRecordsAppendedEventAttributes(); attrs != nil { + resetOffsets = append(resetOffsets, attrs.GetFromOffset()) + } + } + require.Equal(t, []int64{0, 0}, resetOffsets, + "the copied event and the new one name the same offset for different records") +} diff --git a/tests/xdc/stream_failover_test.go b/tests/xdc/stream_failover_test.go new file mode 100644 index 00000000000..dfb23ec3762 --- /dev/null +++ b/tests/xdc/stream_failover_test.go @@ -0,0 +1,410 @@ +package xdc + +import ( + "context" + "slices" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" + 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" + "go.temporal.io/server/api/adminservice/v1" + streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/tests/testcore" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/durationpb" +) + +// streamFailoverBase drives one workflow that owns a stream with records and +// an active subscription on the active cluster, so each suite below can fail +// the namespace over and look at what the standby has. +type streamFailoverBase struct { + xdcBaseSuite +} + +// StreamFailoverSuite runs with state-based replication, the mode that carries +// a CHASM tree between clusters. +type StreamFailoverSuite struct { + streamFailoverBase +} + +// StreamEventReplicationSuite runs with event-based replication, where only +// what History events carry reaches the standby. +type StreamEventReplicationSuite struct { + streamFailoverBase +} + +func TestStreamFailoverSuite(t *testing.T) { + t.Parallel() + s := &StreamFailoverSuite{} + s.enableTransitionHistory = true + suite.Run(t, s) +} + +func TestStreamEventReplicationSuite(t *testing.T) { + t.Parallel() + s := &StreamEventReplicationSuite{} + s.enableTransitionHistory = false + suite.Run(t, s) +} + +func (s *streamFailoverBase) SetupSuite() { + s.dynamicConfigOverrides = map[dynamicconfig.Key]any{ + dynamicconfig.EnableChasm.Key(): true, + dynamicconfig.TransferProcessorMaxPollInterval.Key(): 1 * time.Second, + } + s.setupSuite() +} + +func (s *streamFailoverBase) SetupTest() { + s.setupTest() +} + +func (s *streamFailoverBase) TearDownSuite() { + s.tearDownSuite() +} + +func (s *streamFailoverBase) streamClient(clusterIndex int) streamlib.StreamServiceClient { + conn, err := grpc.NewClient(s.clusters[clusterIndex].Host().FrontendGRPCAddress(), + grpc.WithTransportCredentials(insecure.NewCredentials())) + s.NoError(err) + s.T().Cleanup(func() { _ = conn.Close() }) + return streamlib.NewStreamServiceClient(conn) +} + +// streamNodesOn lists the CHASM node paths of the execution on one cluster that +// belong to its streams and cursors, sorted, or nil when the execution is not +// there. A stream a workflow owns is `Streams#`, its records are data +// nodes under `Streams$$Batches#`, and the subscription is +// `StreamCursors#`. +func (s *streamFailoverBase) streamNodesOn( + ctx context.Context, clusterIndex int, ns string, execution *commonpb.WorkflowExecution, +) []string { + resp, err := s.clusters[clusterIndex].AdminClient().DescribeMutableState(ctx, + &adminservice.DescribeMutableStateRequest{Namespace: ns, Execution: execution}) + if err != nil { + return nil + } + var paths []string + for path := range resp.GetDatabaseMutableState().GetChasmNodes() { + if strings.Contains(path, "Stream") { + paths = append(paths, path) + } + } + slices.Sort(paths) + return paths +} + +// historyOn reads a workflow's history from one cluster, for assertions about +// what a task there did rather than about what replicated. +func (s *streamFailoverBase) historyOn( + ctx context.Context, clusterIndex int, ns string, execution *commonpb.WorkflowExecution, +) []*historypb.HistoryEvent { + var events []*historypb.HistoryEvent + var token []byte + for { + resp, err := s.clusters[clusterIndex].FrontendClient().GetWorkflowExecutionHistory(ctx, + &workflowservice.GetWorkflowExecutionHistoryRequest{ + Namespace: ns, Execution: execution, NextPageToken: token, + }) + if err != nil { + return events + } + events = append(events, resp.GetHistory().GetEvents()...) + token = resp.GetNextPageToken() + if len(token) == 0 { + return events + } + } +} + +func apiBodies(records []*streampb.StreamRecord) []string { + out := make([]string, 0, len(records)) + for _, r := range records { + out = append(out, string(r.GetBody().GetData())) + } + return out +} + +func storedBodies(records []*streamlib.StreamRecord) []string { + out := make([]string, 0, len(records)) + for _, r := range records { + out = append(out, string(r.GetBody().GetData())) + } + return out +} + +func publishCommand(bodies ...string) []*commandpb.Command { + records := make([]*streampb.StreamRecord, len(bodies)) + for i, b := range bodies { + records[i] = &streampb.StreamRecord{Body: &commonpb.Payload{Data: []byte(b)}} + } + return []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + Records: records, + }, + }, + }} +} + +// streamingWorkflow is one execution driven by a raw poller on whichever +// cluster is asked, handing each task the commands queued for it. +type streamingWorkflow struct { + s *streamFailoverBase + ns string + execution *commonpb.WorkflowExecution + tq *taskqueuepb.TaskQueue + + delivered [][]*streampb.StreamSlice + commands [][]*commandpb.Command +} + +func (w *streamingWorkflow) runTask( + clusterIndex int, cmds []*commandpb.Command, +) []*streampb.StreamSlice { + w.commands = append(w.commands, cmds) + //nolint:staticcheck // SA1019: only the deprecated poller can emit this command type. + poller := &testcore.TaskPoller{ + Client: w.s.clusters[clusterIndex].FrontendClient(), + Namespace: w.ns, + TaskQueue: w.tq, + Identity: "tester", + WorkflowTaskHandler: func( + resp *workflowservice.PollWorkflowTaskQueueResponse, + ) ([]*commandpb.Command, error) { + w.delivered = append(w.delivered, resp.GetStreamSlices()) + next := w.commands[0] + w.commands = w.commands[1:] + return next, nil + }, + Logger: w.s.logger, + T: w.s.T(), + } + _, err := poller.PollAndProcessWorkflowTask() + w.s.NoError(err) + return w.delivered[len(w.delivered)-1] +} + +func (w *streamingWorkflow) currentSlice(delivered []*streampb.StreamSlice) *streampb.StreamSlice { + for _, sl := range delivered { + if sl.GetWorkflowTaskCompletedEventId() == 0 { + return sl + } + } + w.s.FailNow("no slice for the current task") + return nil +} + +// pollOnce asks for a workflow task and throws away whatever comes back. Used +// where the point of the test is that the task cannot be started, so the +// ordinary poller, which expects a task and a history, has nothing to work +// with. +func (w *streamingWorkflow) pollOnce(ctx context.Context, clusterIndex int) { + pollCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + //nolint:errcheck // The call is expected to come back empty. + w.s.clusters[clusterIndex].FrontendClient().PollWorkflowTaskQueue(pollCtx, + &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: w.ns, TaskQueue: w.tq, Identity: "tester", + }) +} + +func (w *streamingWorkflow) signal(ctx context.Context, clusterIndex int) { + _, err := w.s.clusters[clusterIndex].FrontendClient().SignalWorkflowExecution(ctx, + &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: w.ns, WorkflowExecution: w.execution, + SignalName: "wake", Identity: "tester", RequestId: uuid.NewString(), + }) + w.s.NoError(err) +} + +// startStreamingWorkflow runs a workflow on the active cluster through one +// publish of three records, a subscription to that stream and the task that +// consumes them, then returns it with the stream nodes the active cluster holds. +func (s *streamFailoverBase) startStreamingWorkflow( + ctx context.Context, ns string, +) (*streamingWorkflow, []string) { + id := "stream-failover-" + uuid.NewString() + tq := &taskqueuepb.TaskQueue{Name: id + "-tq", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + we, err := s.clusters[0].FrontendClient().StartWorkflowExecution(ctx, + &workflowservice.StartWorkflowExecutionRequest{ + RequestId: uuid.NewString(), + Namespace: ns, + WorkflowId: id, + WorkflowType: &commonpb.WorkflowType{Name: "stream-consumer"}, + TaskQueue: tq, + WorkflowRunTimeout: durationpb.New(300 * time.Second), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: "tester", + }) + s.NoError(err) + w := &streamingWorkflow{ + s: s, + ns: ns, + execution: &commonpb.WorkflowExecution{WorkflowId: id, RunId: we.GetRunId()}, + tq: tq, + } + + w.runTask(0, publishCommand("a", "b", "c")) + _, err = s.streamClient(0).SubscribeWorkflow(ctx, &streamlib.SubscribeWorkflowRequest{ + FrontendRequest: &streamlib.SubscribeWorkflowInput{ + Namespace: ns, WorkflowId: id, StreamName: chasmworkflow.DefaultStreamName, StartOffset: 0, + }, + }) + s.NoError(err) + consumed := w.currentSlice(w.runTask(0, nil)) + s.Equal([]string{"a", "b", "c"}, apiBodies(consumed.GetRecords())) + + activeNodes := s.streamNodesOn(ctx, 0, ns, w.execution) + s.Equal([]string{ + "StreamCursors", "StreamCursors#output", + "Streams", "Streams#output", "Streams$output$Batches", "Streams$output$Batches#0", + }, activeNodes, "the active cluster holds the stream, its one batch and the cursor") + return w, activeNodes +} + +// With state-based replication the stream follows the namespace to the +// standby with its records, its consumer state and the subscription, and the +// workflow keeps consuming and publishing there. +func (s *StreamFailoverSuite) TestStreamFollowsTheNamespaceToTheStandby() { + ns := s.createGlobalNamespace() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + w, activeNodes := s.startStreamingWorkflow(ctx, ns) + + // The standby holds the same tree before the namespace moves. + await.Require(ctx, s.T(), func(t *await.T) { + require.Equal(t, activeNodes, s.streamNodesOn(ctx, 1, ns, w.execution), + "the standby's CHASM tree must carry the stream, its batches and the cursor") + }, replicationWaitTime, replicationCheckInterval) + + s.failover(ns, 0, s.clusters[1].ClusterName(), 2) + + // The records and the frontier are readable on the new active cluster. + described, err := s.streamClient(1).DescribeWorkflowStream(ctx, + &streamlib.DescribeWorkflowStreamRequest{ + FrontendRequest: &streamlib.DescribeWorkflowStreamInput{ + Namespace: ns, WorkflowId: w.execution.GetWorkflowId(), + }, + }) + s.NoError(err) + s.Equal(int64(3), described.GetFrontendResponse().GetState().GetHeadOffset()) + + polled, err := s.streamClient(1).PollWorkflowMessages(ctx, &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: ns, WorkflowId: w.execution.GetWorkflowId(), FromOffset: 0, + }, + }) + s.NoError(err) + s.Equal([]string{"a", "b", "c"}, storedBodies(polled.GetFrontendResponse().GetRecords())) + + // The workflow goes on consuming and publishing there. Its first task on + // the new cluster is a cold one, so the range it consumed before the + // failover is re-supplied from the replicated stream. + w.signal(ctx, 1) + afterFailover := w.runTask(1, publishCommand("d")) + var replayed *streampb.StreamSlice + for _, sl := range afterFailover { + if sl.GetWorkflowTaskCompletedEventId() != 0 && sl.GetToOffset() > sl.GetFromOffset() { + replayed = sl + } + } + s.NotNil(replayed, "the cold task on the new cluster re-supplies the consumed range") + s.Equal([]string{"a", "b", "c"}, apiBodies(replayed.GetRecords())) + s.Equal(int64(3), w.currentSlice(afterFailover).GetFromOffset()) + + next := w.currentSlice(w.runTask(1, nil)) + s.Equal(int64(3), next.GetFromOffset()) + s.Equal(int64(4), next.GetToOffset()) + s.Equal([]string{"d"}, apiBodies(next.GetRecords())) + + polled, err = s.streamClient(1).PollWorkflowMessages(ctx, &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: ns, WorkflowId: w.execution.GetWorkflowId(), FromOffset: 0, + }, + }) + s.NoError(err) + s.Equal([]string{"a", "b", "c", "d"}, storedBodies(polled.GetFrontendResponse().GetRecords())) +} + +// With event-based replication the standby rebuilds its state from History +// events, and the events carry the subscription and the offsets it consumed +// but never a record. So the standby ends up with the cursor and without the +// stream: the frontier reads as an unwritten stream, no record can be polled, +// and the workflow's first task there cannot find the stream its cursor names. +// Carrying the stream under this mode would take a replication task of its +// own for the component, the way state-based replication ships the CHASM +// tree; the appended event cannot carry it without putting payloads in History. +func (s *StreamEventReplicationSuite) TestStreamDoesNotReplicateUnderEventBasedReplicationYet() { + ns := s.createGlobalNamespace() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + w, _ := s.startStreamingWorkflow(ctx, ns) + + // Every replication task has been acknowledged by the standby, so what it + // holds now is what this mode carries. + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + require.Equal(t, []string{"StreamCursors", "StreamCursors#output"}, + s.streamNodesOn(ctx, 1, ns, w.execution), + "the standby rebuilds the cursor from the recorded ranges and nothing of the stream") + }, replicationWaitTime, replicationCheckInterval) + + s.failover(ns, 0, s.clusters[1].ClusterName(), 2) + + described, err := s.streamClient(1).DescribeWorkflowStream(ctx, + &streamlib.DescribeWorkflowStreamRequest{ + FrontendRequest: &streamlib.DescribeWorkflowStreamInput{ + Namespace: ns, WorkflowId: w.execution.GetWorkflowId(), + }, + }) + s.NoError(err) + s.Equal(int64(0), described.GetFrontendResponse().GetState().GetHeadOffset(), + "the new active cluster sees a stream nothing has written to") + + polled, err := s.streamClient(1).PollWorkflowMessages(ctx, &streamlib.PollWorkflowMessagesRequest{ + FrontendRequest: &streamlib.PollWorkflowMessagesInput{ + Namespace: ns, WorkflowId: w.execution.GetWorkflowId(), FromOffset: 0, + }, + }) + s.NoError(err) + s.Empty(polled.GetFrontendResponse().GetRecords(), "the records did not follow the namespace") + + // What the workflow itself sees, which is the part that matters to a user. + // Its cursor names a stream this cluster does not hold, so the task cannot + // be started and the failure is written into History with a cause an + // operator can read, rather than coming back from matching forever. + w.signal(ctx, 1) + await.Require(ctx, s.T(), func(t *await.T) { + // Each attempt is driven by a poll, because the failure happens when + // matching tries to start the task rather than when it is scheduled. + w.pollOnce(ctx, 1) + events := s.historyOn(ctx, 1, ns, w.execution) + var failed *historypb.HistoryEvent + for _, e := range events { + if e.GetWorkflowTaskFailedEventAttributes().GetCause() == + enumspb.WORKFLOW_TASK_FAILED_CAUSE_STREAM_RANGE_UNAVAILABLE { + failed = e + } + } + require.NotNil(t, failed, "the first task on the new cluster has to fail with a cause") + require.Contains(t, + failed.GetWorkflowTaskFailedEventAttributes().GetFailure().GetMessage(), + "neither owns nor subscribed to externally") + }, replicationWaitTime, replicationCheckInterval) +}