diff --git a/Makefile b/Makefile index 88d01fe4909..4ae6fc07ac7 100644 --- a/Makefile +++ b/Makefile @@ -525,14 +525,18 @@ workflowcheck: $(WORKFLOWCHECK) check: lint shell-check ##### Tests ##### +# verify-test-log reads the whole file, so a log left behind by an earlier run +# would be judged as if it belonged to this one. clean-test-output: @printf $(COLOR) "Delete test output..." @rm -rf $(TEST_OUTPUT_ROOT) + @rm -f test.log @go clean -testcache build-tests: @printf $(COLOR) "Build tests..." - @CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(TEST_DIRS) + @CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(filter-out $(MIXED_BRAIN_TEST_ROOT)%,$(TEST_DIRS)) + @cd $(MIXED_BRAIN_TEST_ROOT) && CGO_ENABLED=1 go test $(TEST_TAG_FLAG) -exec="true" -count=0 ./... unit-test: clean-test-output @printf $(COLOR) "Run unit tests..." diff --git a/client/frontend/client_gen.go b/client/frontend/client_gen.go index 7d8513a7298..a33cb417953 100644 --- a/client/frontend/client_gen.go +++ b/client/frontend/client_gen.go @@ -199,6 +199,16 @@ func (c *clientImpl) DescribeBatchOperation( return c.client.DescribeBatchOperation(ctx, request, opts...) } +func (c *clientImpl) DescribeChannel( + ctx context.Context, + request *workflowservice.DescribeChannelRequest, + opts ...grpc.CallOption, +) (*workflowservice.DescribeChannelResponse, error) { + ctx, cancel := c.createContext(ctx) + defer cancel() + return c.client.DescribeChannel(ctx, request, opts...) +} + func (c *clientImpl) DescribeDeployment( ctx context.Context, request *workflowservice.DescribeDeploymentRequest, @@ -569,6 +579,16 @@ func (c *clientImpl) ListWorkflowRules( return c.client.ListWorkflowRules(ctx, request, opts...) } +func (c *clientImpl) NotifyChannel( + ctx context.Context, + request *workflowservice.NotifyChannelRequest, + opts ...grpc.CallOption, +) (*workflowservice.NotifyChannelResponse, error) { + ctx, cancel := c.createContext(ctx) + defer cancel() + return c.client.NotifyChannel(ctx, request, opts...) +} + func (c *clientImpl) PatchSchedule( ctx context.Context, request *workflowservice.PatchScheduleRequest, @@ -629,6 +649,16 @@ func (c *clientImpl) PollActivityTaskQueue( return c.client.PollActivityTaskQueue(ctx, request, opts...) } +func (c *clientImpl) PollChannel( + ctx context.Context, + request *workflowservice.PollChannelRequest, + opts ...grpc.CallOption, +) (*workflowservice.PollChannelResponse, error) { + ctx, cancel := c.createContext(ctx) + defer cancel() + return c.client.PollChannel(ctx, request, opts...) +} + func (c *clientImpl) PollNexusOperationExecution( ctx context.Context, request *workflowservice.PollNexusOperationExecutionRequest, @@ -719,6 +749,16 @@ func (c *clientImpl) RecordWorkerHeartbeat( return c.client.RecordWorkerHeartbeat(ctx, request, opts...) } +func (c *clientImpl) RegisterChannelListener( + ctx context.Context, + request *workflowservice.RegisterChannelListenerRequest, + opts ...grpc.CallOption, +) (*workflowservice.RegisterChannelListenerResponse, error) { + ctx, cancel := c.createContext(ctx) + defer cancel() + return c.client.RegisterChannelListener(ctx, request, opts...) +} + func (c *clientImpl) RegisterNamespace( ctx context.Context, request *workflowservice.RegisterNamespaceRequest, @@ -1109,6 +1149,16 @@ func (c *clientImpl) UnpauseWorkflowExecution( return c.client.UnpauseWorkflowExecution(ctx, request, opts...) } +func (c *clientImpl) UnregisterChannelListener( + ctx context.Context, + request *workflowservice.UnregisterChannelListenerRequest, + opts ...grpc.CallOption, +) (*workflowservice.UnregisterChannelListenerResponse, error) { + ctx, cancel := c.createContext(ctx) + defer cancel() + return c.client.UnregisterChannelListener(ctx, request, opts...) +} + func (c *clientImpl) UpdateActivityExecutionOptions( ctx context.Context, request *workflowservice.UpdateActivityExecutionOptionsRequest, diff --git a/client/frontend/metric_client_gen.go b/client/frontend/metric_client_gen.go index a79a8e55ed1..a6c3c049007 100644 --- a/client/frontend/metric_client_gen.go +++ b/client/frontend/metric_client_gen.go @@ -275,6 +275,20 @@ func (c *metricClient) DescribeBatchOperation( return c.client.DescribeBatchOperation(ctx, request, opts...) } +func (c *metricClient) DescribeChannel( + ctx context.Context, + request *workflowservice.DescribeChannelRequest, + opts ...grpc.CallOption, +) (_ *workflowservice.DescribeChannelResponse, retError error) { + + metricsHandler, startTime := c.startMetricsRecording(ctx, "FrontendClientDescribeChannel") + defer func() { + c.finishMetricsRecording(metricsHandler, startTime, retError) + }() + + return c.client.DescribeChannel(ctx, request, opts...) +} + func (c *metricClient) DescribeDeployment( ctx context.Context, request *workflowservice.DescribeDeploymentRequest, @@ -793,6 +807,20 @@ func (c *metricClient) ListWorkflowRules( return c.client.ListWorkflowRules(ctx, request, opts...) } +func (c *metricClient) NotifyChannel( + ctx context.Context, + request *workflowservice.NotifyChannelRequest, + opts ...grpc.CallOption, +) (_ *workflowservice.NotifyChannelResponse, retError error) { + + metricsHandler, startTime := c.startMetricsRecording(ctx, "FrontendClientNotifyChannel") + defer func() { + c.finishMetricsRecording(metricsHandler, startTime, retError) + }() + + return c.client.NotifyChannel(ctx, request, opts...) +} + func (c *metricClient) PatchSchedule( ctx context.Context, request *workflowservice.PatchScheduleRequest, @@ -877,6 +905,20 @@ func (c *metricClient) PollActivityTaskQueue( return c.client.PollActivityTaskQueue(ctx, request, opts...) } +func (c *metricClient) PollChannel( + ctx context.Context, + request *workflowservice.PollChannelRequest, + opts ...grpc.CallOption, +) (_ *workflowservice.PollChannelResponse, retError error) { + + metricsHandler, startTime := c.startMetricsRecording(ctx, "FrontendClientPollChannel") + defer func() { + c.finishMetricsRecording(metricsHandler, startTime, retError) + }() + + return c.client.PollChannel(ctx, request, opts...) +} + func (c *metricClient) PollNexusOperationExecution( ctx context.Context, request *workflowservice.PollNexusOperationExecutionRequest, @@ -1003,6 +1045,20 @@ func (c *metricClient) RecordWorkerHeartbeat( return c.client.RecordWorkerHeartbeat(ctx, request, opts...) } +func (c *metricClient) RegisterChannelListener( + ctx context.Context, + request *workflowservice.RegisterChannelListenerRequest, + opts ...grpc.CallOption, +) (_ *workflowservice.RegisterChannelListenerResponse, retError error) { + + metricsHandler, startTime := c.startMetricsRecording(ctx, "FrontendClientRegisterChannelListener") + defer func() { + c.finishMetricsRecording(metricsHandler, startTime, retError) + }() + + return c.client.RegisterChannelListener(ctx, request, opts...) +} + func (c *metricClient) RegisterNamespace( ctx context.Context, request *workflowservice.RegisterNamespaceRequest, @@ -1549,6 +1605,20 @@ func (c *metricClient) UnpauseWorkflowExecution( return c.client.UnpauseWorkflowExecution(ctx, request, opts...) } +func (c *metricClient) UnregisterChannelListener( + ctx context.Context, + request *workflowservice.UnregisterChannelListenerRequest, + opts ...grpc.CallOption, +) (_ *workflowservice.UnregisterChannelListenerResponse, retError error) { + + metricsHandler, startTime := c.startMetricsRecording(ctx, "FrontendClientUnregisterChannelListener") + defer func() { + c.finishMetricsRecording(metricsHandler, startTime, retError) + }() + + return c.client.UnregisterChannelListener(ctx, request, opts...) +} + func (c *metricClient) UpdateActivityExecutionOptions( ctx context.Context, request *workflowservice.UpdateActivityExecutionOptionsRequest, diff --git a/client/frontend/retryable_client_gen.go b/client/frontend/retryable_client_gen.go index a8a09ef412f..fb53fe62aa4 100644 --- a/client/frontend/retryable_client_gen.go +++ b/client/frontend/retryable_client_gen.go @@ -296,6 +296,21 @@ func (c *retryableClient) DescribeBatchOperation( return resp, err } +func (c *retryableClient) DescribeChannel( + ctx context.Context, + request *workflowservice.DescribeChannelRequest, + opts ...grpc.CallOption, +) (*workflowservice.DescribeChannelResponse, error) { + var resp *workflowservice.DescribeChannelResponse + op := func(ctx context.Context) error { + var err error + resp, err = c.client.DescribeChannel(ctx, request, opts...) + return err + } + err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable) + return resp, err +} + func (c *retryableClient) DescribeDeployment( ctx context.Context, request *workflowservice.DescribeDeploymentRequest, @@ -851,6 +866,21 @@ func (c *retryableClient) ListWorkflowRules( return resp, err } +func (c *retryableClient) NotifyChannel( + ctx context.Context, + request *workflowservice.NotifyChannelRequest, + opts ...grpc.CallOption, +) (*workflowservice.NotifyChannelResponse, error) { + var resp *workflowservice.NotifyChannelResponse + op := func(ctx context.Context) error { + var err error + resp, err = c.client.NotifyChannel(ctx, request, opts...) + return err + } + err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable) + return resp, err +} + func (c *retryableClient) PatchSchedule( ctx context.Context, request *workflowservice.PatchScheduleRequest, @@ -941,6 +971,21 @@ func (c *retryableClient) PollActivityTaskQueue( return resp, err } +func (c *retryableClient) PollChannel( + ctx context.Context, + request *workflowservice.PollChannelRequest, + opts ...grpc.CallOption, +) (*workflowservice.PollChannelResponse, error) { + var resp *workflowservice.PollChannelResponse + op := func(ctx context.Context) error { + var err error + resp, err = c.client.PollChannel(ctx, request, opts...) + return err + } + err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable) + return resp, err +} + func (c *retryableClient) PollNexusOperationExecution( ctx context.Context, request *workflowservice.PollNexusOperationExecutionRequest, @@ -1076,6 +1121,21 @@ func (c *retryableClient) RecordWorkerHeartbeat( return resp, err } +func (c *retryableClient) RegisterChannelListener( + ctx context.Context, + request *workflowservice.RegisterChannelListenerRequest, + opts ...grpc.CallOption, +) (*workflowservice.RegisterChannelListenerResponse, error) { + var resp *workflowservice.RegisterChannelListenerResponse + op := func(ctx context.Context) error { + var err error + resp, err = c.client.RegisterChannelListener(ctx, request, opts...) + return err + } + err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable) + return resp, err +} + func (c *retryableClient) RegisterNamespace( ctx context.Context, request *workflowservice.RegisterNamespaceRequest, @@ -1661,6 +1721,21 @@ func (c *retryableClient) UnpauseWorkflowExecution( return resp, err } +func (c *retryableClient) UnregisterChannelListener( + ctx context.Context, + request *workflowservice.UnregisterChannelListenerRequest, + opts ...grpc.CallOption, +) (*workflowservice.UnregisterChannelListenerResponse, error) { + var resp *workflowservice.UnregisterChannelListenerResponse + op := func(ctx context.Context) error { + var err error + resp, err = c.client.UnregisterChannelListener(ctx, request, opts...) + return err + } + err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable) + return resp, err +} + func (c *retryableClient) UpdateActivityExecutionOptions( ctx context.Context, request *workflowservice.UpdateActivityExecutionOptionsRequest, diff --git a/cmd/tools/getproto/files.go b/cmd/tools/getproto/files.go index 4006de0ad41..160086f7b34 100644 --- a/cmd/tools/getproto/files.go +++ b/cmd/tools/getproto/files.go @@ -22,12 +22,14 @@ import ( namespace "go.temporal.io/api/namespace/v1" nexus "go.temporal.io/api/nexus/v1" nexusoperation "go.temporal.io/api/nexusoperation/v1" + notification "go.temporal.io/api/notification/v1" protocol "go.temporal.io/api/protocol/v1" query "go.temporal.io/api/query/v1" replication "go.temporal.io/api/replication/v1" rules "go.temporal.io/api/rules/v1" schedule "go.temporal.io/api/schedule/v1" sdk "go.temporal.io/api/sdk/v1" + stream "go.temporal.io/api/stream/v1" taskqueue "go.temporal.io/api/taskqueue/v1" update "go.temporal.io/api/update/v1" version "go.temporal.io/api/version/v1" @@ -84,6 +86,7 @@ func init() { importMap["temporal/api/namespace/v1/message.proto"] = namespace.File_temporal_api_namespace_v1_message_proto importMap["temporal/api/nexus/v1/message.proto"] = nexus.File_temporal_api_nexus_v1_message_proto importMap["temporal/api/nexusoperation/v1/message.proto"] = nexusoperation.File_temporal_api_nexusoperation_v1_message_proto + importMap["temporal/api/notification/v1/message.proto"] = notification.File_temporal_api_notification_v1_message_proto importMap["temporal/api/protocol/v1/message.proto"] = protocol.File_temporal_api_protocol_v1_message_proto importMap["temporal/api/query/v1/message.proto"] = query.File_temporal_api_query_v1_message_proto importMap["temporal/api/replication/v1/message.proto"] = replication.File_temporal_api_replication_v1_message_proto @@ -93,6 +96,7 @@ func init() { importMap["temporal/api/sdk/v1/task_complete_metadata.proto"] = sdk.File_temporal_api_sdk_v1_task_complete_metadata_proto importMap["temporal/api/sdk/v1/user_metadata.proto"] = sdk.File_temporal_api_sdk_v1_user_metadata_proto importMap["temporal/api/sdk/v1/worker_config.proto"] = sdk.File_temporal_api_sdk_v1_worker_config_proto + importMap["temporal/api/stream/v1/message.proto"] = stream.File_temporal_api_stream_v1_message_proto importMap["temporal/api/taskqueue/v1/message.proto"] = taskqueue.File_temporal_api_taskqueue_v1_message_proto importMap["temporal/api/update/v1/message.proto"] = update.File_temporal_api_update_v1_message_proto importMap["temporal/api/version/v1/message.proto"] = version.File_temporal_api_version_v1_message_proto diff --git a/common/api/metadata.go b/common/api/metadata.go index 41843262284..08c52b72869 100644 --- a/common/api/metadata.go +++ b/common/api/metadata.go @@ -109,6 +109,11 @@ var ( "RespondNexusTaskFailed": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "RequestCancelWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "SignalWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, + "NotifyChannel": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, + "RegisterChannelListener": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, + "UnregisterChannelListener": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, + "PollChannel": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingCapable}, + "DescribeChannel": {Scope: ScopeNamespace, Access: AccessReadOnly, Polling: PollingNone}, "SignalWithStartWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "ResetWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, "TerminateWorkflowExecution": {Scope: ScopeNamespace, Access: AccessWrite, Polling: PollingNone}, diff --git a/common/metrics/otel_metrics_handler_test.go b/common/metrics/otel_metrics_handler_test.go index a2d0b34eb53..bf08a9460e1 100644 --- a/common/metrics/otel_metrics_handler_test.go +++ b/common/metrics/otel_metrics_handler_test.go @@ -193,9 +193,14 @@ func TestMeter(t *testing.T) { cmpopts.SortSlices(func(x, y metricdata.Metrics) bool { return x.Name < y.Name }), - cmpopts.IgnoreFields(metricdata.DataPoint[int64]{}, "StartTime", "Time"), - cmpopts.IgnoreFields(metricdata.DataPoint[float64]{}, "StartTime", "Time"), - cmpopts.IgnoreFields(metricdata.HistogramDataPoint[int64]{}, "StartTime", "Time", "Bounds"), + // Exemplars are ignored rather than equated empty: the SDK hands back + // either a nil or an allocated-but-empty slice depending on whether + // its reservoir was ever touched, and equating empty globally would + // also hide a nil-versus-empty attribute set or data point. + cmpopts.IgnoreFields(metricdata.DataPoint[int64]{}, "StartTime", "Time", "Exemplars"), + cmpopts.IgnoreFields(metricdata.DataPoint[float64]{}, "StartTime", "Time", "Exemplars"), + cmpopts.IgnoreFields( + metricdata.HistogramDataPoint[int64]{}, "StartTime", "Time", "Bounds", "Exemplars"), ); diff != "" { t.Errorf("mismatch (-want, +got):\n%s", diff) } @@ -264,7 +269,8 @@ func TestMeter_TimerInSeconds(t *testing.T) { cmp.Comparer(func(a1, a2 attribute.Set) bool { return a1.Equals(&a2) }), - cmpopts.IgnoreFields(metricdata.HistogramDataPoint[float64]{}, "StartTime", "Time", "Bounds"), + cmpopts.IgnoreFields( + metricdata.HistogramDataPoint[float64]{}, "StartTime", "Time", "Bounds", "Exemplars"), ); diff != "" { t.Errorf("mismatch (-want, +got):\n%s", diff) } diff --git a/common/persistence/client/quotas.go b/common/persistence/client/quotas.go index 7c8d89a48e9..962aa8e2e4f 100644 --- a/common/persistence/client/quotas.go +++ b/common/persistence/client/quotas.go @@ -29,6 +29,9 @@ var ( "StartWorkflowExecution": 1, "SignalWithStartWorkflowExecution": 1, "SignalWorkflowExecution": 1, + "NotifyChannel": 1, + "RegisterChannelListener": 1, + "UnregisterChannelListener": 1, "RequestCancelWorkflowExecution": 1, "TerminateWorkflowExecution": 1, "GetWorkflowExecutionHistory": 1, diff --git a/common/rpc/interceptor/dc_redirection_policy.go b/common/rpc/interceptor/dc_redirection_policy.go index c7e09738f71..171abde9674 100644 --- a/common/rpc/interceptor/dc_redirection_policy.go +++ b/common/rpc/interceptor/dc_redirection_policy.go @@ -55,6 +55,9 @@ var selectedAPIsForwardingRedirectionPolicyWhitelistedAPIs = map[string]struct{} "StartWorkflowExecution": {}, "SignalWithStartWorkflowExecution": {}, "SignalWorkflowExecution": {}, + "NotifyChannel": {}, + "RegisterChannelListener": {}, + "UnregisterChannelListener": {}, "UpdateWorkflowExecution": {}, "RequestCancelWorkflowExecution": {}, "TerminateWorkflowExecution": {}, diff --git a/common/rpc/interceptor/dc_redirection_policy_test.go b/common/rpc/interceptor/dc_redirection_policy_test.go index 41169da45a7..bc970b1eece 100644 --- a/common/rpc/interceptor/dc_redirection_policy_test.go +++ b/common/rpc/interceptor/dc_redirection_policy_test.go @@ -129,6 +129,9 @@ func (s *selectedAPIsForwardingRedirectionPolicySuite) TestSelectedAPIs() { "StartWorkflowExecution": {}, "SignalWithStartWorkflowExecution": {}, "SignalWorkflowExecution": {}, + "NotifyChannel": {}, + "RegisterChannelListener": {}, + "UnregisterChannelListener": {}, "UpdateWorkflowExecution": {}, "RequestCancelWorkflowExecution": {}, "TerminateWorkflowExecution": {}, diff --git a/common/rpc/interceptor/logtags/workflow_service_server_gen.go b/common/rpc/interceptor/logtags/workflow_service_server_gen.go index 8de3d6ad4ef..4e95bddb173 100644 --- a/common/rpc/interceptor/logtags/workflow_service_server_gen.go +++ b/common/rpc/interceptor/logtags/workflow_service_server_gen.go @@ -99,6 +99,10 @@ func (wt *WorkflowTags) extractFromWorkflowServiceServerMessage(message any) []t return nil case *workflowservice.DescribeBatchOperationResponse: return nil + case *workflowservice.DescribeChannelRequest: + return nil + case *workflowservice.DescribeChannelResponse: + return nil case *workflowservice.DescribeDeploymentRequest: return nil case *workflowservice.DescribeDeploymentResponse: @@ -261,6 +265,10 @@ func (wt *WorkflowTags) extractFromWorkflowServiceServerMessage(message any) []t return nil case *workflowservice.ListWorkflowRulesResponse: return nil + case *workflowservice.NotifyChannelRequest: + return nil + case *workflowservice.NotifyChannelResponse: + return nil case *workflowservice.PatchScheduleRequest: return nil case *workflowservice.PatchScheduleResponse: @@ -303,6 +311,10 @@ func (wt *WorkflowTags) extractFromWorkflowServiceServerMessage(message any) []t tag.WorkflowID(r.GetWorkflowExecution().GetWorkflowId()), tag.WorkflowRunID(r.GetWorkflowExecution().GetRunId()), } + case *workflowservice.PollChannelRequest: + return nil + case *workflowservice.PollChannelResponse: + return nil case *workflowservice.PollNexusOperationExecutionRequest: return []tag.Tag{ tag.OperationID(r.GetOperationId()), @@ -363,6 +375,10 @@ func (wt *WorkflowTags) extractFromWorkflowServiceServerMessage(message any) []t return nil case *workflowservice.RecordWorkerHeartbeatResponse: return nil + case *workflowservice.RegisterChannelListenerRequest: + return nil + case *workflowservice.RegisterChannelListenerResponse: + return nil case *workflowservice.RegisterNamespaceRequest: return nil case *workflowservice.RegisterNamespaceResponse: @@ -596,6 +612,10 @@ func (wt *WorkflowTags) extractFromWorkflowServiceServerMessage(message any) []t } case *workflowservice.UnpauseWorkflowExecutionResponse: return nil + case *workflowservice.UnregisterChannelListenerRequest: + return nil + case *workflowservice.UnregisterChannelListenerResponse: + return nil case *workflowservice.UpdateActivityExecutionOptionsRequest: return []tag.Tag{ tag.WorkflowID(r.GetWorkflowId()), diff --git a/common/rpc/interceptor/redirection.go b/common/rpc/interceptor/redirection.go index 80a1cf01423..9c62519efa4 100644 --- a/common/rpc/interceptor/redirection.go +++ b/common/rpc/interceptor/redirection.go @@ -79,6 +79,11 @@ var ( "RespondNexusTaskFailed": func() any { return &workflowservice.RespondNexusTaskFailedResponse{} }, "SignalWithStartWorkflowExecution": func() any { return &workflowservice.SignalWithStartWorkflowExecutionResponse{} }, "SignalWorkflowExecution": func() any { return &workflowservice.SignalWorkflowExecutionResponse{} }, + "NotifyChannel": func() any { return &workflowservice.NotifyChannelResponse{} }, + "RegisterChannelListener": func() any { return &workflowservice.RegisterChannelListenerResponse{} }, + "UnregisterChannelListener": func() any { return &workflowservice.UnregisterChannelListenerResponse{} }, + "PollChannel": func() any { return &workflowservice.PollChannelResponse{} }, + "DescribeChannel": func() any { return &workflowservice.DescribeChannelResponse{} }, "StartWorkflowExecution": func() any { return &workflowservice.StartWorkflowExecutionResponse{} }, "ExecuteMultiOperation": func() any { return &workflowservice.ExecuteMultiOperationResponse{} }, "UpdateWorkflowExecution": func() any { return &workflowservice.UpdateWorkflowExecutionResponse{} }, diff --git a/common/rpc/interceptor/redirection_test.go b/common/rpc/interceptor/redirection_test.go index 11e6df3a681..c74ec389340 100644 --- a/common/rpc/interceptor/redirection_test.go +++ b/common/rpc/interceptor/redirection_test.go @@ -137,6 +137,11 @@ func (s *redirectionInterceptorSuite) TestGlobalAPI() { "RespondNexusTaskFailed": {}, "SignalWithStartWorkflowExecution": {}, "SignalWorkflowExecution": {}, + "NotifyChannel": {}, + "RegisterChannelListener": {}, + "UnregisterChannelListener": {}, + "PollChannel": {}, + "DescribeChannel": {}, "StartWorkflowExecution": {}, "ExecuteMultiOperation": {}, "UpdateWorkflowExecution": {}, diff --git a/common/rpc/interceptor/telemetry.go b/common/rpc/interceptor/telemetry.go index f906da64032..98b85b8441c 100644 --- a/common/rpc/interceptor/telemetry.go +++ b/common/rpc/interceptor/telemetry.go @@ -70,6 +70,9 @@ var ( "RecordActivityTaskHeartbeatById": {}, "ResetWorkflowExecution": {}, "SignalWorkflowExecution": {}, + "NotifyChannel": {}, + "RegisterChannelListener": {}, + "UnregisterChannelListener": {}, "SignalWithStartWorkflowExecution": {}, "CreateSchedule": {}, "UpdateSchedule": {}, diff --git a/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go b/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go index 7193985a28a..04291d5abde 100644 --- a/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go +++ b/common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go @@ -422,6 +422,26 @@ func (mr *MockWorkflowServiceClientMockRecorder) DescribeBatchOperation(ctx, in return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DescribeBatchOperation", reflect.TypeOf((*MockWorkflowServiceClient)(nil).DescribeBatchOperation), varargs...) } +// DescribeChannel mocks base method. +func (m *MockWorkflowServiceClient) DescribeChannel(ctx context.Context, in *workflowservice.DescribeChannelRequest, opts ...grpc.CallOption) (*workflowservice.DescribeChannelResponse, error) { + m.ctrl.T.Helper() + varargs := []any{ctx, in} + for _, a := range opts { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "DescribeChannel", varargs...) + ret0, _ := ret[0].(*workflowservice.DescribeChannelResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// DescribeChannel indicates an expected call of DescribeChannel. +func (mr *MockWorkflowServiceClientMockRecorder) DescribeChannel(ctx, in any, opts ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, in}, opts...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DescribeChannel", reflect.TypeOf((*MockWorkflowServiceClient)(nil).DescribeChannel), varargs...) +} + // DescribeDeployment mocks base method. func (m *MockWorkflowServiceClient) DescribeDeployment(ctx context.Context, in *workflowservice.DescribeDeploymentRequest, opts ...grpc.CallOption) (*workflowservice.DescribeDeploymentResponse, error) { m.ctrl.T.Helper() @@ -1162,6 +1182,26 @@ func (mr *MockWorkflowServiceClientMockRecorder) ListWorkflowRules(ctx, in any, return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListWorkflowRules", reflect.TypeOf((*MockWorkflowServiceClient)(nil).ListWorkflowRules), varargs...) } +// NotifyChannel mocks base method. +func (m *MockWorkflowServiceClient) NotifyChannel(ctx context.Context, in *workflowservice.NotifyChannelRequest, opts ...grpc.CallOption) (*workflowservice.NotifyChannelResponse, error) { + m.ctrl.T.Helper() + varargs := []any{ctx, in} + for _, a := range opts { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "NotifyChannel", varargs...) + ret0, _ := ret[0].(*workflowservice.NotifyChannelResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// NotifyChannel indicates an expected call of NotifyChannel. +func (mr *MockWorkflowServiceClientMockRecorder) NotifyChannel(ctx, in any, opts ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, in}, opts...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NotifyChannel", reflect.TypeOf((*MockWorkflowServiceClient)(nil).NotifyChannel), varargs...) +} + // PatchSchedule mocks base method. func (m *MockWorkflowServiceClient) PatchSchedule(ctx context.Context, in *workflowservice.PatchScheduleRequest, opts ...grpc.CallOption) (*workflowservice.PatchScheduleResponse, error) { m.ctrl.T.Helper() @@ -1282,6 +1322,26 @@ func (mr *MockWorkflowServiceClientMockRecorder) PollActivityTaskQueue(ctx, in a return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PollActivityTaskQueue", reflect.TypeOf((*MockWorkflowServiceClient)(nil).PollActivityTaskQueue), varargs...) } +// PollChannel mocks base method. +func (m *MockWorkflowServiceClient) PollChannel(ctx context.Context, in *workflowservice.PollChannelRequest, opts ...grpc.CallOption) (*workflowservice.PollChannelResponse, error) { + m.ctrl.T.Helper() + varargs := []any{ctx, in} + for _, a := range opts { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "PollChannel", varargs...) + ret0, _ := ret[0].(*workflowservice.PollChannelResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// PollChannel indicates an expected call of PollChannel. +func (mr *MockWorkflowServiceClientMockRecorder) PollChannel(ctx, in any, opts ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, in}, opts...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PollChannel", reflect.TypeOf((*MockWorkflowServiceClient)(nil).PollChannel), varargs...) +} + // PollNexusOperationExecution mocks base method. func (m *MockWorkflowServiceClient) PollNexusOperationExecution(ctx context.Context, in *workflowservice.PollNexusOperationExecutionRequest, opts ...grpc.CallOption) (*workflowservice.PollNexusOperationExecutionResponse, error) { m.ctrl.T.Helper() @@ -1462,6 +1522,26 @@ func (mr *MockWorkflowServiceClientMockRecorder) RecordWorkerHeartbeat(ctx, in a return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordWorkerHeartbeat", reflect.TypeOf((*MockWorkflowServiceClient)(nil).RecordWorkerHeartbeat), varargs...) } +// RegisterChannelListener mocks base method. +func (m *MockWorkflowServiceClient) RegisterChannelListener(ctx context.Context, in *workflowservice.RegisterChannelListenerRequest, opts ...grpc.CallOption) (*workflowservice.RegisterChannelListenerResponse, error) { + m.ctrl.T.Helper() + varargs := []any{ctx, in} + for _, a := range opts { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "RegisterChannelListener", varargs...) + ret0, _ := ret[0].(*workflowservice.RegisterChannelListenerResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// RegisterChannelListener indicates an expected call of RegisterChannelListener. +func (mr *MockWorkflowServiceClientMockRecorder) RegisterChannelListener(ctx, in any, opts ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, in}, opts...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterChannelListener", reflect.TypeOf((*MockWorkflowServiceClient)(nil).RegisterChannelListener), varargs...) +} + // RegisterNamespace mocks base method. func (m *MockWorkflowServiceClient) RegisterNamespace(ctx context.Context, in *workflowservice.RegisterNamespaceRequest, opts ...grpc.CallOption) (*workflowservice.RegisterNamespaceResponse, error) { m.ctrl.T.Helper() @@ -2242,6 +2322,26 @@ func (mr *MockWorkflowServiceClientMockRecorder) UnpauseWorkflowExecution(ctx, i return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UnpauseWorkflowExecution", reflect.TypeOf((*MockWorkflowServiceClient)(nil).UnpauseWorkflowExecution), varargs...) } +// UnregisterChannelListener mocks base method. +func (m *MockWorkflowServiceClient) UnregisterChannelListener(ctx context.Context, in *workflowservice.UnregisterChannelListenerRequest, opts ...grpc.CallOption) (*workflowservice.UnregisterChannelListenerResponse, error) { + m.ctrl.T.Helper() + varargs := []any{ctx, in} + for _, a := range opts { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "UnregisterChannelListener", varargs...) + ret0, _ := ret[0].(*workflowservice.UnregisterChannelListenerResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// UnregisterChannelListener indicates an expected call of UnregisterChannelListener. +func (mr *MockWorkflowServiceClientMockRecorder) UnregisterChannelListener(ctx, in any, opts ...any) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]any{ctx, in}, opts...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UnregisterChannelListener", reflect.TypeOf((*MockWorkflowServiceClient)(nil).UnregisterChannelListener), varargs...) +} + // UpdateActivityExecutionOptions mocks base method. func (m *MockWorkflowServiceClient) UpdateActivityExecutionOptions(ctx context.Context, in *workflowservice.UpdateActivityExecutionOptionsRequest, opts ...grpc.CallOption) (*workflowservice.UpdateActivityExecutionOptionsResponse, error) { m.ctrl.T.Helper() diff --git a/go.mod b/go.mod index b9c7674a01d..e641ddac975 100644 --- a/go.mod +++ b/go.mod @@ -236,3 +236,5 @@ require ( ) tool golang.org/x/perf/cmd/benchstat + +replace go.temporal.io/api => github.com/moedash/api-go v1.63.6-0.20261003021656-71792676a9fd diff --git a/go.sum b/go.sum index ec1c673c1b4..c73f4cefa6d 100644 --- a/go.sum +++ b/go.sum @@ -321,6 +321,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/moedash/api-go v1.63.6-0.20261003021656-71792676a9fd h1:40ANlQLRwkA2/uY3o7U7v8kElCSclu55fA/RtRTyfQk= +github.com/moedash/api-go v1.63.6-0.20261003021656-71792676a9fd/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= @@ -481,8 +483,6 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= -go.temporal.io/api v1.63.6-0.20260909222256-20151aa90480 h1:1w2xEZFpL7R9NagEQeWTh7oZmZU3YUJGogusn9/LfCg= -go.temporal.io/api v1.63.6-0.20260909222256-20151aa90480/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab h1:99wXW0317BBi49d6xgMdA0EZtvA+xbBUWV4HsTEGEcg= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw= go.temporal.io/sdk v1.48.0 h1:WDctKDVuh0Z8Nf7euAyqs/EwcPg1JTIIq1Fut8Tq118= diff --git a/service/frontend/configs/quotas.go b/service/frontend/configs/quotas.go index 1f024759b23..0d59b57446b 100644 --- a/service/frontend/configs/quotas.go +++ b/service/frontend/configs/quotas.go @@ -43,6 +43,7 @@ var ( "/temporal.api.workflowservice.v1.WorkflowService/PollNexusTaskQueue": 1, "/temporal.api.workflowservice.v1.WorkflowService/PollNexusOperationExecution": 1, "/temporal.api.workflowservice.v1.WorkflowService/PollWorkflowExecutionTimeSkipping": 1, + "/temporal.api.workflowservice.v1.WorkflowService/PollChannel": 1, // Long-running if activity outcome is not already available "/temporal.api.workflowservice.v1.WorkflowService/PollActivityExecution": 1, @@ -87,6 +88,9 @@ var ( // P1: External Event APIs "/temporal.api.workflowservice.v1.WorkflowService/SignalWorkflowExecution": 1, + "/temporal.api.workflowservice.v1.WorkflowService/NotifyChannel": 1, + "/temporal.api.workflowservice.v1.WorkflowService/RegisterChannelListener": 1, + "/temporal.api.workflowservice.v1.WorkflowService/UnregisterChannelListener": 1, "/temporal.api.workflowservice.v1.WorkflowService/SignalWithStartWorkflowExecution": 1, "/temporal.api.workflowservice.v1.WorkflowService/StartWorkflowExecution": 1, "/temporal.api.workflowservice.v1.WorkflowService/UpdateWorkflowExecution": 1, @@ -156,6 +160,7 @@ var ( // P3: Status Querying APIs "/temporal.api.workflowservice.v1.WorkflowService/DescribeWorkflowExecution": 3, + "/temporal.api.workflowservice.v1.WorkflowService/DescribeChannel": 3, "/temporal.api.workflowservice.v1.WorkflowService/DescribeActivityExecution": 3, "/temporal.api.workflowservice.v1.WorkflowService/DescribeTaskQueue": 3, "/temporal.api.workflowservice.v1.WorkflowService/GetWorkerBuildIdCompatibility": 3, @@ -188,6 +193,7 @@ var ( "/temporal.api.workflowservice.v1.WorkflowService/PollNexusOperationExecution": 4, "/temporal.api.workflowservice.v1.WorkflowService/PollActivityExecution": 4, // TODO(saa-preview): should it be 4 or 3? "/temporal.api.workflowservice.v1.WorkflowService/PollWorkflowTaskQueue": 4, + "/temporal.api.workflowservice.v1.WorkflowService/PollChannel": 4, "/temporal.api.workflowservice.v1.WorkflowService/PollActivityTaskQueue": 4, "/temporal.api.workflowservice.v1.WorkflowService/PollWorkflowExecutionUpdate": 4, "/temporal.api.workflowservice.v1.WorkflowService/PollNexusTaskQueue": 4, diff --git a/service/frontend/workflow_handler.go b/service/frontend/workflow_handler.go index 0abdd0caba0..b8cf8d22509 100644 --- a/service/frontend/workflow_handler.go +++ b/service/frontend/workflow_handler.go @@ -2359,6 +2359,51 @@ func (wh *WorkflowHandler) SignalWorkflowExecution(ctx context.Context, request }, nil } +// NotifyChannel is declared by the pinned API and served by a later layer of +// this series. +func (wh *WorkflowHandler) NotifyChannel( + _ context.Context, + _ *workflowservice.NotifyChannelRequest, +) (*workflowservice.NotifyChannelResponse, error) { + return nil, serviceerror.NewUnimplemented("NotifyChannel is not implemented") +} + +// RegisterChannelListener is declared by the pinned API and served by a later +// layer of this series. +func (wh *WorkflowHandler) RegisterChannelListener( + _ context.Context, + _ *workflowservice.RegisterChannelListenerRequest, +) (*workflowservice.RegisterChannelListenerResponse, error) { + return nil, serviceerror.NewUnimplemented("RegisterChannelListener is not implemented") +} + +// UnregisterChannelListener is declared by the pinned API and served by a +// later layer of this series. +func (wh *WorkflowHandler) UnregisterChannelListener( + _ context.Context, + _ *workflowservice.UnregisterChannelListenerRequest, +) (*workflowservice.UnregisterChannelListenerResponse, error) { + return nil, serviceerror.NewUnimplemented("UnregisterChannelListener is not implemented") +} + +// PollChannel is declared by the pinned API and served by a later layer of +// this series. +func (wh *WorkflowHandler) PollChannel( + _ context.Context, + _ *workflowservice.PollChannelRequest, +) (*workflowservice.PollChannelResponse, error) { + return nil, serviceerror.NewUnimplemented("PollChannel is not implemented") +} + +// DescribeChannel is declared by the pinned API and served by a later layer of +// this series. +func (wh *WorkflowHandler) DescribeChannel( + _ context.Context, + _ *workflowservice.DescribeChannelRequest, +) (*workflowservice.DescribeChannelResponse, error) { + return nil, serviceerror.NewUnimplemented("DescribeChannel is not implemented") +} + // SignalWithStartWorkflowExecution is used to ensure sending signal to a workflow. // If the workflow is running, this results in WorkflowExecutionSignaled event being recorded in the history // and a workflow task being created for the execution. diff --git a/service/history/historybuilder/event_store.go b/service/history/historybuilder/event_store.go index a6fb18a4b9e..dc080fe96d6 100644 --- a/service/history/historybuilder/event_store.go +++ b/service/history/historybuilder/event_store.go @@ -331,7 +331,11 @@ func (b *EventStore) bufferEvent( enumspb.EVENT_TYPE_UPSERT_WORKFLOW_SEARCH_ATTRIBUTES, enumspb.EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, - enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED: + enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED, + enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED, + enumspb.EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED, + enumspb.EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED, + enumspb.EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED: // do not buffer event if event is directly generated from a corresponding command return false diff --git a/service/history/historybuilder/history_builder_test.go b/service/history/historybuilder/history_builder_test.go index d7b13d5fd6c..3b5835e18a5 100644 --- a/service/history/historybuilder/history_builder_test.go +++ b/service/history/historybuilder/history_builder_test.go @@ -2272,6 +2272,10 @@ func (s *historyBuilderSuite) TestBufferEvent() { enumspb.EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED: true, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED: true, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED: true, + enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED: true, + enumspb.EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED: true, + enumspb.EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED: true, + enumspb.EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED: true, } // events corresponding to message from client will be assigned an event ID immediately diff --git a/tests/api_fork_test.go b/tests/api_fork_test.go new file mode 100644 index 00000000000..c4e092ed005 --- /dev/null +++ b/tests/api_fork_test.go @@ -0,0 +1,59 @@ +package tests + +import ( + "testing" + + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + streampb "go.temporal.io/api/stream/v1" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/protobuf/proto" +) + +// The api fork has to round-trip the new shapes over the wire, not merely +// compile. A field added without its descriptor would compile and silently +// drop on marshal. +func TestApiForkCarriesStreamShapes(t *testing.T) { + require.Equal(t, enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, enumspb.CommandType(19)) + + cmd := &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + StreamName: "s1", + Records: []*streampb.StreamRecord{{Topic: "tokens"}}, + }, + }, + } + b, err := proto.Marshal(cmd) + require.NoError(t, err) + var back commandpb.Command + require.NoError(t, proto.Unmarshal(b, &back)) + require.Equal(t, "s1", back.GetAppendStreamRecordsCommandAttributes().GetStreamName()) + require.Equal(t, "tokens", + back.GetAppendStreamRecordsCommandAttributes().GetRecords()[0].GetTopic()) + + resp := &workflowservice.PollWorkflowTaskQueueResponse{ + StreamSlices: []*streampb.StreamSlice{{StreamId: "s1", FromOffset: 4, ToOffset: 7}}, + } + rb, err := proto.Marshal(resp) + require.NoError(t, err) + var rback workflowservice.PollWorkflowTaskQueueResponse + require.NoError(t, proto.Unmarshal(rb, &rback)) + require.Equal(t, int64(7), rback.GetStreamSlices()[0].GetToOffset()) + + attrs := &historypb.WorkflowTaskCompletedEventAttributes{ + ConsumedStreamRanges: []*streampb.StreamRange{{StreamId: "s1", FromOffset: 4, ToOffset: 4}}, + } + ab, err := proto.Marshal(attrs) + require.NoError(t, err) + var aback historypb.WorkflowTaskCompletedEventAttributes + require.NoError(t, proto.Unmarshal(ab, &aback)) + // An empty range has to survive the round trip: it is the fact that a + // subscription observed nothing, which replay must reproduce. + require.Len(t, aback.GetConsumedStreamRanges(), 1) + require.Equal(t, + aback.GetConsumedStreamRanges()[0].GetFromOffset(), aback.GetConsumedStreamRanges()[0].GetToOffset()) +}