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..820e400ca5b 100644 --- a/cmd/tools/getproto/files.go +++ b/cmd/tools/getproto/files.go @@ -22,6 +22,7 @@ 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" @@ -84,6 +85,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 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/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/routing_key_extractor_gen.go b/common/rpc/interceptor/routing_key_extractor_gen.go index 0edb8074fe5..4560d2e3029 100644 --- a/common/rpc/interceptor/routing_key_extractor_gen.go +++ b/common/rpc/interceptor/routing_key_extractor_gen.go @@ -26,6 +26,8 @@ func workflowServiceRequestRoutingKey(req any) namespace.RoutingKey { return namespace.RoutingKey{ID: r.GetActivityId()} case *workflowservice.DescribeBatchOperationRequest: return namespace.RoutingKey{ID: r.GetJobId()} + case *workflowservice.DescribeChannelRequest: + return namespace.RoutingKey{ID: r.GetChannel()} case *workflowservice.DescribeScheduleRequest: return namespace.RoutingKey{ID: r.GetScheduleId()} case *workflowservice.DescribeTaskQueueRequest: @@ -50,6 +52,8 @@ func workflowServiceRequestRoutingKey(req any) namespace.RoutingKey { return namespace.RoutingKey{ID: r.GetScheduleId()} case *workflowservice.ListTaskQueuePartitionsRequest: return namespace.RoutingKey{ID: r.GetTaskQueue().GetName()} + case *workflowservice.NotifyChannelRequest: + return namespace.RoutingKey{ID: r.GetNotification().GetChannel()} case *workflowservice.PatchScheduleRequest: return namespace.RoutingKey{ID: r.GetScheduleId()} case *workflowservice.PauseActivityExecutionRequest: @@ -62,6 +66,8 @@ func workflowServiceRequestRoutingKey(req any) namespace.RoutingKey { return namespace.RoutingKey{ID: r.GetActivityId()} case *workflowservice.PollActivityTaskQueueRequest: return namespace.RoutingKey{ID: r.GetPollerGroupId(), Strategy: namespace.RoutingStrategyPollerGroup} + case *workflowservice.PollChannelRequest: + return namespace.RoutingKey{ID: r.GetChannel()} case *workflowservice.PollNexusTaskQueueRequest: return namespace.RoutingKey{ID: r.GetPollerGroupId(), Strategy: namespace.RoutingStrategyPollerGroup} case *workflowservice.PollWorkflowExecutionTimeSkippingRequest: @@ -78,6 +84,8 @@ func workflowServiceRequestRoutingKey(req any) namespace.RoutingKey { return namespace.RoutingKey{ID: routingIDFromResourceID(r.GetResourceId())} case *workflowservice.RecordWorkerHeartbeatRequest: return namespace.RoutingKey{ID: routingIDFromResourceID(r.GetResourceId())} + case *workflowservice.RegisterChannelListenerRequest: + return namespace.RoutingKey{ID: r.GetChannel()} case *workflowservice.RequestCancelActivityExecutionRequest: return namespace.RoutingKey{ID: r.GetActivityId()} case *workflowservice.RequestCancelWorkflowExecutionRequest: @@ -142,6 +150,8 @@ func workflowServiceRequestRoutingKey(req any) namespace.RoutingKey { return namespace.RoutingKey{ID: r.GetExecution().GetWorkflowId()} case *workflowservice.UnpauseWorkflowExecutionRequest: return namespace.RoutingKey{ID: r.GetWorkflowId()} + case *workflowservice.UnregisterChannelListenerRequest: + return namespace.RoutingKey{ID: r.GetChannel()} case *workflowservice.UpdateActivityExecutionOptionsRequest: return namespace.RoutingKey{ID: routingIDFromResourceID(r.GetResourceId())} case *workflowservice.UpdateActivityOptionsRequest: 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..9b16af256b8 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.20261003080812-cc8cb1b67d9a diff --git a/go.sum b/go.sum index ec1c673c1b4..d6a6a05c696 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.20261003080812-cc8cb1b67d9a h1:3ewglu5lF3kSjC+Hc7vI53mMyHfxut+94CE5aOpCKVI= +github.com/moedash/api-go v1.63.6-0.20261003080812-cc8cb1b67d9a/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..07720afa5c2 100644 --- a/service/history/historybuilder/event_store.go +++ b/service/history/historybuilder/event_store.go @@ -331,7 +331,9 @@ 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_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..053ed987afd 100644 --- a/service/history/historybuilder/history_builder_test.go +++ b/service/history/historybuilder/history_builder_test.go @@ -2272,6 +2272,8 @@ 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_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 @@ -2321,7 +2323,8 @@ func (s *historyBuilderSuite) TestBufferEvent() { commandType := enumspb.CommandType(ct) // Unspecified is not counted. // ProtocolMessage command doesn't have corresponding event. - if commandType == enumspb.COMMAND_TYPE_UNSPECIFIED || commandType == enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE { + if commandType == enumspb.COMMAND_TYPE_UNSPECIFIED || + commandType == enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE { continue } commandsWithEventsCount++