diff --git a/chasm/lib/nexusoperation/config.go b/chasm/lib/nexusoperation/config.go index abeb2b7196c..87d258a73df 100644 --- a/chasm/lib/nexusoperation/config.go +++ b/chasm/lib/nexusoperation/config.go @@ -60,6 +60,15 @@ one per operation onto the caller's next Workflow Task scheduled event. When fal delivery is refused with a 400, which tells the handler to stop sending progress.`, ) +var ProgressMinInterval = dynamicconfig.NewNamespaceDurationSetting( + "nexusoperation.progressMinInterval", + time.Second, + `Experimental. The shortest time between a Workflow Task that carried Nexus operation progress +and the next task that progress schedules for the same caller Workflow. Progress that arrives +sooner waits, and newer progress folds into it. A task scheduled for any other reason still +carries it. This bounds the History a long stream costs the caller.`, +) + var ChasmWorkflowOperationsRolloutPercent = dynamicconfig.NewNamespaceIntSetting( "nexusoperation.chasmWorkflowOperationsRolloutPercent", 0, diff --git a/chasm/lib/nexusoperation/operation.go b/chasm/lib/nexusoperation/operation.go index 03bb28ee9f7..252fe4c12c7 100644 --- a/chasm/lib/nexusoperation/operation.go +++ b/chasm/lib/nexusoperation/operation.go @@ -108,6 +108,20 @@ type Operation struct { // Callbacks holds completion callbacks to be invoked when this reaches a terminal state. // Keyed by completionCallbackID(requestID, index). Callbacks chasm.Map[string, *callback.Callback] + + // PendingProgress is the latest progress no Workflow Task has carried yet. + PendingProgress chasm.Field[*nexuspb.NexusOperationProgress] + // DeliveredProgress is the latest progress a Workflow Task carried. A delivery whose + // counter is not above it is stale. + DeliveredProgress chasm.Field[*nexuspb.NexusOperationProgress] +} + +// ProgressStore is implemented by a parent that delivers the operation's progress to its caller. +// A parent without it, such as none for a standalone operation, drops progress. +type ProgressStore interface { + // OnNexusOperationProgress records that the operation holds pending progress. The parent + // decides which Workflow Task takes it. + OnNexusOperationProgress(ctx chasm.MutableContext, operation *Operation) error } // NewOperation creates a new Operation component with the given persisted state. @@ -298,10 +312,9 @@ func (o *Operation) HandleNexusCompletion( return serviceerror.NewNotFound("operation not found") } - // Progress never stands in for the start, and nothing folds it onto a Workflow Task yet, so it - // is accepted and dropped here. - if completion.GetProgress() != nil { - return nil + // Progress never stands in for the start, so it is handled before completion-before-start. + if progress := completion.GetProgress(); progress != nil { + return o.onProgress(ctx, progress) } links := completion.GetLinks() @@ -329,6 +342,56 @@ func (o *Operation) HandleNexusCompletion( } } +// onProgress keeps the progress with the highest counter for the caller. Progress before the +// operation started is dropped, since it cannot stand in for the start response, and so is progress +// after it closed, since its completion supersedes it. +func (o *Operation) onProgress( + ctx chasm.MutableContext, + progress *nexuspb.NexusOperationProgress, +) error { + if o.Status != nexusoperationpb.OPERATION_STATUS_STARTED { + return nil + } + parent, ok := o.Store.TryGet(ctx) + if !ok { + return nil + } + store, ok := parent.(ProgressStore) + if !ok { + return nil + } + for _, seen := range []chasm.Field[*nexuspb.NexusOperationProgress]{ + o.PendingProgress, + o.DeliveredProgress, + } { + if previous, ok := seen.TryGet(ctx); ok && previous.GetCounter() >= progress.GetCounter() { + return nil + } + } + o.PendingProgress = chasm.NewDataField(ctx, progress) + return store.OnNexusOperationProgress(ctx, o) +} + +// TakePendingProgress returns the pending progress and records it as delivered. +func (o *Operation) TakePendingProgress( + ctx chasm.MutableContext, +) (*nexuspb.NexusOperationProgress, bool) { + progress, ok := o.PendingProgress.TryGet(ctx) + if !ok { + return nil, false + } + o.DeliveredProgress = chasm.NewDataField(ctx, progress) + o.PendingProgress = chasm.NewEmptyField[*nexuspb.NexusOperationProgress]() + return progress, true +} + +// DropPendingProgress forgets the pending progress without delivering it. +func (o *Operation) DropPendingProgress(ctx chasm.MutableContext) { + if _, ok := o.PendingProgress.TryGet(ctx); ok { + o.PendingProgress = chasm.NewEmptyField[*nexuspb.NexusOperationProgress]() + } +} + // loadStartArgs is a ReadComponent callback that loads the start arguments from the operation. func (o *Operation) loadStartArgs( ctx chasm.Context, diff --git a/chasm/lib/nexusoperation/progress_test.go b/chasm/lib/nexusoperation/progress_test.go new file mode 100644 index 00000000000..c80346fe765 --- /dev/null +++ b/chasm/lib/nexusoperation/progress_test.go @@ -0,0 +1,140 @@ +package nexusoperation + +import ( + "testing" + + "github.com/stretchr/testify/require" + nexuspb "go.temporal.io/api/nexus/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/chasm" + nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1" +) + +// progressStoreComponent is a parent that takes progress. +type progressStoreComponent struct { + mockStoreComponent + + pending int +} + +func (s *progressStoreComponent) OnNexusOperationProgress(chasm.MutableContext, *Operation) error { + s.pending++ + return nil +} + +func TestHandleNexusProgress(t *testing.T) { + progress := func(counter int64) *persistencespb.ChasmNexusCompletion { + return &persistencespb.ChasmNexusCompletion{ + Outcome: &persistencespb.ChasmNexusCompletion_Progress{ + Progress: &nexuspb.NexusOperationProgress{Counter: counter}, + }, + } + } + newOp := func(status nexusoperationpb.OperationStatus, store OperationStore) *Operation { + op := newTestOperation() + op.Status = status + if store != nil { + op.Store = chasm.NewMockParentPtr(store) + } + return op + } + pendingCounter := func(ctx chasm.Context, op *Operation) int64 { + p, _ := op.PendingProgress.TryGet(ctx) + return p.GetCounter() + } + deliveredCounter := func(ctx chasm.Context, op *Operation) int64 { + p, _ := op.DeliveredProgress.TryGet(ctx) + return p.GetCounter() + } + + for _, tc := range []struct { + name string + status nexusoperationpb.OperationStatus + store OperationStore + counters []int64 + wantPending int64 + }{ + { + name: "HighestCounterWins", + status: nexusoperationpb.OPERATION_STATUS_STARTED, + store: &progressStoreComponent{}, + counters: []int64{1, 3, 2}, + wantPending: 3, + }, + { + name: "DroppedBeforeStart", + status: nexusoperationpb.OPERATION_STATUS_SCHEDULED, + store: &progressStoreComponent{}, + counters: []int64{1}, + }, + { + name: "DroppedAfterClose", + status: nexusoperationpb.OPERATION_STATUS_SUCCEEDED, + store: &progressStoreComponent{}, + counters: []int64{1}, + }, + { + name: "DroppedWithoutCaller", + status: nexusoperationpb.OPERATION_STATUS_STARTED, + counters: []int64{1}, + }, + { + name: "DroppedByAParentThatTakesNoProgress", + status: nexusoperationpb.OPERATION_STATUS_STARTED, + store: &mockStoreComponent{}, + counters: []int64{1}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + op := newOp(tc.status, tc.store) + for _, counter := range tc.counters { + require.NoError(t, op.HandleNexusCompletion(ctx, progress(counter))) + } + require.Equal(t, tc.wantPending, pendingCounter(ctx, op)) + require.Equal(t, tc.status, op.Status, "progress never changes the operation status") + }) + } + + t.Run("StaleAfterDelivery", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + store := &progressStoreComponent{} + op := newOp(nexusoperationpb.OPERATION_STATUS_STARTED, store) + require.NoError(t, op.HandleNexusCompletion(ctx, progress(5))) + taken, ok := op.TakePendingProgress(ctx) + require.True(t, ok) + require.Equal(t, int64(5), taken.GetCounter()) + require.Zero(t, pendingCounter(ctx, op)) + + require.NoError(t, op.HandleNexusCompletion(ctx, progress(4))) + require.Zero(t, pendingCounter(ctx, op), "a counter at or below the delivered one is stale") + require.NoError(t, op.HandleNexusCompletion(ctx, progress(6))) + require.Equal(t, int64(6), pendingCounter(ctx, op)) + require.Equal(t, 2, store.pending) + }) + + t.Run("NewerProgressWaitsUntilATaskCarriesIt", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + store := &progressStoreComponent{} + op := newOp(nexusoperationpb.OPERATION_STATUS_STARTED, store) + require.NoError(t, op.HandleNexusCompletion(ctx, progress(1))) + _, ok := op.TakePendingProgress(ctx) + require.True(t, ok) + + // Whether it folds into a task that has not started is the parent's call; the operation + // keeps it until a task carries it. + require.NoError(t, op.HandleNexusCompletion(ctx, progress(2))) + require.Equal(t, int64(2), pendingCounter(ctx, op)) + require.Equal(t, int64(1), deliveredCounter(ctx, op)) + require.Equal(t, 2, store.pending) + }) + + t.Run("DropPendingProgress", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + op := newOp(nexusoperationpb.OPERATION_STATUS_STARTED, &progressStoreComponent{}) + require.NoError(t, op.HandleNexusCompletion(ctx, progress(1))) + op.DropPendingProgress(ctx) + require.Zero(t, pendingCounter(ctx, op)) + require.Zero(t, deliveredCounter(ctx, op), "dropped progress is not delivered") + }) +} diff --git a/chasm/lib/workflow/gen/workflowpb/v1/state.go-helpers.pb.go b/chasm/lib/workflow/gen/workflowpb/v1/state.go-helpers.pb.go index d5ffe6df633..14efeb6c366 100644 --- a/chasm/lib/workflow/gen/workflowpb/v1/state.go-helpers.pb.go +++ b/chasm/lib/workflow/gen/workflowpb/v1/state.go-helpers.pb.go @@ -189,3 +189,40 @@ func (this *StreamWakeState) Equal(that interface{}) bool { return proto.Equal(this, that1) } + +// Marshal an object of type NexusProgressState to the protobuf v3 wire format +func (val *NexusProgressState) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type NexusProgressState from the protobuf v3 wire format +func (val *NexusProgressState) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *NexusProgressState) Size() int { + return proto.Size(val) +} + +// Equal returns whether two NexusProgressState values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *NexusProgressState) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *NexusProgressState + switch t := that.(type) { + case *NexusProgressState: + that1 = t + case NexusProgressState: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} diff --git a/chasm/lib/workflow/gen/workflowpb/v1/state.pb.go b/chasm/lib/workflow/gen/workflowpb/v1/state.pb.go index 61401cb900d..b21e51c4d94 100644 --- a/chasm/lib/workflow/gen/workflowpb/v1/state.pb.go +++ b/chasm/lib/workflow/gen/workflowpb/v1/state.pb.go @@ -324,6 +324,53 @@ func (x *StreamWakeState) GetSources() []string { return nil } +// The Workflow's Nexus operation progress, keyed by the scheduled event ID of each operation as a +// decimal string. Kept on the Workflow so a transaction learns whether any progress waits without +// loading every operation. The progress itself stays on each operation. +type NexusProgressState struct { + state protoimpl.MessageState `protogen:"open.v1"` + Index *TaskCarriedIndex `protobuf:"bytes,1,opt,name=index,proto3" json:"index,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NexusProgressState) Reset() { + *x = NexusProgressState{} + mi := &file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NexusProgressState) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NexusProgressState) ProtoMessage() {} + +func (x *NexusProgressState) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NexusProgressState.ProtoReflect.Descriptor instead. +func (*NexusProgressState) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDescGZIP(), []int{5} +} + +func (x *NexusProgressState) GetIndex() *TaskCarriedIndex { + if x != nil { + return x.Index + } + return nil +} + var File_temporal_server_chasm_lib_workflow_proto_v1_state_proto protoreflect.FileDescriptor const file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDesc = "" + @@ -349,7 +396,9 @@ const file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDesc = "" "\n" + "WakesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x128\n" + - "\x05value\x18\x02 \x01(\v2\".temporal.api.stream.v1.StreamWakeR\x05value:\x028\x01BDZBgo.temporal.io/server/chasm/lib/workflow/gen/workflowpb;workflowpbb\x06proto3" + "\x05value\x18\x02 \x01(\v2\".temporal.api.stream.v1.StreamWakeR\x05value:\x028\x01\"i\n" + + "\x12NexusProgressState\x12S\n" + + "\x05index\x18\x01 \x01(\v2=.temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndexR\x05indexBDZBgo.temporal.io/server/chasm/lib/workflow/gen/workflowpb;workflowpbb\x06proto3" var ( file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDescOnce sync.Once @@ -363,28 +412,30 @@ func file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDescGZIP() return file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDescData } -var file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_msgTypes = make([]protoimpl.MessageInfo, 7) var file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_goTypes = []any{ (*NexusOperationParentData)(nil), // 0: temporal.server.chasm.lib.workflow.proto.v1.NexusOperationParentData (*NexusCancellationParentData)(nil), // 1: temporal.server.chasm.lib.workflow.proto.v1.NexusCancellationParentData (*IncomingSignalData)(nil), // 2: temporal.server.chasm.lib.workflow.proto.v1.IncomingSignalData (*TaskCarriedIndex)(nil), // 3: temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex (*StreamWakeState)(nil), // 4: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState - nil, // 5: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.WakesEntry - (*timestamppb.Timestamp)(nil), // 6: google.protobuf.Timestamp - (*v1.StreamWake)(nil), // 7: temporal.api.stream.v1.StreamWake + (*NexusProgressState)(nil), // 5: temporal.server.chasm.lib.workflow.proto.v1.NexusProgressState + nil, // 6: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.WakesEntry + (*timestamppb.Timestamp)(nil), // 7: google.protobuf.Timestamp + (*v1.StreamWake)(nil), // 8: temporal.api.stream.v1.StreamWake } var file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_depIdxs = []int32{ - 6, // 0: temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex.last_carried_time:type_name -> google.protobuf.Timestamp - 6, // 1: temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex.release_time:type_name -> google.protobuf.Timestamp + 7, // 0: temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex.last_carried_time:type_name -> google.protobuf.Timestamp + 7, // 1: temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex.release_time:type_name -> google.protobuf.Timestamp 3, // 2: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.index:type_name -> temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex - 5, // 3: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.wakes:type_name -> temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.WakesEntry - 7, // 4: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.WakesEntry.value:type_name -> temporal.api.stream.v1.StreamWake - 5, // [5:5] is the sub-list for method output_type - 5, // [5:5] is the sub-list for method input_type - 5, // [5:5] is the sub-list for extension type_name - 5, // [5:5] is the sub-list for extension extendee - 0, // [0:5] is the sub-list for field type_name + 6, // 3: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.wakes:type_name -> temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.WakesEntry + 3, // 4: temporal.server.chasm.lib.workflow.proto.v1.NexusProgressState.index:type_name -> temporal.server.chasm.lib.workflow.proto.v1.TaskCarriedIndex + 8, // 5: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeState.WakesEntry.value:type_name -> temporal.api.stream.v1.StreamWake + 6, // [6:6] is the sub-list for method output_type + 6, // [6:6] is the sub-list for method input_type + 6, // [6:6] is the sub-list for extension type_name + 6, // [6:6] is the sub-list for extension extendee + 0, // [0:6] is the sub-list for field type_name } func init() { file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_init() } @@ -398,7 +449,7 @@ func file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDesc), len(file_temporal_server_chasm_lib_workflow_proto_v1_state_proto_rawDesc)), NumEnums: 0, - NumMessages: 6, + NumMessages: 7, NumExtensions: 0, NumServices: 0, }, diff --git a/chasm/lib/workflow/gen/workflowpb/v1/tasks.go-helpers.pb.go b/chasm/lib/workflow/gen/workflowpb/v1/tasks.go-helpers.pb.go index 362d03311e1..2b700999880 100644 --- a/chasm/lib/workflow/gen/workflowpb/v1/tasks.go-helpers.pb.go +++ b/chasm/lib/workflow/gen/workflowpb/v1/tasks.go-helpers.pb.go @@ -41,3 +41,40 @@ func (this *StreamWakeReleaseTask) Equal(that interface{}) bool { return proto.Equal(this, that1) } + +// Marshal an object of type NexusProgressReleaseTask to the protobuf v3 wire format +func (val *NexusProgressReleaseTask) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type NexusProgressReleaseTask from the protobuf v3 wire format +func (val *NexusProgressReleaseTask) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *NexusProgressReleaseTask) Size() int { + return proto.Size(val) +} + +// Equal returns whether two NexusProgressReleaseTask values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *NexusProgressReleaseTask) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *NexusProgressReleaseTask + switch t := that.(type) { + case *NexusProgressReleaseTask: + that1 = t + case NexusProgressReleaseTask: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} diff --git a/chasm/lib/workflow/gen/workflowpb/v1/tasks.pb.go b/chasm/lib/workflow/gen/workflowpb/v1/tasks.pb.go index 5feecd58c92..f68e43f9374 100644 --- a/chasm/lib/workflow/gen/workflowpb/v1/tasks.pb.go +++ b/chasm/lib/workflow/gen/workflowpb/v1/tasks.pb.go @@ -59,12 +59,51 @@ func (*StreamWakeReleaseTask) Descriptor() ([]byte, []int) { return file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDescGZIP(), []int{0} } +// Lets Nexus operation progress that waited out nexusoperation.progressMinInterval schedule a +// Workflow Task. +type NexusProgressReleaseTask struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NexusProgressReleaseTask) Reset() { + *x = NexusProgressReleaseTask{} + mi := &file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NexusProgressReleaseTask) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NexusProgressReleaseTask) ProtoMessage() {} + +func (x *NexusProgressReleaseTask) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NexusProgressReleaseTask.ProtoReflect.Descriptor instead. +func (*NexusProgressReleaseTask) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDescGZIP(), []int{1} +} + var File_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto protoreflect.FileDescriptor const file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDesc = "" + "\n" + "7temporal/server/chasm/lib/workflow/proto/v1/tasks.proto\x12+temporal.server.chasm.lib.workflow.proto.v1\"\x17\n" + - "\x15StreamWakeReleaseTaskBDZBgo.temporal.io/server/chasm/lib/workflow/gen/workflowpb;workflowpbb\x06proto3" + "\x15StreamWakeReleaseTask\"\x1a\n" + + "\x18NexusProgressReleaseTaskBDZBgo.temporal.io/server/chasm/lib/workflow/gen/workflowpb;workflowpbb\x06proto3" var ( file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDescOnce sync.Once @@ -78,9 +117,10 @@ func file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDescGZIP() return file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDescData } -var file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_goTypes = []any{ - (*StreamWakeReleaseTask)(nil), // 0: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeReleaseTask + (*StreamWakeReleaseTask)(nil), // 0: temporal.server.chasm.lib.workflow.proto.v1.StreamWakeReleaseTask + (*NexusProgressReleaseTask)(nil), // 1: temporal.server.chasm.lib.workflow.proto.v1.NexusProgressReleaseTask } var file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_depIdxs = []int32{ 0, // [0:0] is the sub-list for method output_type @@ -101,7 +141,7 @@ func file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_workflow_proto_v1_tasks_proto_rawDesc)), NumEnums: 0, - NumMessages: 1, + NumMessages: 2, NumExtensions: 0, NumServices: 0, }, diff --git a/chasm/lib/workflow/library.go b/chasm/lib/workflow/library.go index 12b3bd7d444..6221a4170a9 100644 --- a/chasm/lib/workflow/library.go +++ b/chasm/lib/workflow/library.go @@ -85,6 +85,7 @@ func (l *library) Components() []*chasm.RegistrableComponent { func (l *library) Tasks() []*chasm.RegistrableTask { return []*chasm.RegistrableTask{ chasm.NewRegistrablePureTask("streamWakeRelease", streamWakeReleaseTaskHandler{}), + chasm.NewRegistrablePureTask("nexusProgressRelease", nexusProgressReleaseTaskHandler{}), } } diff --git a/chasm/lib/workflow/nexus_commands_test.go b/chasm/lib/workflow/nexus_commands_test.go index d2bf45dc980..80c6158f829 100644 --- a/chasm/lib/workflow/nexus_commands_test.go +++ b/chasm/lib/workflow/nexus_commands_test.go @@ -738,7 +738,7 @@ func TestHandleCancelCommand(t *testing.T) { event := tcx.history.Events[0] // TODO: Complete the operation using CHASM equivalent of CompletedEventDefinition. - tcx.wf.removeNexusOperation(event.EventId) + tcx.wf.removeNexusOperation(tcx.chasmCtx, event.EventId) // The operation is no longer in the CHASM tree, so the handler defers to the HSM command handler // via ErrCommandTargetNotFound. @@ -771,7 +771,7 @@ func TestHandleCancelCommand(t *testing.T) { event := tcx.history.Events[0] // TODO: Complete the operation using CHASM equivalent of CompletedEventDefinition. - tcx.wf.removeNexusOperation(event.EventId) + tcx.wf.removeNexusOperation(tcx.chasmCtx, event.EventId) // Try to cancel - should succeed because there's a buffered completion. err = tcx.cancelHandler(tcx.chasmCtx, tcx.wf, commandValidator{maxPayloadSize: 1}, &commandpb.Command{ diff --git a/chasm/lib/workflow/nexus_events.go b/chasm/lib/workflow/nexus_events.go index 11612767d0a..dafe282cab8 100644 --- a/chasm/lib/workflow/nexus_events.go +++ b/chasm/lib/workflow/nexus_events.go @@ -230,7 +230,7 @@ func (d CompletedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, }); err != nil { return err } - wf.removeNexusOperation(attrs.GetScheduledEventId()) + wf.removeNexusOperation(ctx, attrs.GetScheduledEventId()) return nil } @@ -269,7 +269,7 @@ func (d FailedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, eve }); err != nil { return err } - wf.removeNexusOperation(attrs.GetScheduledEventId()) + wf.removeNexusOperation(ctx, attrs.GetScheduledEventId()) return nil } @@ -308,7 +308,7 @@ func (d CanceledEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, e }); err != nil { return err } - wf.removeNexusOperation(attrs.GetScheduledEventId()) + wf.removeNexusOperation(ctx, attrs.GetScheduledEventId()) return nil } @@ -344,7 +344,7 @@ func (d TimedOutEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, e }); err != nil { return err } - wf.removeNexusOperation(attrs.GetScheduledEventId()) + wf.removeNexusOperation(ctx, attrs.GetScheduledEventId()) return nil } diff --git a/chasm/lib/workflow/nexus_methods.go b/chasm/lib/workflow/nexus_methods.go index e20ca619d21..ad1374c175e 100644 --- a/chasm/lib/workflow/nexus_methods.go +++ b/chasm/lib/workflow/nexus_methods.go @@ -34,8 +34,9 @@ func (w *Workflow) addNexusOperation( } // removeNexusOperation removes a Nexus operation from the workflow. -func (w *Workflow) removeNexusOperation(key int64) { +func (w *Workflow) removeNexusOperation(ctx chasm.MutableContext, key int64) { delete(w.Operations, key) + w.forgetNexusProgress(ctx, key) } // pendingNexusOperationCount returns the number of pending Nexus operations in the workflow. diff --git a/chasm/lib/workflow/nexus_progress.go b/chasm/lib/workflow/nexus_progress.go new file mode 100644 index 00000000000..a82f5fcd2b0 --- /dev/null +++ b/chasm/lib/workflow/nexus_progress.go @@ -0,0 +1,181 @@ +package workflow + +import ( + "cmp" + "slices" + "strconv" + "time" + + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/nexusoperation" + chasmworkflowpb "go.temporal.io/server/chasm/lib/workflow/gen/workflowpb/v1" + "go.temporal.io/server/common" +) + +var _ nexusoperation.ProgressStore = (*Workflow)(nil) + +// nexusProgressKey is the scheduled event ID that keys an operation in the Workflow. +func nexusProgressKey(op *nexusoperation.Operation) (int64, bool) { + parentData := &chasmworkflowpb.NexusOperationParentData{} + if err := op.GetParentData().UnmarshalTo(parentData); err != nil { + return 0, false + } + return parentData.GetScheduledEventId(), true +} + +// nexusProgressIndexKey is the key of an operation in the carried index. +func nexusProgressIndexKey(scheduledEventID int64) string { + return strconv.FormatInt(scheduledEventID, 10) +} + +// nexusProgressIndex answers the index, or an empty one when no operation ever had progress. +func (w *Workflow) nexusProgressIndex(ctx chasm.Context) *chasmworkflowpb.TaskCarriedIndex { + if state, ok := w.NexusProgress.TryGet(ctx); ok && state.GetIndex() != nil { + return state.GetIndex() + } + return &chasmworkflowpb.TaskCarriedIndex{} +} + +// setNexusProgressIndex stores the index, or drops it when it is empty, so a Workflow whose +// operations report no progress carries nothing. +func (w *Workflow) setNexusProgressIndex( + ctx chasm.MutableContext, + index *chasmworkflowpb.TaskCarriedIndex, +) { + if len(index.GetPending()) == 0 && len(index.GetScheduled()) == 0 && + len(index.GetFolded()) == 0 && index.GetLastCarriedTime() == nil { + if _, ok := w.NexusProgress.TryGet(ctx); ok { + w.NexusProgress = chasm.NewEmptyField[*chasmworkflowpb.NexusProgressState]() + } + return + } + w.NexusProgress = chasm.NewDataField(ctx, &chasmworkflowpb.NexusProgressState{Index: index}) +} + +// OnNexusOperationProgress records that an operation holds progress for the next Workflow Task +// scheduled event. Progress for an operation that rides a scheduled Workflow Task that has not +// started folds into that task: it asks for no task of its own until that one starts. +func (w *Workflow) OnNexusOperationProgress( + ctx chasm.MutableContext, + op *nexusoperation.Operation, +) error { + key, ok := nexusProgressKey(op) + if !ok { + op.DropPendingProgress(ctx) + return nil + } + w.setNexusProgressIndex( + ctx, + carriedIndexAdd(w.nexusProgressIndex(ctx), nexusProgressIndexKey(key)), + ) + return nil +} + +// HasPendingNexusProgress reports whether any operation holds progress for the next Workflow Task +// scheduled event. It reads one field, so it is cheap on every transaction close. +func (w *Workflow) HasPendingNexusProgress(ctx chasm.Context) bool { + return len(w.nexusProgressIndex(ctx).GetPending()) > 0 +} + +// HasScheduledNexusProgress reports whether a scheduled Workflow Task carries progress. +func (w *Workflow) HasScheduledNexusProgress(ctx chasm.Context) bool { + index := w.nexusProgressIndex(ctx) + return len(index.GetScheduled()) > 0 || len(index.GetFolded()) > 0 +} + +// TakeNexusProgress returns the pending progress of every operation in scheduled event ID order, +// each naming its operation, for a Workflow Task scheduled event that is being written. The event +// acknowledges it, and until the task starts newer progress for those operations folds into it. +func (w *Workflow) TakeNexusProgress(ctx chasm.MutableContext) []*nexuspb.NexusOperationProgress { + previous := w.nexusProgressIndex(ctx) + index, keys := carriedIndexTake(previous, ctx.Now(w)) + var taken []*nexuspb.NexusOperationProgress + for _, indexKey := range keys { + key, err := strconv.ParseInt(indexKey, 10, 64) + if err != nil { + continue + } + field, ok := w.Operations[key] + if !ok { + continue + } + progress, ok := field.Get(ctx).TakePendingProgress(ctx) + if !ok { + continue + } + named := common.CloneProto(progress) + named.Operation = &nexuspb.NexusOperationProgress_ScheduledEventId{ScheduledEventId: key} + taken = append(taken, named) + } + // The index sorts its keys as strings, and the event lists operations by scheduled event ID. + slices.SortFunc(taken, func(a, b *nexuspb.NexusOperationProgress) int { + return cmp.Compare(a.GetScheduledEventId(), b.GetScheduledEventId()) + }) + if len(taken) == 0 { + // No task carried progress, so the rate limit keeps its last mark. + index.LastCarriedTime = previous.GetLastCarriedTime() + } + w.setNexusProgressIndex(ctx, index) + return taken +} + +// NexusProgressReadyAt answers when pending progress may schedule a Workflow Task of its own: +// minInterval after the last task that carried progress. +func (w *Workflow) NexusProgressReadyAt(ctx chasm.Context, minInterval time.Duration) time.Time { + return carriedIndexReadyAt(w.nexusProgressIndex(ctx), minInterval) +} + +// HoldNexusProgress keeps pending progress waiting until readyAt, when a NexusProgressReleaseTask +// lets it schedule a Workflow Task. A task scheduled for any other reason still takes it sooner. +func (w *Workflow) HoldNexusProgress(ctx chasm.MutableContext, readyAt time.Time) { + index, arm := carriedIndexHold(w.nexusProgressIndex(ctx), ctx.Now(w), readyAt) + if !arm { + return + } + w.setNexusProgressIndex(ctx, index) + ctx.AddTask( + w, + chasm.TaskAttributes{ScheduledTime: readyAt}, + &chasmworkflowpb.NexusProgressReleaseTask{}, + ) +} + +// releaseNexusProgress lets held progress schedule a Workflow Task. It changes the Workflow, so +// the transaction that closes after it schedules the task. +func (w *Workflow) releaseNexusProgress(ctx chasm.MutableContext) { + index := cloneCarriedIndex(w.nexusProgressIndex(ctx)) + index.ReleaseTime = nil + w.setNexusProgressIndex(ctx, index) +} + +// nexusProgressHeld reports whether a release is armed for pending progress. +func (w *Workflow) nexusProgressHeld(ctx chasm.Context) bool { + return carriedIndexHeld(w.nexusProgressIndex(ctx)) +} + +// ClearScheduledNexusProgress forgets which operations the scheduled Workflow Task carries, once +// that task starts, fails or times out. +func (w *Workflow) ClearScheduledNexusProgress(ctx chasm.MutableContext) { + if !w.HasScheduledNexusProgress(ctx) { + return + } + w.setNexusProgressIndex(ctx, carriedIndexClearScheduled(w.nexusProgressIndex(ctx))) +} + +// forgetNexusProgress removes a closed operation from the index. +func (w *Workflow) forgetNexusProgress(ctx chasm.MutableContext, key int64) { + indexKey := nexusProgressIndexKey(key) + index := w.nexusProgressIndex(ctx) + if !slices.Contains(index.GetPending(), indexKey) && + !slices.Contains(index.GetScheduled(), indexKey) && + !slices.Contains(index.GetFolded(), indexKey) { + return + } + index = cloneCarriedIndex(index) + isKey := func(k string) bool { return k == indexKey } + index.Pending = slices.DeleteFunc(index.Pending, isKey) + index.Scheduled = slices.DeleteFunc(index.Scheduled, isKey) + index.Folded = slices.DeleteFunc(index.Folded, isKey) + w.setNexusProgressIndex(ctx, index) +} diff --git a/chasm/lib/workflow/nexus_progress_tasks.go b/chasm/lib/workflow/nexus_progress_tasks.go new file mode 100644 index 00000000000..8f46e34e533 --- /dev/null +++ b/chasm/lib/workflow/nexus_progress_tasks.go @@ -0,0 +1,29 @@ +package workflow + +import ( + "go.temporal.io/server/chasm" + chasmworkflowpb "go.temporal.io/server/chasm/lib/workflow/gen/workflowpb/v1" +) + +type nexusProgressReleaseTaskHandler struct { + chasm.PureTaskHandlerBase +} + +func (nexusProgressReleaseTaskHandler) Validate( + ctx chasm.Context, + wf *Workflow, + _ chasm.TaskInvocation, + _ *chasmworkflowpb.NexusProgressReleaseTask, +) (bool, error) { + return wf.nexusProgressHeld(ctx), nil +} + +func (nexusProgressReleaseTaskHandler) Execute( + ctx chasm.MutableContext, + wf *Workflow, + _ chasm.TaskAttributes, + _ *chasmworkflowpb.NexusProgressReleaseTask, +) error { + wf.releaseNexusProgress(ctx) + return nil +} diff --git a/chasm/lib/workflow/nexus_progress_test.go b/chasm/lib/workflow/nexus_progress_test.go new file mode 100644 index 00000000000..496ea445c5f --- /dev/null +++ b/chasm/lib/workflow/nexus_progress_test.go @@ -0,0 +1,166 @@ +package workflow + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + nexuspb "go.temporal.io/api/nexus/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/nexusoperation" + nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1" + chasmworkflowpb "go.temporal.io/server/chasm/lib/workflow/gen/workflowpb/v1" + "google.golang.org/protobuf/types/known/anypb" +) + +func TestNexusProgressIndex(t *testing.T) { + newWorkflowWithOperations := func( + t *testing.T, + ctx chasm.MutableContext, + keys ...int64, + ) *Workflow { + wf := &Workflow{} + for _, key := range keys { + parentData, err := anypb.New( + &chasmworkflowpb.NexusOperationParentData{ScheduledEventId: key}, + ) + require.NoError(t, err) + op := nexusoperation.NewOperation(&nexusoperationpb.OperationState{ + Status: nexusoperationpb.OPERATION_STATUS_STARTED, + ParentData: parentData, + }) + op.Store = chasm.NewMockParentPtr[nexusoperation.OperationStore](wf) + wf.addNexusOperation(ctx, key, op) + } + return wf + } + deliver := func(t *testing.T, ctx chasm.MutableContext, wf *Workflow, key, counter int64) { + err := wf.Operations[key].Get(ctx). + HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{ + Outcome: &persistencespb.ChasmNexusCompletion_Progress{ + Progress: &nexuspb.NexusOperationProgress{Counter: counter}, + }, + }) + require.NoError(t, err) + } + counters := func(taken []*nexuspb.NexusOperationProgress) map[int64]int64 { + out := map[int64]int64{} + for _, p := range taken { + out[p.GetScheduledEventId()] = p.GetCounter() + } + return out + } + + t.Run("TakesPendingProgressPerOperationInScheduledOrder", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + // The index keys sort as strings, where "12" comes before "5". + wf := newWorkflowWithOperations(t, ctx, 12, 5) + require.False(t, wf.HasPendingNexusProgress(ctx)) + deliver(t, ctx, wf, 12, 1) + deliver(t, ctx, wf, 5, 2) + deliver(t, ctx, wf, 12, 3) + require.True(t, wf.HasPendingNexusProgress(ctx)) + + taken := wf.TakeNexusProgress(ctx) + require.Len(t, taken, 2) + require.Equal(t, int64(5), taken[0].GetScheduledEventId()) + require.Equal(t, map[int64]int64{5: 2, 12: 3}, counters(taken)) + require.False(t, wf.HasPendingNexusProgress(ctx)) + require.True(t, wf.HasScheduledNexusProgress(ctx)) + }) + + t.Run("FoldsIntoTheUnstartedTaskThenWaitsForTheNextOne", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + wf := newWorkflowWithOperations(t, ctx, 5, 9) + deliver(t, ctx, wf, 5, 1) + require.Len(t, wf.TakeNexusProgress(ctx), 1) + + deliver(t, ctx, wf, 5, 2) + deliver(t, ctx, wf, 5, 3) + require.False(t, wf.HasPendingNexusProgress(ctx), "operation 5 rides the unstarted task") + deliver(t, ctx, wf, 9, 1) + require.True(t, wf.HasPendingNexusProgress(ctx), "operation 9 is not on that task") + require.Equal(t, map[int64]int64{9: 1}, counters(wf.TakeNexusProgress(ctx)), + "folded progress is not taken while its task waits to start") + + // The task started with counter 1 on its event, so counter 3 needs one more task. + wf.ClearScheduledNexusProgress(ctx) + require.False(t, wf.HasScheduledNexusProgress(ctx)) + require.True(t, wf.HasPendingNexusProgress(ctx)) + require.Equal(t, map[int64]int64{5: 3}, counters(wf.TakeNexusProgress(ctx))) + }) + + t.Run("ForgetsAClosedOperation", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + wf := newWorkflowWithOperations(t, ctx, 5, 9) + deliver(t, ctx, wf, 5, 1) + deliver(t, ctx, wf, 9, 1) + require.Len(t, wf.TakeNexusProgress(ctx), 2) + deliver(t, ctx, wf, 5, 2) + deliver(t, ctx, wf, 9, 2) + wf.removeNexusOperation(ctx, 5) + wf.ClearScheduledNexusProgress(ctx) + require.Equal( + t, + map[int64]int64{9: 2}, + counters(wf.TakeNexusProgress(ctx)), + "a closed operation's folded progress is gone", + ) + wf.removeNexusOperation(ctx, 9) + require.False(t, wf.HasScheduledNexusProgress(ctx)) + state, ok := wf.NexusProgress.TryGet(ctx) + require.True(t, ok) + index := state.GetIndex() + require.Empty(t, index.GetPending()) + require.Empty(t, index.GetScheduled()) + require.Empty(t, index.GetFolded()) + require.NotNil( + t, + index.GetLastCarriedTime(), + "only the last carried time stays, for the rate limit", + ) + }) + + t.Run("StoresNoIndexWithoutProgress", func(t *testing.T) { + ctx := &chasm.MockMutableContext{} + wf := newWorkflowWithOperations(t, ctx, 5) + wf.ClearScheduledNexusProgress(ctx) + require.Empty(t, wf.TakeNexusProgress(ctx)) + _, ok := wf.NexusProgress.TryGet(ctx) + require.False(t, ok) + }) + + t.Run("HoldsProgressUntilTheIntervalSinceTheLastCarry", func(t *testing.T) { + now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC) + ctx := &chasm.MockMutableContext{MockContext: chasm.MockContext{ + HandleNow: func(chasm.Component) time.Time { return now }, + }} + wf := newWorkflowWithOperations(t, ctx, 5) + require.True(t, wf.NexusProgressReadyAt(ctx, time.Second).IsZero(), "nothing carried yet") + deliver(t, ctx, wf, 5, 1) + require.Len(t, wf.TakeNexusProgress(ctx), 1) + wf.ClearScheduledNexusProgress(ctx) + readyAt := wf.NexusProgressReadyAt(ctx, time.Second) + require.Equal(t, now.Add(time.Second), readyAt) + + deliver(t, ctx, wf, 5, 2) + wf.HoldNexusProgress(ctx, readyAt) + wf.HoldNexusProgress(ctx, readyAt) + require.Len(t, ctx.Tasks, 1, "one armed release covers the same deadline") + require.Equal(t, readyAt, ctx.Tasks[0].Attributes.ScheduledTime) + require.True(t, wf.nexusProgressHeld(ctx)) + require.True(t, wf.HasPendingNexusProgress(ctx), "held progress still rides any other task") + + wf.releaseNexusProgress(ctx) + require.False(t, wf.nexusProgressHeld(ctx)) + require.True(t, wf.HasPendingNexusProgress(ctx)) + + wf.HoldNexusProgress(ctx, readyAt) + require.Len(t, ctx.Tasks, 2, "a released hold arms again") + now = now.Add(time.Second) + require.Len(t, wf.TakeNexusProgress(ctx), 1) + require.False(t, wf.nexusProgressHeld(ctx), "a task that takes the progress ends the hold") + require.Equal(t, now.Add(time.Second), wf.NexusProgressReadyAt(ctx, time.Second)) + }) +} diff --git a/chasm/lib/workflow/proto/v1/state.proto b/chasm/lib/workflow/proto/v1/state.proto index 7a5128e677d..8d655dd4b5d 100644 --- a/chasm/lib/workflow/proto/v1/state.proto +++ b/chasm/lib/workflow/proto/v1/state.proto @@ -57,3 +57,10 @@ message StreamWakeState { // history.maximumStreamWakeSources. repeated string sources = 3; } + +// The Workflow's Nexus operation progress, keyed by the scheduled event ID of each operation as a +// decimal string. Kept on the Workflow so a transaction learns whether any progress waits without +// loading every operation. The progress itself stays on each operation. +message NexusProgressState { + TaskCarriedIndex index = 1; +} diff --git a/chasm/lib/workflow/proto/v1/tasks.proto b/chasm/lib/workflow/proto/v1/tasks.proto index 792db2acda4..444f1fa57b5 100644 --- a/chasm/lib/workflow/proto/v1/tasks.proto +++ b/chasm/lib/workflow/proto/v1/tasks.proto @@ -6,3 +6,7 @@ option go_package = "go.temporal.io/server/chasm/lib/workflow/gen/workflowpb;wor // Lets stream wakes that waited out streamwake.minInterval schedule a Workflow Task. message StreamWakeReleaseTask {} + +// Lets Nexus operation progress that waited out nexusoperation.progressMinInterval schedule a +// Workflow Task. +message NexusProgressReleaseTask {} diff --git a/chasm/lib/workflow/workflow.go b/chasm/lib/workflow/workflow.go index 2c6342472dc..98fe5ed3d8a 100644 --- a/chasm/lib/workflow/workflow.go +++ b/chasm/lib/workflow/workflow.go @@ -49,6 +49,10 @@ type Workflow struct { // StreamWakes holds the wakes of the run's stream readers that wait for, or ride, a Workflow // Task, so a transaction can tell whether a task must be scheduled for them. StreamWakes chasm.Field[*chasmworkflowpb.StreamWakeState] + + // NexusProgress indexes the Nexus operations whose progress waits for, or rides, a Workflow + // Task, so a transaction can tell without loading every operation. + NexusProgress chasm.Field[*chasmworkflowpb.NexusProgressState] } func NewWorkflow( diff --git a/common/nexus/nexusrpc/completion.go b/common/nexus/nexusrpc/completion.go index 6db219efe9a..9c5e8b2792b 100644 --- a/common/nexus/nexusrpc/completion.go +++ b/common/nexus/nexusrpc/completion.go @@ -104,8 +104,11 @@ type CompleteOperationOptions struct { // Error to send with the completion. If set, the completion is unsuccessful. Error *nexus.OperationError // Result to deliver with the completion. Uses the client's serializer to serialize the result into the request body. - // Only used for successful completions. + // Only used for successful completions and progress. Result any + // Progress makes the request a progress delivery rather than a completion: the state is running + // and Result holds the progress body. The operation stays open. + Progress bool } // nolint:revive // This method is long but it's more readable to keep the logic in one place since it's all related to @@ -154,7 +157,11 @@ func (c CompleteOperationOptions) applyToHTTPRequest(cc *CompletionHTTPClient, r addContentHeaderToHTTPHeader(reader.Header, request.Header) } request.Body = reader.ReadCloser - request.Header.Set(headerOperationState, string(nexus.OperationStateSucceeded)) + state := nexus.OperationStateSucceeded + if c.Progress { + state = nexus.OperationStateRunning + } + request.Header.Set(headerOperationState, string(state)) } if c.Header != nil { diff --git a/common/nexus/nexusrpc/completion_test.go b/common/nexus/nexusrpc/completion_test.go index 90addfc5660..5e918f80159 100644 --- a/common/nexus/nexusrpc/completion_test.go +++ b/common/nexus/nexusrpc/completion_test.go @@ -3,6 +3,7 @@ package nexusrpc_test import ( "context" "errors" + "io" "net/url" "testing" "time" @@ -258,3 +259,46 @@ func TestBadRequestCompletion(t *testing.T) { require.ErrorAs(t, err, &handlerErr) require.Equal(t, nexus.HandlerErrorTypeBadRequest, handlerErr.Type) } + +type progressHandler struct { + got chan *nexusrpc.CompletionRequest + body chan []byte +} + +func (h *progressHandler) CompleteOperation( + ctx context.Context, + completion *nexusrpc.CompletionRequest, +) error { + body, err := io.ReadAll(completion.Result.Reader) + if err != nil { + return err + } + h.got <- completion + h.body <- body + return nil +} + +func TestProgressDelivery(t *testing.T) { + h := &progressHandler{ + got: make(chan *nexusrpc.CompletionRequest, 1), + body: make(chan []byte, 1), + } + ctx, callbackURL, teardown := setupForCompletion(t, h, nil, nil) + defer teardown() + + c := nexusrpc.NewCompletionHTTPClient(nexusrpc.CompletionHTTPClientOptions{}) + err := c.CompleteOperation(ctx, callbackURL, nexusrpc.CompleteOperationOptions{ + Result: &nexus.Content{ + Header: nexus.Header{"type": "application/json"}, + Data: []byte(`{"counter": 2}`), + }, + Progress: true, + OperationToken: "test-operation-token", + }) + require.NoError(t, err) + completion := <-h.got + require.Equal(t, nexus.OperationStateRunning, completion.State) + require.Equal(t, "test-operation-token", completion.OperationToken) + require.True(t, completion.CloseTime.IsZero(), "progress leaves the operation open") + require.JSONEq(t, `{"counter": 2}`, string(<-h.body)) +} diff --git a/service/frontend/nexus_completion_http_handler.go b/service/frontend/nexus_completion_http_handler.go index db93f128df0..454b70abce2 100644 --- a/service/frontend/nexus_completion_http_handler.go +++ b/service/frontend/nexus_completion_http_handler.go @@ -521,10 +521,13 @@ func (h *nexusCompletionHandler) forwardCompleteOperation(ctx context.Context, r Links: r.Links, } case nexus.OperationStateRunning: - // A 400 would turn progress off for the callback. Progress is not forwarded, and the next - // delivery after a failover reaches the active cluster. - return nexus.NewHandlerErrorf( - nexus.HandlerErrorTypeUnavailable, "operation progress is not forwarded") + completion = nexusrpc.CompleteOperationOptions{ + Result: r.Result.Reader, + Progress: true, + OperationToken: r.OperationToken, + StartTime: r.StartTime, + Links: r.Links, + } default: return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid operation state: %q", r.State) } diff --git a/service/history/configs/config.go b/service/history/configs/config.go index 7e4b752e2c0..a40f17bfa73 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -82,6 +82,7 @@ type Config struct { EnableCHASMSkipPersistence dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableChasmNexusWorkflowOperations dynamicconfig.BoolPropertyFnWithNamespaceFilter ChasmNexusWorkflowOperationsRolloutPercent dynamicconfig.IntPropertyFnWithNamespaceFilter + NexusOperationProgressMinInterval dynamicconfig.DurationPropertyFnWithNamespaceFilter StreamWakeMinInterval dynamicconfig.DurationPropertyFnWithNamespaceFilter MaximumStreamWakeSources dynamicconfig.IntPropertyFnWithNamespaceFilter EnableCHASMCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter @@ -549,6 +550,7 @@ func NewConfig( EnableCHASMSkipPersistence: dynamicconfig.EnableCHASMSkipPersistence.Get(dc), EnableChasmNexusWorkflowOperations: nexusoperation.EnableChasmWorkflowOperations.Get(dc), ChasmNexusWorkflowOperationsRolloutPercent: nexusoperation.ChasmWorkflowOperationsRolloutPercent.Get(dc), + NexusOperationProgressMinInterval: nexusoperation.ProgressMinInterval.Get(dc), StreamWakeMinInterval: dynamicconfig.StreamWakeMinInterval.Get(dc), MaximumStreamWakeSources: dynamicconfig.MaximumStreamWakeSources.Get(dc), ChasmMaxInMemoryPureTasks: dynamicconfig.ChasmMaxInMemoryPureTasks.Get(dc), diff --git a/service/history/workflow/mutable_state_impl_test.go b/service/history/workflow/mutable_state_impl_test.go index 364c68fa4c6..1740100ddd1 100644 --- a/service/history/workflow/mutable_state_impl_test.go +++ b/service/history/workflow/mutable_state_impl_test.go @@ -38,6 +38,7 @@ import ( chasmnexus "go.temporal.io/server/chasm/lib/nexusoperation" "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1" chasmworkflow "go.temporal.io/server/chasm/lib/workflow" + chasmworkflowpb "go.temporal.io/server/chasm/lib/workflow/gen/workflowpb/v1" "go.temporal.io/server/common" "go.temporal.io/server/common/clock" "go.temporal.io/server/common/cluster" @@ -74,6 +75,7 @@ import ( "go.temporal.io/server/service/history/tests" "go.uber.org/mock/gomock" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/fieldmaskpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -2418,6 +2420,110 @@ func (s *mutableStateSuite) TestStreamWakesAcrossFailoverConversion() { }) } +// TestNexusProgressAcrossFailoverConversion checks that a transient Workflow Task a failover +// converts doesn't carry Nexus operation progress, and that the progress waits for a later task +// instead of being dropped. +func (s *mutableStateSuite) TestNexusProgressAcrossFailoverConversion() { + version := int64(2000) + // prepare leaves a failed first task, so the next task is a transient retry. + prepare := func() { + workflowID := "some random workflow ID" + runID := uuid.NewString() + s.mutableState = TestGlobalMutableState( + s.mockShard, s.mockEventsCache, s.logger, version, workflowID, runID) + _, _ = s.prepareTransientWorkflowTaskCompletionFirstBatchApplied(version, workflowID, runID) + s.NoError(s.mutableState.ApplyWorkflowTaskFailedEvent()) + versionHistory, err := versionhistory.GetCurrentVersionHistory( + s.mutableState.GetExecutionInfo().GetVersionHistories()) + s.NoError(err) + s.NoError( + versionhistory.AddOrUpdateVersionHistoryItem( + versionHistory, + &historyspb.VersionHistoryItem{ + EventId: s.mutableState.GetNextEventID() - 1, + Version: version, + }, + ), + ) + + registry := chasm.NewRegistry(s.logger) + s.NoError(registry.Register(&chasm.CoreLibrary{})) + s.NoError(registry.Register(chasmworkflow.NewLibrary(chasmworkflow.NewRegistry()))) + s.NoError(registry.Register(chasmnexus.NewNilLibrary())) + tree := chasm.NewEmptyTree( + registry, + s.mutableState, + chasm.DefaultPathEncoder, + s.logger, + metrics.NoopMetricsHandler, + ) + s.mutableState.chasmTree = tree + ctx := chasm.NewMutableContext(context.Background(), tree) + wf := chasmworkflow.NewWorkflow(ctx, chasm.NewMSPointer(s.mutableState)) + parentData, err := anypb.New(&chasmworkflowpb.NexusOperationParentData{ScheduledEventId: 5}) + s.NoError(err) + op := chasmnexus.NewOperation(&nexusoperationpb.OperationState{ + Status: nexusoperationpb.OPERATION_STATUS_STARTED, + ParentData: parentData, + }) + wf.Operations = chasm.Map[int64, *chasmnexus.Operation]{5: chasm.NewComponentField(ctx, op)} + s.NoError(tree.SetRootComponent(wf)) + op.PendingProgress = chasm.NewDataField(ctx, &nexuspb.NexusOperationProgress{Counter: 1}) + s.NoError(wf.OnNexusOperationProgress(ctx, op)) + s.NotEmpty(s.mutableState.pendingCarriers()) + } + scheduledProgress := func() []*nexuspb.NexusOperationProgress { + mutation, err := s.mutableState.hBuilder.Finish(true) + s.NoError(err) + var carried []*nexuspb.NexusOperationProgress + for _, batch := range mutation.DBEventsBatches { + for _, event := range batch { + attrs := event.GetWorkflowTaskScheduledEventAttributes() + carried = append(carried, attrs.GetNexusOperationProgress()...) + } + } + return carried + } + + s.Run("ScheduledByProgressAfterTheVersionChanged", func() { + prepare() + s.NoError(s.mutableState.UpdateCurrentVersion(version+1, true)) + s.NoError(s.mutableState.scheduleWorkflowTaskForCarriedState()) + s.True(s.mutableState.HasPendingWorkflowTask()) + s.Empty(scheduledProgress(), "the converted task doesn't carry the progress") + s.NotEmpty(s.mutableState.pendingCarriers(), "the progress waits for a later task") + }) + + s.Run("StartedAfterTheVersionChanged", func() { + prepare() + wt, err := s.mutableState.AddWorkflowTaskScheduledEventAsHeartbeat( + true, timestamp.TimeNowPtrUtc(), enumsspb.WORKFLOW_TASK_TYPE_NORMAL) + s.NoError(err) + s.NoError(s.mutableState.UpdateCurrentVersion(version+1, true)) + f, err := tqid.NewTaskQueueFamily("", "tq") + s.NoError(err) + _, started, err := s.mutableState.AddWorkflowTaskStartedEvent( + wt.ScheduledEventID, + uuid.NewString(), + &taskqueuepb.TaskQueue{ + Name: f.TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW).NormalPartition(5).RpcName(), + }, + "random identity", + nil, + nil, + nil, + false, + nil, + 0, + ) + s.NoError(err) + s.NotNil(started) + s.Equal(int32(1), started.Attempt, "the transient task was converted") + s.Empty(scheduledProgress(), "the converted task doesn't carry the progress") + s.NotEmpty(s.mutableState.pendingCarriers()) + }) +} + func (s *mutableStateSuite) TestTransientWorkflowTaskStart_CurrentVersionChanged() { version := int64(2000) workflowID := "some random workflow ID" diff --git a/service/history/workflow/nexus_progress.go b/service/history/workflow/nexus_progress.go new file mode 100644 index 00000000000..32955816980 --- /dev/null +++ b/service/history/workflow/nexus_progress.go @@ -0,0 +1,56 @@ +package workflow + +import ( + "time" + + historypb "go.temporal.io/api/history/v1" + "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" +) + +// nexusProgressCarrier carries the progress of the Workflow's Nexus operations, keyed by +// operation, on the Workflow Task scheduled event. +type nexusProgressCarrier struct{} + +func (nexusProgressCarrier) hasPending(wf *chasmworkflow.Workflow, ctx chasm.Context) bool { + return wf.HasPendingNexusProgress(ctx) +} + +func (nexusProgressCarrier) hasScheduled(wf *chasmworkflow.Workflow, ctx chasm.Context) bool { + return wf.HasScheduledNexusProgress(ctx) +} + +func (nexusProgressCarrier) take( + wf *chasmworkflow.Workflow, + ctx chasm.MutableContext, + attrs *historypb.WorkflowTaskScheduledEventAttributes, +) { + attrs.NexusOperationProgress = append(attrs.NexusOperationProgress, wf.TakeNexusProgress(ctx)...) +} + +func (nexusProgressCarrier) clearScheduled(wf *chasmworkflow.Workflow, ctx chasm.MutableContext) { + wf.ClearScheduledNexusProgress(ctx) +} + +func (nexusProgressCarrier) readyAt( + wf *chasmworkflow.Workflow, + ctx chasm.Context, + minInterval time.Duration, +) time.Time { + return wf.NexusProgressReadyAt(ctx, minInterval) +} + +func (nexusProgressCarrier) hold( + wf *chasmworkflow.Workflow, + ctx chasm.MutableContext, + readyAt time.Time, +) { + wf.HoldNexusProgress(ctx, readyAt) +} + +func (nexusProgressCarrier) minInterval(ms *MutableStateImpl) time.Duration { + if ms.config.NexusOperationProgressMinInterval == nil { + return 0 + } + return ms.config.NexusOperationProgressMinInterval(ms.GetNamespaceEntry().Name().String()) +} diff --git a/service/history/workflow/task_carried.go b/service/history/workflow/task_carried.go index f1d1ed65ebd..7810c447542 100644 --- a/service/history/workflow/task_carried.go +++ b/service/history/workflow/task_carried.go @@ -32,7 +32,7 @@ type taskCarrier interface { } // taskCarriers lists every kind of state that rides Workflow Task scheduled events. -var taskCarriers = []taskCarrier{streamWakeCarrier{}} +var taskCarriers = []taskCarrier{streamWakeCarrier{}, nexusProgressCarrier{}} type streamWakeCarrier struct{} diff --git a/tests/nexus_progress_test.go b/tests/nexus_progress_test.go index 5c3116384c5..15eb61bd7ed 100644 --- a/tests/nexus_progress_test.go +++ b/tests/nexus_progress_test.go @@ -2,15 +2,22 @@ package tests import ( "context" + "fmt" "net/http" "strings" + "sync" + "testing" "time" "github.com/nexus-rpc/sdk-go/nexus" - "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" + nexuspb "go.temporal.io/api/nexus/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + updatepb "go.temporal.io/api/update/v1" + "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/client" "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" @@ -18,6 +25,8 @@ import ( commonnexus "go.temporal.io/server/common/nexus" "go.temporal.io/server/common/nexus/nexusrpc" "go.temporal.io/server/common/nexus/nexustest" + "go.temporal.io/server/common/testing/testvars" + "go.temporal.io/server/common/testing/updateutils" "go.temporal.io/server/tests/testcore" ) @@ -45,8 +54,8 @@ func postNexusProgress(ctx context.Context, callbackURL, callbackToken, body str // TestNexusOperationProgressIntake drives progress deliveries through the completion endpoint. // Every refusal is a 400, which tells the handler to stop sending progress; an accepted delivery -// changes nothing in the caller's History; progress before the start response is dropped; and a -// closed operation answers 404. +// rides the next Workflow Task scheduled event; progress before the start response is dropped; and +// a closed operation answers 404. func (s *NexusWorkflowTestSuite) TestNexusOperationProgressIntake(chasmEnabled bool) { for _, enabled := range []bool{false, true} { env := s.newTestEnv( @@ -144,10 +153,18 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationProgressIntake(chasmEnabled b } accepted := `{"position": "cursor-2", "counter": "2", "metadata": {"topic": "t"}}` s.Equal(http.StatusOK, progress(accepted)) - // Nothing folds progress yet, so an accepted delivery leaves History as it was. - require.Never(s.T(), func() bool { - return len(env.GetHistory(env.Namespace().String(), wfExec)) != len(started) - }, 300*time.Millisecond, 50*time.Millisecond) + // An accepted delivery rides the next Workflow Task's scheduled event. + scheduledEventID := s.RequireHistoryEvent( + started, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED).GetEventId() + s.Await(func(s *NexusWorkflowTestSuite) { + hist := env.GetHistory(env.Namespace().String(), wfExec) + carried := carriedNexusProgress(hist[len(started):]) + s.Len(carried, 1) + s.Equal(scheduledEventID, carried[0].GetScheduledEventId()) + s.Equal(int64(2), carried[0].GetCounter()) + s.Equal("cursor-2", carried[0].GetPosition()) + s.Equal("t", carried[0].GetMetadata()["topic"]) + }, 10*time.Second, 50*time.Millisecond) } s.NoError(s.sendNexusCompletionRequest(ctx, callbackURL, nexusrpc.CompleteOperationOptions{ @@ -167,3 +184,841 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationProgressIntake(chasmEnabled b w.Stop() } } + +func scheduleNexusOperationCmd(t *testing.T, endpoint string) *commandpb.Command { + attrs := &commandpb.ScheduleNexusOperationCommandAttributes{ + Endpoint: endpoint, + Service: "service", + Operation: "operation", + Input: testcore.MustToPayload(t, "input"), + } + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, + Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{ + ScheduleNexusOperationCommandAttributes: attrs, + }, + } +} + +// carriedNexusProgress lists the Nexus operation progress the scheduled events in hist carry. +func carriedNexusProgress(hist []*historypb.HistoryEvent) []*nexuspb.NexusOperationProgress { + var carried []*nexuspb.NexusOperationProgress + for _, event := range hist { + carried = append( + carried, + event.GetWorkflowTaskScheduledEventAttributes().GetNexusOperationProgress()...) + } + return carried +} + +// TestNexusOperationProgress checks how progress reaches a CHASM caller Workflow: on Workflow Task +// scheduled events with no event of its own, folded into a scheduled task that has not started and +// then carried by one follow-up task so the highest counter arrives, dropped after completion, and +// replayed. The HSM caller refuses progress. +func (s *NexusWorkflowTestSuite) TestNexusOperationProgress(chasmEnabled bool) { + env := s.newTestEnv(chasmEnabled, testcore.WithDynamicConfig(chasmnexus.EnableProgress, true)) + ctx := s.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + + var callbackToken, callbackURL string + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + callbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + var result string + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + if err := fut.Get(ctx, &result); err != nil { + return "", err + } + workflow.GetSignalChannel(ctx, "finish").Receive(ctx, nil) + return result, nil + } + run, err := env.SdkClient(). + ExecuteWorkflow(ctx, client.StartWorkflowOptions{TaskQueue: taskQueue}, callerWF) + s.NoError(err) + newWorker := func() worker.Worker { + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + s.NoError(w.Start()) + return w + } + w := newWorker() + + wfExec := &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: run.GetRunID()} + history := func() []*historypb.HistoryEvent { + return env.GetHistory(env.Namespace().String(), wfExec) + } + // Idle means the Workflow is blocked on the operation or the Signal with no task in flight. + waitIdle := func( + check func(s *NexusWorkflowTestSuite, hist []*historypb.HistoryEvent), + ) []*historypb.HistoryEvent { + var hist []*historypb.HistoryEvent + s.Await(func(s *NexusWorkflowTestSuite) { + desc, err := env.SdkClient().DescribeWorkflowExecution(ctx, run.GetID(), run.GetRunID()) + s.NoError(err) + s.Nil(desc.GetPendingWorkflowTask()) + hist = history() + s.Equal(enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED, hist[len(hist)-1].GetEventType()) + check(s, hist) + }, 10*time.Second, 50*time.Millisecond) + return hist + } + progress := func(counter int) int { + status, err := postNexusProgress( + ctx, + callbackURL, + callbackToken, + fmt.Sprintf(`{"position": "p%d", "counter": %d}`, counter, counter), + ) + s.NoError(err) + return status + } + onlyWorkflowTaskEvents := func(hist []*historypb.HistoryEvent) { + for _, event := range hist { + s.Contains([]enumspb.EventType{ + enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, + enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED, + enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED, + }, event.GetEventType(), "progress must add no event of its own") + } + } + + beforeBurst := waitIdle(func(s *NexusWorkflowTestSuite, hist []*historypb.HistoryEvent) { + s.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + }) + if !chasmEnabled { + s.Equal( + http.StatusBadRequest, + progress(1), + "an HSM caller refuses progress, so the handler stops sending it", + ) + w.Stop() + return + } + scheduledEventID := s.scheduledEventID(beforeBurst) + + // With no worker polling, the task the first delivery schedules stays unstarted, and later + // deliveries fold into it instead of asking for tasks of their own. + w.Stop() + s.Equal(http.StatusOK, progress(1)) + s.Await(func(s *NexusWorkflowTestSuite) { + s.Len(carriedNexusProgress(history()[len(beforeBurst):]), 1) + }, 10*time.Second, 50*time.Millisecond) + s.Equal(http.StatusOK, progress(2)) + s.Equal(http.StatusOK, progress(3)) + s.Equal(http.StatusOK, progress(2), "a stale counter is accepted and dropped") + w = newWorker() + afterFold := waitIdle(func(s *NexusWorkflowTestSuite, hist []*historypb.HistoryEvent) { + carried := carriedNexusProgress(hist[len(beforeBurst):]) + s.NotEmpty(carried) + s.Equal( + int64(3), + carried[len(carried)-1].GetCounter(), + "the Workflow sees the highest counter", + ) + }) + folded := afterFold[len(beforeBurst):] + onlyWorkflowTaskEvents(folded) + carried := carriedNexusProgress(folded) + s.Len(carried, 2) + s.Equal(scheduledEventID, carried[0].GetScheduledEventId()) + s.Equal( + int64(1), + carried[0].GetCounter(), + "the task that was already scheduled carries what it had", + ) + s.Equal(int64(3), carried[1].GetCounter(), "one follow-up task carries what folded into it") + var scheduledTasks int + for _, event := range folded { + if event.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED { + scheduledTasks++ + } + } + s.Equal(2, scheduledTasks, "the burst costs at most one extra task") + + // After that task started, newer progress needs a task of its own. + s.Equal(http.StatusOK, progress(4)) + afterNext := waitIdle(func(s *NexusWorkflowTestSuite, hist []*historypb.HistoryEvent) { + s.Len(carriedNexusProgress(hist[len(afterFold):]), 1) + }) + onlyWorkflowTaskEvents(afterNext[len(afterFold):]) + s.Equal(int64(4), carriedNexusProgress(afterNext[len(afterFold):])[0].GetCounter()) + s.Equal(http.StatusOK, progress(3), "a counter below a folded one is stale") + + s.NoError(s.sendNexusCompletionRequest(ctx, callbackURL, nexusrpc.CompleteOperationOptions{ + Result: testcore.MustToPayload(s.T(), "result"), + Header: nexus.Header{commonnexus.CallbackTokenHeader: callbackToken}, + })) + afterCompletion := waitIdle(func(s *NexusWorkflowTestSuite, hist []*historypb.HistoryEvent) { + s.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) + }) + s.Equal(http.StatusNotFound, progress(5), "the operation is gone with its completion") + s.Len(history(), len(afterCompletion), "progress after completion must not touch the Workflow") + + s.NoError(env.SdkClient().SignalWorkflow(ctx, run.GetID(), run.GetRunID(), "finish", nil)) + var result string + s.NoError(run.Get(ctx, &result)) + s.Equal("result", result) + w.Stop() + + replayer := worker.NewWorkflowReplayer() + replayer.RegisterWorkflow(callerWF) + s.NoError(replayer.ReplayWorkflowHistory(nil, &historypb.History{Events: history()})) + + resp, err := env.FrontendClient(). + GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ + Namespace: env.Namespace().String(), + Execution: wfExec, + }) + s.NoError(err) + s.Len(carriedNexusProgress(resp.GetHistory().GetEvents()), 3, "the progress is part of History") +} + +// TestNexusOperationProgressSkipsHeartbeatTasks checks that progress arriving during a long local +// activity stays off the tasks a worker forces to heartbeat it, and rides the next normal task. +func (s *NexusWorkflowTestSuite) TestNexusOperationProgressSkipsHeartbeatTasks(chasmEnabled bool) { + if !chasmEnabled { + // An HSM caller refuses progress; TestNexusOperationProgress covers it. + return + } + env := s.newTestEnv(chasmEnabled, testcore.WithDynamicConfig(chasmnexus.EnableProgress, true)) + ctx := s.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + + var callbackToken, callbackURL string + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + callbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + laStarted := make(chan struct{}) + var laStartedOnce sync.Once + releaseLA := make(chan struct{}) + longLocalActivity := func(ctx context.Context) error { + laStartedOnce.Do(func() { close(laStarted) }) + select { + case <-releaseLA: + return nil + case <-ctx.Done(): + return ctx.Err() + } + } + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + if err := fut.GetNexusOperationExecution().Get(ctx, nil); err != nil { + return "", err + } + laCtx := workflow.WithLocalActivityOptions( + ctx, + workflow.LocalActivityOptions{StartToCloseTimeout: time.Minute}, + ) + la := workflow.ExecuteLocalActivity(laCtx, longLocalActivity) + if err := la.Get(ctx, nil); err != nil { + return "", err + } + var result string + err := fut.Get(ctx, &result) + return result, err + } + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + w.RegisterActivity(longLocalActivity) + s.NoError(w.Start()) + defer w.Stop() + + // A short task timeout makes the worker heartbeat the local activity often. + run, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + TaskQueue: taskQueue, + WorkflowTaskTimeout: time.Second, + }, callerWF) + s.NoError(err) + wfExec := &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: run.GetRunID()} + history := func() []*historypb.HistoryEvent { + return env.GetHistory(env.Namespace().String(), wfExec) + } + select { + case <-laStarted: + case <-ctx.Done(): + s.FailNow("the local activity did not start") + } + + beforeProgress := history() + status, err := postNexusProgress( + ctx, + callbackURL, + callbackToken, + `{"position": "p1", "counter": 1}`, + ) + s.NoError(err) + s.Equal(http.StatusOK, status) + countScheduled := func(hist []*historypb.HistoryEvent) int { + var n int + for _, event := range hist { + if event.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED { + n++ + } + } + return n + } + var duringLA []*historypb.HistoryEvent + s.Await(func(s *NexusWorkflowTestSuite) { + duringLA = history()[len(beforeProgress):] + s.GreaterOrEqual( + countScheduled(duringLA), + 2, + "the worker keeps heartbeating the local activity", + ) + }, 10*time.Second, 50*time.Millisecond) + s.Empty(carriedNexusProgress(duringLA), "no heartbeat task carries progress") + + close(releaseLA) + var afterLA []*historypb.HistoryEvent + s.Await(func(s *NexusWorkflowTestSuite) { + afterLA = history()[len(beforeProgress):] + s.Len(carriedNexusProgress(afterLA), 1) + }, 10*time.Second, 50*time.Millisecond) + markerIdx := -1 + carrierIdx := -1 + for i, event := range afterLA { + if event.GetEventType() == enumspb.EVENT_TYPE_MARKER_RECORDED { + markerIdx = i + } + if len(event.GetWorkflowTaskScheduledEventAttributes().GetNexusOperationProgress()) > 0 { + carrierIdx = i + } + } + s.Positive(markerIdx, "the local activity's result is recorded") + s.Greater( + carrierIdx, + markerIdx, + "the first normal task after the heartbeats carries the progress", + ) + s.Equal(int64(1), carriedNexusProgress(afterLA)[0].GetCounter()) + + s.NoError(s.sendNexusCompletionRequest(ctx, callbackURL, nexusrpc.CompleteOperationOptions{ + Result: testcore.MustToPayload(s.T(), "result"), + Header: nexus.Header{commonnexus.CallbackTokenHeader: callbackToken}, + })) + var result string + s.NoError(run.Get(ctx, &result)) + s.Equal("result", result) + + replayer := worker.NewWorkflowReplayer() + replayer.RegisterWorkflow(callerWF) + s.NoError(replayer.ReplayWorkflowHistory(nil, &historypb.History{Events: history()})) +} + +// TestNexusOperationProgressSkipsHeartbeatRetries checks that the retry of a failed heartbeat task +// holds progress like the task it retries, and that the next normal task carries it. A Signal +// arrives before the retry starts, so the retry gets a scheduled event of its own. +func (s *NexusWorkflowTestSuite) TestNexusOperationProgressSkipsHeartbeatRetries( + chasmEnabled bool, +) { + if !chasmEnabled { + // An HSM caller refuses progress; TestNexusOperationProgress covers it. + return + } + env := s.newTestEnv(chasmEnabled, testcore.WithDynamicConfig(chasmnexus.EnableProgress, true)) + ctx := s.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + ns := env.Namespace().String() + + var callbackToken, callbackURL string + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + callbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + run, err := env.SdkClient(). + ExecuteWorkflow(ctx, client.StartWorkflowOptions{TaskQueue: taskQueue}, "workflow") + s.NoError(err) + frontend := env.FrontendClient() + poll := func() *workflowservice.PollWorkflowTaskQueueResponse { + resp, err := frontend.PollWorkflowTaskQueue( + ctx, + &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: ns, + TaskQueue: &taskqueuepb.TaskQueue{ + Name: taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Identity: "test", + }, + ) + s.NoError(err) + s.NotEmpty(resp.GetTaskToken()) + return resp + } + respond := func( + task *workflowservice.PollWorkflowTaskQueueResponse, + forceCreate bool, + commands ...*commandpb.Command, + ) { + _, err := frontend.RespondWorkflowTaskCompleted( + ctx, + &workflowservice.RespondWorkflowTaskCompletedRequest{ + Namespace: ns, + Identity: "test", + TaskToken: task.GetTaskToken(), + Commands: commands, + ForceCreateNewWorkflowTask: forceCreate, + }, + ) + s.NoError(err) + } + // lastScheduled is the scheduled event of the task a poll answered. + lastScheduled := func( + task *workflowservice.PollWorkflowTaskQueueResponse, + ) *historypb.HistoryEvent { + events := task.GetHistory().GetEvents() + for i := len(events) - 1; i >= 0; i-- { + if events[i].GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED { + return events[i] + } + } + s.FailNow("the task has no scheduled event") + return nil + } + + respond(poll(), false, scheduleNexusOperationCmd(s.T(), endpointName)) + startedTask := poll() + s.RequireHistoryEvent( + startedTask.GetHistory().GetEvents(), + enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, + ) + + // The worker forces a heartbeat task, takes it, gets progress meanwhile, and fails it. + respond(startedTask, true) + heartbeatTask := poll() + s.Empty( + lastScheduled( + heartbeatTask, + ).GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress(), + ) + status, err := postNexusProgress( + ctx, + callbackURL, + callbackToken, + `{"position": "p1", "counter": 1}`, + ) + s.NoError(err) + s.Equal(http.StatusOK, status) + _, err = frontend.RespondWorkflowTaskFailed( + ctx, + &workflowservice.RespondWorkflowTaskFailedRequest{ + Namespace: ns, + Identity: "test", + TaskToken: heartbeatTask.GetTaskToken(), + Cause: enumspb.WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE, + }, + ) + s.NoError(err) + s.NoError(env.SdkClient().SignalWorkflow(ctx, run.GetID(), run.GetRunID(), "wake", nil)) + + retryTask := poll() + s.RequireHistoryEvent( + retryTask.GetHistory().GetEvents(), + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, + ) + s.Empty( + lastScheduled( + retryTask, + ).GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress(), + "the retry of a heartbeat task carries no progress", + ) + respond(retryTask, false) + + normalTask := poll() + carried := lastScheduled( + normalTask, + ).GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress() + s.Len(carried, 1, "the next normal task carries the progress") + s.Equal(int64(1), carried[0].GetCounter()) + respond(normalTask, false, completeWorkflowCmd()) + s.NoError(run.Get(ctx, nil)) +} + +// TestNexusOperationProgressRidesSpeculativeTasks checks that progress reaches the Workflow when it +// arrives while a speculative task for an Update is pending, and the Update is then rejected. The +// progress transaction turns the speculative task into a normal one: one that has not started +// carries the progress, and one already started is kept, so a follow-up task carries it. +func (s *NexusWorkflowTestSuite) TestNexusOperationProgressRidesSpeculativeTasks( + chasmEnabled bool, +) { + if !chasmEnabled { + // An HSM caller refuses progress; TestNexusOperationProgress covers it. + return + } + env := s.newTestEnv(chasmEnabled, testcore.WithDynamicConfig(chasmnexus.EnableProgress, true)) + ctx := s.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + ns := env.Namespace().String() + tv := testvars.New(s.T()) + updates := updateutils.New(s.T()) + + var callbackToken, callbackURL string + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + callbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + run, err := env.SdkClient(). + ExecuteWorkflow(ctx, client.StartWorkflowOptions{TaskQueue: taskQueue}, "workflow") + s.NoError(err) + wfExec := &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: run.GetRunID()} + frontend := env.FrontendClient() + poll := func() *workflowservice.PollWorkflowTaskQueueResponse { + resp, err := frontend.PollWorkflowTaskQueue( + ctx, + &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: ns, + TaskQueue: &taskqueuepb.TaskQueue{ + Name: taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Identity: "test", + }, + ) + s.NoError(err) + s.NotEmpty(resp.GetTaskToken()) + return resp + } + respond := func( + task *workflowservice.PollWorkflowTaskQueueResponse, + commands ...*commandpb.Command, + ) { + _, err := frontend.RespondWorkflowTaskCompleted( + ctx, + &workflowservice.RespondWorkflowTaskCompletedRequest{ + Namespace: ns, + Identity: "test", + TaskToken: task.GetTaskToken(), + Commands: commands, + }, + ) + s.NoError(err) + } + rejectUpdate := func(task *workflowservice.PollWorkflowTaskQueueResponse) { + s.Len(task.GetMessages(), 1, "the task carries the Update request") + _, err := frontend.RespondWorkflowTaskCompleted( + ctx, + &workflowservice.RespondWorkflowTaskCompletedRequest{ + Namespace: ns, + Identity: "test", + TaskToken: task.GetTaskToken(), + Messages: updates.UpdateRejectMessages(tv, task.GetMessages()[0]), + }, + ) + s.NoError(err) + } + sendUpdate := func(updateID string) <-chan *updatepb.Outcome { + done := make(chan *updatepb.Outcome, 1) + go func() { + resp, err := frontend.UpdateWorkflowExecution( + ctx, + &workflowservice.UpdateWorkflowExecutionRequest{ + Namespace: ns, + WorkflowExecution: wfExec, + Request: &updatepb.Request{ + Meta: &updatepb.Meta{UpdateId: updateID}, + Input: &updatepb.Input{Name: "update"}, + }, + WaitPolicy: &updatepb.WaitPolicy{ + LifecycleStage: enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED, + }, + }, + ) + s.NoError(err) + done <- resp.GetOutcome() + }() + return done + } + // carried is the progress on the scheduled event of the task a poll answered. + carried := func( + task *workflowservice.PollWorkflowTaskQueueResponse, + ) []*nexuspb.NexusOperationProgress { + events := task.GetHistory().GetEvents() + for i := len(events) - 1; i >= 0; i-- { + if events[i].GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED { + return events[i].GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress() + } + } + return nil + } + postProgress := func(counter int) { + status, err := postNexusProgress( + ctx, + callbackURL, + callbackToken, + fmt.Sprintf(`{"position": "p%d", "counter": %d}`, counter, counter), + ) + s.NoError(err) + s.Equal(http.StatusOK, status) + } + + respond(poll(), scheduleNexusOperationCmd(s.T(), endpointName)) + startedTask := poll() + s.RequireHistoryEvent( + startedTask.GetHistory().GetEvents(), + enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, + ) + respond(startedTask) + + // An Update schedules a speculative task, and progress arrives before any worker takes it. + firstUpdate := sendUpdate("unstarted") + s.Await(func(s *NexusWorkflowTestSuite) { + desc, err := env.SdkClient().DescribeWorkflowExecution(ctx, run.GetID(), run.GetRunID()) + s.NoError(err) + s.NotNil(desc.GetPendingWorkflowTask()) + }, 10*time.Second, 50*time.Millisecond) + postProgress(1) + unstartedTask := poll() + s.Len(carried(unstartedTask), 1, "the task that was speculative carries the progress") + s.Equal(int64(1), carried(unstartedTask)[0].GetCounter()) + rejectUpdate(unstartedTask) + s.NotNil((<-firstUpdate).GetFailure(), "the Update was rejected") + + // This time the worker has already started the speculative task when progress arrives. + secondUpdate := sendUpdate("started") + startedSpeculative := poll() + s.Len(startedSpeculative.GetMessages(), 1) + postProgress(2) + rejectUpdate(startedSpeculative) + s.NotNil((<-secondUpdate).GetFailure(), "the Update was rejected") + followUp := poll() + s.Len(carried(followUp), 1, "a follow-up task carries the progress the started task could not") + s.Equal(int64(2), carried(followUp)[0].GetCounter()) + respond(followUp, completeWorkflowCmd()) + s.NoError(run.Get(ctx, nil)) +} + +func (s *NexusWorkflowTestSuite) scheduledEventID(hist []*historypb.HistoryEvent) int64 { + return s.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED).GetEventId() +} + +// rawProgressCaller drives a caller Workflow with raw Workflow Task polls, so a test controls +// when each task starts and how it ends. +type rawProgressCaller struct { + s *NexusWorkflowTestSuite + env *NexusTestEnv + ctx context.Context + taskQueue string + run client.WorkflowRun + callbackToken string + callbackURL string +} + +func (s *NexusWorkflowTestSuite) newRawProgressCaller( + opts ...testcore.TestOption, +) *rawProgressCaller { + opts = append(opts, testcore.WithDynamicConfig(chasmnexus.EnableProgress, true)) + c := &rawProgressCaller{ + s: s, + env: s.newTestEnv(true, opts...), + ctx: s.Context(), + taskQueue: testcore.RandomizeStr(s.T().Name()), + } + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + c.callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + c.callbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := c.env.createRandomExternalNexusServer(c.ctx, s.T(), h) + run, err := c.env.SdkClient().ExecuteWorkflow( + c.ctx, client.StartWorkflowOptions{TaskQueue: c.taskQueue}, "workflow") + s.NoError(err) + c.run = run + c.respond(c.poll(), scheduleNexusOperationCmd(s.T(), endpointName)) + started := c.poll() + s.RequireHistoryEvent( + started.GetHistory().GetEvents(), + enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, + ) + c.respond(started) + return c +} + +func (c *rawProgressCaller) poll() *workflowservice.PollWorkflowTaskQueueResponse { + resp, err := c.env.FrontendClient(). + PollWorkflowTaskQueue(c.ctx, &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: c.env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{ + Name: c.taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Identity: "test", + }) + c.s.NoError(err) + c.s.NotEmpty(resp.GetTaskToken()) + return resp +} + +func (c *rawProgressCaller) respond( + task *workflowservice.PollWorkflowTaskQueueResponse, + commands ...*commandpb.Command, +) { + _, err := c.env.FrontendClient(). + RespondWorkflowTaskCompleted(c.ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ + Namespace: c.env.Namespace().String(), + Identity: "test", + TaskToken: task.GetTaskToken(), + Commands: commands, + }) + c.s.NoError(err) +} + +func (c *rawProgressCaller) complete(task *workflowservice.PollWorkflowTaskQueueResponse) { + c.respond(task, completeWorkflowCmd()) + c.s.NoError(c.run.Get(c.ctx, nil)) +} + +func (c *rawProgressCaller) postProgress(counter int) { + body := fmt.Sprintf(`{"position": "p%d", "counter": %d}`, counter, counter) + status, err := postNexusProgress(c.ctx, c.callbackURL, c.callbackToken, body) + c.s.NoError(err) + c.s.Equal(http.StatusOK, status) +} + +// lastScheduled is the scheduled event of the task a poll answered. +func lastScheduled(task *workflowservice.PollWorkflowTaskQueueResponse) *historypb.HistoryEvent { + events := task.GetHistory().GetEvents() + for i := len(events) - 1; i >= 0; i-- { + if events[i].GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED { + return events[i] + } + } + return nil +} + +// TestNexusOperationProgressMinInterval checks that progress schedules a task of its own no sooner +// than nexusoperation.progressMinInterval after the last task that carried progress, so a long +// stream can't fill the caller's History. +func (s *NexusWorkflowTestSuite) TestNexusOperationProgressMinInterval(chasmEnabled bool) { + if !chasmEnabled { + // An HSM caller refuses progress; TestNexusOperationProgress covers it. + return + } + const interval = 2 * time.Second + c := s.newRawProgressCaller( + testcore.WithDynamicConfig(chasmnexus.ProgressMinInterval, interval), + ) + + c.postProgress(1) + first := c.poll() + s.Equal(int64(1), lastScheduled(first).GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress()[0].GetCounter()) + c.respond(first) + + c.postProgress(2) + c.postProgress(3) + second := c.poll() + carried := lastScheduled( + second, + ).GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress() + s.Len(carried, 1) + s.Equal(int64(3), carried[0].GetCounter(), "progress that waited folds into one task") + gap := lastScheduled( + second, + ).GetEventTime(). + AsTime(). + Sub(lastScheduled(first).GetEventTime().AsTime()) + s.GreaterOrEqual( + gap, + interval, + "progress waits out the interval since the last task that carried it", + ) + c.complete(second) +} + +// TestNexusOperationProgressOnFailedTask checks where progress is after the task that carried it +// fails. It stays on that task's scheduled event, which the retry's History includes, and the +// retry's own scheduled event doesn't repeat it. So a worker reads progress from every scheduled +// event since its last completed task. +func (s *NexusWorkflowTestSuite) TestNexusOperationProgressOnFailedTask(chasmEnabled bool) { + if !chasmEnabled { + // An HSM caller refuses progress; TestNexusOperationProgress covers it. + return + } + c := s.newRawProgressCaller() + + c.postProgress(1) + failing := c.poll() + s.Len( + lastScheduled( + failing, + ).GetWorkflowTaskScheduledEventAttributes(). + GetNexusOperationProgress(), + 1, + ) + _, err := c.env.FrontendClient(). + RespondWorkflowTaskFailed(c.ctx, &workflowservice.RespondWorkflowTaskFailedRequest{ + Namespace: c.env.Namespace().String(), + Identity: "test", + TaskToken: failing.GetTaskToken(), + Cause: enumspb.WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE, + }) + s.NoError(err) + + retry := c.poll() + s.Equal(int32(2), retry.GetAttempt()) + s.Empty( + lastScheduled(retry).GetWorkflowTaskScheduledEventAttributes().GetNexusOperationProgress(), + "the retry doesn't repeat what the failed task carried", + ) + carried := carriedNexusProgress(retry.GetHistory().GetEvents()) + s.Len(carried, 1, "the failed task's scheduled event still carries the progress") + s.Equal(int64(1), carried[0].GetCounter()) + c.complete(retry) +} diff --git a/tests/xdc/nexus_progress_forwarding_test.go b/tests/xdc/nexus_progress_forwarding_test.go new file mode 100644 index 00000000000..871f1ce1d2f --- /dev/null +++ b/tests/xdc/nexus_progress_forwarding_test.go @@ -0,0 +1,234 @@ +package xdc + +import ( + "context" + "fmt" + "slices" + "testing" + "time" + + "github.com/google/uuid" + "github.com/nexus-rpc/sdk-go/nexus" + "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" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/operatorservice/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/client" + chasmnexus "go.temporal.io/server/chasm/lib/nexusoperation" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/log" + cnexus "go.temporal.io/server/common/nexus" + "go.temporal.io/server/common/nexus/nexusrpc" + "go.temporal.io/server/common/nexus/nexustest" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/tests/testcore" +) + +// NexusProgressForwardingSuite runs CHASM caller Workflows, the only ones that take Nexus +// operation progress, across two clusters. +type NexusProgressForwardingSuite struct { + xdcBaseSuite +} + +func TestNexusProgressForwardingSuite(t *testing.T) { + t.Parallel() + s := &NexusProgressForwardingSuite{} + s.enableTransitionHistory = true + suite.Run(t, s) +} + +func (s *NexusProgressForwardingSuite) SetupSuite() { + s.dynamicConfigOverrides = map[dynamicconfig.Key]any{ + dynamicconfig.RefreshNexusEndpointsMinWait.Key(): 1 * time.Millisecond, + dynamicconfig.EnableChasm.Key(): true, + dynamicconfig.EnableCHASMCallbacks.Key(): true, + chasmnexus.Enabled.Key(): true, + chasmnexus.EnableChasmWorkflowOperations.Key(): true, + chasmnexus.ChasmWorkflowOperationsRolloutPercent.Key(): 100, + chasmnexus.EnableProgress.Key(): true, + } + rpsKey := dynamicconfig.FrontendGlobalNamespaceNamespaceReplicationInducingAPIsRPS.Key() + s.dynamicConfigOverrides[rpsKey] = 1000 + s.setupSuite() +} + +func (s *NexusProgressForwardingSuite) SetupTest() { + s.setupTest() +} + +func (s *NexusProgressForwardingSuite) TearDownSuite() { + s.tearDownSuite() +} + +// TestProgressForwardedFromStandbyToActive posts progress to the standby cluster, which forwards it +// to the active one instead of refusing it, so the caller's next Workflow Task carries it. +func (s *NexusProgressForwardingSuite) TestProgressForwardedFromStandbyToActive() { + // The callback URL points at the standby cluster, so every delivery lands there first. + standbyCallbackURL := "http://" + s.clusters[1].Host(). + FrontendHTTPAddress() + + "/namespaces/{{.NamespaceName}}/nexus/callback" + s.clusters[0].OverrideDynamicConfig(s.T(), chasmnexus.CallbackURLTemplate, standbyCallbackURL) + s.clusters[1].OverrideDynamicConfig(s.T(), chasmnexus.CallbackURLTemplate, standbyCallbackURL) + + ctx := testcore.NewContext() + ns := s.createGlobalNamespace() + taskQueue := fmt.Sprintf("%v-%v", "test-task-queue", uuid.New()) + endpointName := testcore.RandomizedNexusEndpoint(s.T().Name()) + + var callbackToken, callbackURL string + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, + operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(cnexus.CallbackTokenHeader) + callbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + listenAddr := nexustest.AllocListenAddress() + nexustest.NewNexusServer(s.T(), listenAddr, h) + createEndpointReq := &operatorservice.CreateNexusEndpointRequest{ + Spec: &nexuspb.EndpointSpec{ + Name: endpointName, + Target: &nexuspb.EndpointTarget{ + Variant: &nexuspb.EndpointTarget_External_{ + External: &nexuspb.EndpointTarget_External{Url: "http://" + listenAddr}, + }, + }, + }, + } + for _, cluster := range s.clusters { + _, err := cluster.OperatorClient().CreateNexusEndpoint(ctx, createEndpointReq) + s.NoError(err) + } + + activeSDKClient, err := client.Dial(client.Options{ + HostPort: s.clusters[0].Host().FrontendGRPCAddress(), + Namespace: ns, + Logger: log.NewSdkLogger(s.logger), + }) + s.NoError(err) + run, err := activeSDKClient.ExecuteWorkflow( + ctx, + client.StartWorkflowOptions{TaskQueue: taskQueue}, + "workflow", + ) + s.NoError(err) + + active := s.clusters[0].FrontendClient() + poll := func() *workflowservice.PollWorkflowTaskQueueResponse { + resp, err := active.PollWorkflowTaskQueue( + ctx, + &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: ns, + TaskQueue: &taskqueuepb.TaskQueue{ + Name: taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Identity: "test", + }, + ) + s.NoError(err) + return resp + } + respond := func( + resp *workflowservice.PollWorkflowTaskQueueResponse, + commands ...*commandpb.Command, + ) { + _, err := active.RespondWorkflowTaskCompleted( + ctx, + &workflowservice.RespondWorkflowTaskCompletedRequest{ + Identity: "test", + TaskToken: resp.TaskToken, + Commands: commands, + }, + ) + s.NoError(err) + } + + respond(poll(), scheduleNexusOperationCmd(s.T(), endpointName)) + startedTask := poll() + scheduledIdx := slices.IndexFunc( + startedTask.History.Events, + func(e *historypb.HistoryEvent) bool { + return e.GetNexusOperationScheduledEventAttributes() != nil + }, + ) + s.Positive(scheduledIdx) + s.Positive(slices.IndexFunc(startedTask.History.Events, func(e *historypb.HistoryEvent) bool { + return e.GetNexusOperationStartedEventAttributes() != nil + })) + respond(startedTask) + + // The standby has to know the operation before it can forward progress for it. + await.RequireTrue(s.T(), func() bool { + resp, err := s.clusters[1].FrontendClient(). + DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: ns, + Execution: &commonpb.WorkflowExecution{ + WorkflowId: run.GetID(), + RunId: run.GetRunID(), + }, + }) + return err == nil && len(resp.PendingNexusOperations) > 0 + }, 5*time.Second, 200*time.Millisecond) + + c := nexusrpc.NewCompletionHTTPClient(nexusrpc.CompletionHTTPClientOptions{}) + err = c.CompleteOperation(ctx, callbackURL, nexusrpc.CompleteOperationOptions{ + Result: &nexus.Content{ + Header: nexus.Header{"type": "application/json"}, + Data: []byte(`{"position": "p1", "counter": 1}`), + }, + Progress: true, + Header: nexus.Header{cnexus.CallbackTokenHeader: callbackToken}, + }) + s.NoError(err, "the standby forwards progress rather than refusing it") + + progressTask := poll() + var carried []*nexuspb.NexusOperationProgress + for _, event := range progressTask.History.Events { + carried = append( + carried, + event.GetWorkflowTaskScheduledEventAttributes().GetNexusOperationProgress()...) + } + s.Len(carried, 1) + s.Equal(startedTask.History.Events[scheduledIdx].GetEventId(), carried[0].GetScheduledEventId()) + s.Equal(int64(1), carried[0].GetCounter()) + s.Equal("p1", carried[0].GetPosition()) + respond(progressTask, completeWorkflowCmd()) + s.NoError(run.Get(ctx, nil)) +} + +func scheduleNexusOperationCmd(t *testing.T, endpoint string) *commandpb.Command { + attrs := &commandpb.ScheduleNexusOperationCommandAttributes{ + Endpoint: endpoint, + Service: "service", + Operation: "operation", + Input: testcore.MustToPayload(t, "input"), + } + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, + Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{ + ScheduleNexusOperationCommandAttributes: attrs, + }, + } +} + +func completeWorkflowCmd() *commandpb.Command { + attrs := &commandpb.CompleteWorkflowExecutionCommandAttributes{} + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{ + CompleteWorkflowExecutionCommandAttributes: attrs, + }, + } +}