[CHASM] Split and wire up engine and visibilityMgr request interceptors (#8804)

## What changed?
- The `ChasmRequestInterceptor` has been split in two!
`ChasmRequestEngineInterceptor` does everything except inject
`chasm.VisibilityManager`, which is done by
`ChasmRequestVisibilityInterceptor`.
- Note that the test doesn't include a matching `hasVisibilityCtx`
assert to `hasEngineCtx` because I think making that getter public would
make the public API earnestly more confusing for consumers

## Why?
- Frontend needs visibilityMgr injected so that it can handle FE->Vis
queries without hopping through history. The previous interceptor
assumed `chasm.Engine` was always available, which it isn't, for
history. I'd also workshopped making engine an optional fx parameter,
but eh.

## How did you test it?
- Validated with scheduler functional tests, which use
`chasm.ListExecutions` directly in frontend's `WorkflowHandler`.

chasm/interceptor_test.go

@@ -43,20 +43,35 @@ func (l *ServiceLibrary) RegisterServices(server *grpc.Server) {
 	testspb.RegisterTestServiceServer(server, ServiceHandler{})
 }
 
-func TestChasmRequestInterceptor_ShouldRespond(t *testing.T) {
+func TestChasmEngineInterceptor_ShouldRespond(t *testing.T) {
 	ctrl := gomock.NewController(t)
+
 	mockEngine := chasm.NewMockEngine(ctrl)
-	mockVisibilityManager := chasm.NewMockVisibilityManager(ctrl)
-	requestInterceptor := chasm.ChasmRequestInterceptorProvider(
+	engineInterceptor := chasm.ChasmEngineInterceptorProvider(
 		mockEngine,
-		mockVisibilityManager,
 		log.NewNoopLogger(),
 		metrics.NoopMetricsHandler,
 	)
 
-	server, address := startTestServer(t, grpc.UnaryInterceptor(requestInterceptor.Intercept))
+	server, address := startTestServer(t, grpc.UnaryInterceptor(engineInterceptor.Intercept))
 	defer server.Stop()
 
+	response := testRoundTrip(t, address)
+	require.True(t, response.HasEngineCtx)
+}
+
+func TestChasmVisibilityInterceptor_ShouldRespond(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	mockVisibilityManager := chasm.NewMockVisibilityManager(ctrl)
+	visibilityInterceptor := chasm.ChasmVisibilityInterceptorProvider(mockVisibilityManager)
+
+	server, address := startTestServer(t, grpc.UnaryInterceptor(visibilityInterceptor.Intercept))
+	defer server.Stop()
+	testRoundTrip(t, address)
+}
+
+func testRoundTrip(t *testing.T, address string) *testspb.TestResponse {
 	conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		t.Fatalf("failed to connect: %v", err)
@@ -74,9 +89,9 @@ func TestChasmRequestInterceptor_ShouldRespond(t *testing.T) {
 		RequestId: "test-request-id",
 	})
 	require.NoError(t, err)
-
 	require.Equal(t, "test-request-id", response.GetRequestId())
-	require.True(t, response.HasEngineCtx)
+
+	return response
 }
 
 func startTestServer(t *testing.T, opt ...grpc.ServerOption) (*grpc.Server, string) {

chasm/interceptors.go

@@ -11,18 +11,17 @@ import (
 
 const chasmRequestPrefix = "/temporal.server.chasm"
 
-// ChasmRequestInterceptor Interceptor that intercepts RPC requests, detects Chasm-specific calls and does additional
-// boilerplate processing before handing off.
-type ChasmRequestInterceptor struct {
+// ChasmEngineInterceptor Interceptor that intercepts RPC requests,
+// detects CHASM-specific calls and does additional boilerplate processing before
+// handing off. Visibility is injected separately with
+// ChasmVisibilityInterceptor.
+type ChasmEngineInterceptor struct {
 	engine         Engine
-	visibilityMgr  VisibilityManager
 	logger         log.Logger
 	metricsHandler metrics.Handler
 }
 
-var _ grpc.UnaryServerInterceptor = (*ChasmRequestInterceptor)(nil).Intercept
-
-func (i *ChasmRequestInterceptor) Intercept(
+func (i *ChasmEngineInterceptor) Intercept(
 	ctx context.Context,
 	req interface{},
 	info *grpc.UnaryServerInfo,
@@ -33,21 +32,39 @@ func (i *ChasmRequestInterceptor) Intercept(
 	}
 
 	ctx = NewEngineContext(ctx, i.engine)
-	ctx = NewVisibilityManagerContext(ctx, i.visibilityMgr)
-
 	return handler(ctx, req)
 }
 
-func ChasmRequestInterceptorProvider(
+func ChasmEngineInterceptorProvider(
 	engine Engine,
-	visibilityMgr VisibilityManager,
 	logger log.Logger,
 	metricsHandler metrics.Handler,
-) *ChasmRequestInterceptor {
-	return &ChasmRequestInterceptor{
+) *ChasmEngineInterceptor {
+	return &ChasmEngineInterceptor{
 		engine:         engine,
-		visibilityMgr:  visibilityMgr,
 		logger:         logger,
 		metricsHandler: metricsHandler,
 	}
 }
+
+// ChasmVisibilityInterceptor intercepts RPC requests and adds the CHASM
+// VisibilityManager to their context.
+type ChasmVisibilityInterceptor struct {
+	visibilityMgr VisibilityManager
+}
+
+func (i *ChasmVisibilityInterceptor) Intercept(
+	ctx context.Context,
+	req interface{},
+	info *grpc.UnaryServerInfo,
+	handler grpc.UnaryHandler,
+) (resp interface{}, retError error) {
+	ctx = NewVisibilityManagerContext(ctx, i.visibilityMgr)
+	return handler(ctx, req)
+}
+
+func ChasmVisibilityInterceptorProvider(visibilityMgr VisibilityManager) *ChasmVisibilityInterceptor {
+	return &ChasmVisibilityInterceptor{
+		visibilityMgr: visibilityMgr,
+	}
+}

common/persistence/visibility/chasm_visibility_manager.go

@@ -28,6 +28,13 @@ func NewChasmVisibilityManager(
 	}
 }
 
+func ChasmVisibilityManagerProvider(
+	registry *chasm.Registry,
+	visibilityMgr manager.VisibilityManager,
+) chasm.VisibilityManager {
+	return NewChasmVisibilityManager(registry, visibilityMgr)
+}
+
 // ListExecutions implements the Engine interface for visibility queries.
 func (e *ChasmVisibilityManager) ListExecutions(
 	ctx context.Context,

service/frontend/fx.go

@@ -113,6 +113,8 @@ var Module = fx.Options(
 	fx.Invoke(EndpointRegistryLifetimeHooks),
 	fx.Provide(schedulerpb.NewSchedulerServiceLayeredClient),
 	nexusfrontend.Module,
+	fx.Provide(visibility.ChasmVisibilityManagerProvider),
+	fx.Provide(chasm.ChasmVisibilityInterceptorProvider),
 )
 
 func NewServiceProvider(
@@ -213,6 +215,7 @@ func GrpcServerOptionsProvider(
 	authInterceptor *authorization.Interceptor,
 	maskInternalErrorDetailsInterceptor *interceptor.MaskInternalErrorDetailsInterceptor,
 	slowRequestLoggerInterceptor *interceptor.SlowRequestLoggerInterceptor,
+	chasmRequestVisibilityInterceptor *chasm.ChasmVisibilityInterceptor,
 	customInterceptors []grpc.UnaryServerInterceptor,
 	customStreamInterceptors []grpc.StreamServerInterceptor,
 	metricsHandler metrics.Handler,
@@ -266,6 +269,7 @@ func GrpcServerOptionsProvider(
 		sdkVersionInterceptor.Intercept,
 		callerInfoInterceptor.Intercept,
 		slowRequestLoggerInterceptor.Intercept,
+		chasmRequestVisibilityInterceptor.Intercept,
 	}
 	if len(customInterceptors) > 0 {
 		// TODO: Deprecate WithChainedFrontendGrpcInterceptors and provide a inner custom interceptor
@@ -736,6 +740,7 @@ func HandlerProvider(
 	versionChecker *VersionChecker,
 	namespaceReplicationQueue FEReplicatorNamespaceReplicationQueue,
 	visibilityMgr manager.VisibilityManager,
+	chasmVisibilityMgr chasm.VisibilityManager,
 	logger log.SnTaggedLogger,
 	throttledLogger log.ThrottledLogger,
 	persistenceExecutionManager persistence.ExecutionManager,

service/history/fx.go

@@ -59,12 +59,13 @@ var Module = fx.Options(
 	fx.Provide(RateLimitInterceptorProvider),
 	fx.Provide(HealthSignalAggregatorProvider),
 	fx.Provide(HealthCheckInterceptorProvider),
-	fx.Provide(chasm.ChasmRequestInterceptorProvider),
+	fx.Provide(chasm.ChasmEngineInterceptorProvider),
+	fx.Provide(chasm.ChasmVisibilityInterceptorProvider),
 	fx.Provide(HistoryAdditionalInterceptorsProvider),
 	fx.Provide(service.GrpcServerOptionsProvider),
 	fx.Provide(ESProcessorConfigProvider),
 	fx.Provide(VisibilityManagerProvider),
-	fx.Provide(ChasmVisibilityManagerProvider),
+	fx.Provide(visibility.ChasmVisibilityManagerProvider),
 	fx.Provide(ThrottledLoggerRpsFnProvider),
 	fx.Provide(PersistenceRateLimitingParamsProvider),
 	service.PersistenceLazyLoadedServiceResolverModule,
@@ -208,9 +209,15 @@ func HealthCheckInterceptorProvider(
 }
 
 func HistoryAdditionalInterceptorsProvider(
-	healthCheckInterceptor *interceptor.HealthCheckInterceptor, chasmRequestInterceptor *chasm.ChasmRequestInterceptor,
+	healthCheckInterceptor *interceptor.HealthCheckInterceptor,
+	chasmRequestEngineInterceptor *chasm.ChasmEngineInterceptor,
+	chasmRequestVisibilityInterceptor *chasm.ChasmVisibilityInterceptor,
 ) []grpc.UnaryServerInterceptor {
-	return []grpc.UnaryServerInterceptor{healthCheckInterceptor.UnaryIntercept, chasmRequestInterceptor.Intercept}
+	return []grpc.UnaryServerInterceptor{
+		healthCheckInterceptor.UnaryIntercept,
+		chasmRequestEngineInterceptor.Intercept,
+		chasmRequestVisibilityInterceptor.Intercept,
+	}
 }
 
 func RateLimitInterceptorProvider(