Extract repository_generations and valid_primaries views

This commit extracts the logic for getting the valid primaries into a
view for easier reuse. This logic will be reused when implementing lazy
failovers in `praefect dataloss` and the read-only repository metric.
Both of them currently check whether the primary stored in the database
is outdated to determine whether the repository is in read-only mode or
not. This won't be sufficient with lazy failovers anymore. The repository
would only get a new primary if it is accessed while the current primary
is unhealthy and there are valid primary candidates available. As such,
both of these tools will need to check whether there are valid primaries
to failover before denoting the repository as read-only.

The repository_generations view is also extracted and used in the valid_primaries
view. Many queries currently check the storage_repositories records to figure
out what is the latest generation for a given repository. The view deduplicates
this logic.

2 files changed, 54 insertions, 39 deletions

diff --git a/internal/praefect/datastore/migrations/20210525173505_valid_primaries_view.go b/internal/praefect/datastore/migrations/20210525173505_valid_primaries_view.go
new file mode 100644
index 000000000..84e4161c1
--- /dev/null
+++ b/internal/praefect/datastore/migrations/20210525173505_valid_primaries_view.go
@@ -0,0 +1,53 @@
+package migrations
+
+import migrate "github.com/rubenv/sql-migrate"
+
+func init() {
+	m := &migrate.Migration{
+		Id: "20210525173505_valid_primaries_view",
+		Up: []string{`
+CREATE VIEW repository_generations AS
+	SELECT virtual_storage, relative_path, MAX(generation) AS generation
+	FROM storage_repositories
+	GROUP BY virtual_storage, relative_path
+			`,
+			`
+CREATE VIEW valid_primaries AS
+	WITH candidates AS (
+		SELECT virtual_storage, relative_path, storage, repository_assignments.storage IS NOT NULL AS assigned
+		FROM storage_repositories
+		JOIN repository_generations USING (virtual_storage, relative_path, generation)
+		JOIN healthy_storages USING (virtual_storage, storage)
+		LEFT JOIN repository_assignments USING (virtual_storage, relative_path, storage)
+		WHERE NOT EXISTS (
+			-- This check exists to prevent us from electing a primary that is pending deletion. The primary
+			-- could accept a write and lose it when the deletion is carried out.
+			SELECT true
+			FROM replication_queue
+			WHERE state NOT IN ('completed', 'dead', 'cancelled')
+			AND job->>'change' = 'delete_replica'
+			AND job->>'virtual_storage' = virtual_storage
+			AND job->>'relative_path' = relative_path
+			AND job->>'target_node_storage' = storage
+		)
+	)
+
+	SELECT virtual_storage, relative_path, storage
+	FROM candidates
+	WHERE assigned OR (
+		SELECT NOT EXISTS (
+			SELECT FROM candidates AS assigned_candidates
+			WHERE assigned
+			AND assigned_candidates.virtual_storage = candidates.virtual_storage
+			AND assigned_candidates.relative_path   = candidates.relative_path
+		)
+	)
+						`},
+		Down: []string{
+			"DROP VIEW valid_primaries",
+			"DROP VIEW repository_generations",
+		},
+	}
+
+	allMigrations = append(allMigrations, m)
+}
diff --git a/internal/praefect/nodes/per_repository.go b/internal/praefect/nodes/per_repository.go
index 68294bfec..37239d350 100644
--- a/internal/praefect/nodes/per_repository.go
+++ b/internal/praefect/nodes/per_repository.go
@@ -87,45 +87,7 @@ func (pr *PerRepositoryElector) Run(ctx context.Context, trigger <-chan struct{}
 
 func (pr *PerRepositoryElector) performFailovers(ctx context.Context) error {
 	rows, err := pr.db.QueryContext(ctx, `
-WITH repository_generations AS (
-	SELECT virtual_storage, relative_path, MAX(generation) AS generation
-	FROM storage_repositories
-	GROUP BY virtual_storage, relative_path
-),
-
-candidates AS (
-	SELECT virtual_storage, relative_path, storage, repository_assignments.storage IS NOT NULL AS assigned
-	FROM storage_repositories
-	JOIN repository_generations USING (virtual_storage, relative_path, generation)
-	JOIN healthy_storages USING (virtual_storage, storage)
-	LEFT JOIN repository_assignments USING (virtual_storage, relative_path, storage)
-	WHERE NOT EXISTS (
-		-- This check exists to prevent us from electing a primary that is pending deletion. The primary
-		-- could accept a write and lose it when the deletion is carried out.
-		SELECT true
-		FROM replication_queue
-		WHERE state NOT IN ('completed', 'dead', 'cancelled')
-		AND job->>'change' = 'delete_replica'
-		AND job->>'virtual_storage' = virtual_storage
-		AND job->>'relative_path' = relative_path
-		AND job->>'target_node_storage' = storage
-	)
-),
-
-valid_primaries AS (
-	SELECT virtual_storage, relative_path, storage
-	FROM candidates
-	WHERE assigned OR (
-		SELECT NOT EXISTS (
-			SELECT FROM candidates AS assigned_candidates
-			WHERE assigned
-			AND assigned_candidates.virtual_storage = candidates.virtual_storage
-			AND assigned_candidates.relative_path   = candidates.relative_path
-		)
-	)
-),
-
-updated AS (
+WITH updated AS (
 	UPDATE repositories
 		SET "primary" = (
 			SELECT storage