intercept RepositoryExists calls in Praefect

GitLab calls RepositoryExists prior to creating a repository. Since
Praefect has no records of the repository before it is created,
PerRepositoryRouter's accessor routing fails as it doesn't know
where to send the requests without the database records.

This commit fixes the problem by intercepting RepositoryExists in
Praefect and checking from the database whether the repository
exists or not. Praefect then responds with the information without
involving the coordinator or any Gitalys.

This only affects the repository specific primary stack as the SQL
elector based stack just redirects the call to the virtual storage's
primary if no records of a repository exist.

1 files changed, 22 insertions, 16 deletions

diff --git a/internal/praefect/server.go b/internal/praefect/server.go
index 927450a53..dabac9dd3 100644
--- a/internal/praefect/server.go
+++ b/internal/praefect/server.go
@@ -54,24 +54,30 @@ func NewGRPCServer(
 		grpc_ctxtags.WithFieldExtractorForInitialReq(fieldextractors.FieldExtractor),
 	}
 
+	streamInterceptors := []grpc.StreamServerInterceptor{
+		grpc_ctxtags.StreamServerInterceptor(ctxTagOpts...),
+		grpccorrelation.StreamServerCorrelationInterceptor(), // Must be above the metadata handler
+		middleware.MethodTypeStreamInterceptor(registry),
+		metadatahandler.StreamInterceptor,
+		grpc_prometheus.StreamServerInterceptor,
+		grpc_logrus.StreamServerInterceptor(logger),
+		featureflag.StreamInterceptor,
+		sentryhandler.StreamLogHandler,
+		cancelhandler.Stream, // Should be below LogHandler
+		grpctracing.StreamServerTracingInterceptor(),
+		auth.StreamServerInterceptor(conf.Auth),
+		// Panic handler should remain last so that application panics will be
+		// converted to errors and logged
+		panichandler.StreamPanicHandler,
+	}
+
+	if conf.Failover.ElectionStrategy == config.ElectionStrategyPerRepository {
+		streamInterceptors = append(streamInterceptors, RepositoryExistsStreamInterceptor(rs))
+	}
+
 	grpcOpts = append(grpcOpts, proxyRequiredOpts(director)...)
 	grpcOpts = append(grpcOpts, []grpc.ServerOption{
-		grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
-			grpc_ctxtags.StreamServerInterceptor(ctxTagOpts...),
-			grpccorrelation.StreamServerCorrelationInterceptor(), // Must be above the metadata handler
-			middleware.MethodTypeStreamInterceptor(registry),
-			metadatahandler.StreamInterceptor,
-			grpc_prometheus.StreamServerInterceptor,
-			grpc_logrus.StreamServerInterceptor(logger),
-			featureflag.StreamInterceptor,
-			sentryhandler.StreamLogHandler,
-			cancelhandler.Stream, // Should be below LogHandler
-			grpctracing.StreamServerTracingInterceptor(),
-			auth.StreamServerInterceptor(conf.Auth),
-			// Panic handler should remain last so that application panics will be
-			// converted to errors and logged
-			panichandler.StreamPanicHandler,
-		)),
+		grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(streamInterceptors...)),
 		grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
 			grpc_ctxtags.UnaryServerInterceptor(ctxTagOpts...),
 			grpccorrelation.UnaryServerCorrelationInterceptor(), // Must be above the metadata handler