From 8125995d1ad9c8b02d519c3d99bb25579051ba55 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Sat, 3 Oct 2026 01:13:45 -0700 Subject: [PATCH 1/3] Pinned the api fork that declares the notification channel API. The pin adds five WorkflowService RPCs, so the generated clients, interceptors and mock follow, and the frontend answers the RPCs with Unimplemented until the channel service serves them. --- client/frontend/client_gen.go | 50 +++++++++ client/frontend/metric_client_gen.go | 70 ++++++++++++ client/frontend/retryable_client_gen.go | 75 +++++++++++++ cmd/tools/getproto/files.go | 2 + .../logtags/workflow_service_server_gen.go | 20 ++++ .../interceptor/routing_key_extractor_gen.go | 10 ++ .../v1/service_grpc.pb.mock.go | 100 ++++++++++++++++++ go.mod | 2 + go.sum | 4 +- service/frontend/workflow_handler.go | 45 ++++++++ 10 files changed, 376 insertions(+), 2 deletions(-) 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/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/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/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/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. From 57c698431ec60e86e7a9e29d70755439133fa1cc Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Sat, 3 Oct 2026 01:14:25 -0700 Subject: [PATCH 2/3] Declared authorization, quotas and redirection for the channel RPCs. Every WorkflowService method needs a row in these tables, and the tests that walk the service fail on a method without one. Channel writes are treated like a Signal and the long poll like the other polls. --- common/api/metadata.go | 5 +++++ common/persistence/client/quotas.go | 3 +++ common/rpc/interceptor/dc_redirection_policy.go | 3 +++ common/rpc/interceptor/dc_redirection_policy_test.go | 3 +++ common/rpc/interceptor/redirection.go | 5 +++++ common/rpc/interceptor/redirection_test.go | 5 +++++ common/rpc/interceptor/telemetry.go | 3 +++ service/frontend/configs/quotas.go | 6 ++++++ 8 files changed, 33 insertions(+) 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/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/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, From b1e3c5e8a44878ee6f7dddd1a6a93ad2ef28035d Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Sat, 3 Oct 2026 01:14:30 -0700 Subject: [PATCH 3/3] Kept the two channel command events out of the event buffer. An event a command writes is never buffered, and the buffer test walks every event type the pinned API declares. --- service/history/historybuilder/event_store.go | 4 +++- service/history/historybuilder/history_builder_test.go | 5 ++++- 2 files changed, 7 insertions(+), 2 deletions(-) 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++