Merge branch 'smh-repository-generations-read-only' into 'master'

Enforce read-only mode per repository

Closes #2783 and #2862

See merge request gitlab-org/gitaly!2405

6 files changed, 99 insertions, 19 deletions

diff --git a/changelogs/unreleased/smh-repository-generations-read-only.yml b/changelogs/unreleased/smh-repository-generations-read-only.yml
new file mode 100644
index 000000000..775b01a1f
--- /dev/null
+++ b/changelogs/unreleased/smh-repository-generations-read-only.yml
@@ -0,0 +1,5 @@
+---
+title: Enforce read-only mode per repository
+merge_request: 2405
+author:
+type: changed
diff --git a/internal/praefect/coordinator.go b/internal/praefect/coordinator.go
index 9237ef4ab..bdf9e70b3 100644
--- a/internal/praefect/coordinator.go
+++ b/internal/praefect/coordinator.go
@@ -24,11 +24,9 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-type ReadOnlyStorageError string
-
-func (storage ReadOnlyStorageError) Error() string {
-	return fmt.Sprintf("storage %q is in read-only mode", string(storage))
-}
+// ErrRepositoryReadOnly is returned when the repository is in read-only mode. This happens
+// if the primary does not have the latest changes.
+var ErrRepositoryReadOnly = helper.ErrPreconditionFailedf("repository is in read-only mode")
 
 // getReplicationDetails determines the type of job and additional details based on the method name and incoming message
 func getReplicationDetails(methodName string, m proto.Message) (datastore.ChangeType, datastore.Params, error) {
@@ -234,8 +232,12 @@ func (c *Coordinator) mutatorStreamParameters(ctx context.Context, call grpcCall
 		return nil, fmt.Errorf("mutator call: get shard: %w", err)
 	}
 
-	if c.conf.Failover.ReadOnlyAfterFailover && shard.IsReadOnly {
-		return nil, helper.ErrPreconditionFailed(ReadOnlyStorageError(call.targetRepo.GetStorageName()))
+	if c.conf.Failover.ReadOnlyAfterFailover {
+		if latest, err := c.rs.IsLatestGeneration(ctx, virtualStorage, call.targetRepo.RelativePath, shard.Primary.GetStorage()); err != nil {
+			return nil, fmt.Errorf("check generation: %w", err)
+		} else if !latest {
+			return nil, ErrRepositoryReadOnly
+		}
 	}
 
 	primaryMessage, err := rewrittenRepositoryMessage(call.methodInfo, call.msg, shard.Primary.GetStorage())
diff --git a/internal/praefect/coordinator_test.go b/internal/praefect/coordinator_test.go
index 7dbc5553e..85e62276b 100644
--- a/internal/praefect/coordinator_test.go
+++ b/internal/praefect/coordinator_test.go
@@ -72,29 +72,43 @@ func TestStreamDirectorReadOnlyEnforcement(t *testing.T) {
 		},
 	} {
 		t.Run(fmt.Sprintf("read-only: %v, enabled: %v", tc.readOnly, tc.readOnlyEnabled), func(t *testing.T) {
+			const (
+				virtualStorage = "test-virtual-storage"
+				relativePath   = "test-repository"
+				storage        = "test-storage"
+			)
 			conf := config.Config{
 				Failover: config.Failover{ReadOnlyAfterFailover: tc.readOnlyEnabled},
 				VirtualStorages: []*config.VirtualStorage{
 					&config.VirtualStorage{
-						Name: "praefect",
+						Name: virtualStorage,
 						Nodes: []*config.Node{
 							&config.Node{
 								Address: "tcp://gitaly-primary.example.com",
-								Storage: "praefect-internal-1",
+								Storage: storage,
 							},
 						},
 					},
 				},
 			}
 
-			const storageName = "test-storage"
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			rs := datastore.NewMemoryRepositoryStore(nil)
+			require.NoError(t, rs.SetGeneration(ctx, virtualStorage, relativePath, storage, 1))
+			if tc.readOnly {
+				require.NoError(t, rs.SetGeneration(ctx, virtualStorage, relativePath, storage, 0))
+			}
+
 			coordinator := NewCoordinator(
 				datastore.NewMemoryReplicationEventQueue(conf),
-				nil,
-				&nodes.MockManager{GetShardFunc: func(storage string) (nodes.Shard, error) {
+				rs,
+				&nodes.MockManager{GetShardFunc: func(vs string) (nodes.Shard, error) {
+					require.Equal(t, virtualStorage, vs)
 					return nodes.Shard{
 						IsReadOnly: tc.readOnly,
-						Primary:    &nodes.MockNode{StorageName: storageName},
+						Primary:    &nodes.MockNode{StorageName: storage},
 					}, nil
 				}},
 				transactions.NewManager(),
@@ -102,18 +116,15 @@ func TestStreamDirectorReadOnlyEnforcement(t *testing.T) {
 				protoregistry.GitalyProtoPreregistered,
 			)
 
-			ctx, cancel := testhelper.Context()
-			defer cancel()
-
 			frame, err := proto.Marshal(&gitalypb.CleanupRequest{Repository: &gitalypb.Repository{
-				StorageName:  storageName,
-				RelativePath: "only-for-validation",
+				StorageName:  virtualStorage,
+				RelativePath: relativePath,
 			}})
 			require.NoError(t, err)
 
 			_, err = coordinator.StreamDirector(ctx, "/gitaly.RepositoryService/Cleanup", &mockPeeker{frame: frame})
 			if tc.shouldError {
-				require.True(t, errors.Is(err, ReadOnlyStorageError(storageName)))
+				require.Equal(t, ErrRepositoryReadOnly, err)
 				testhelper.RequireGrpcError(t, err, codes.FailedPrecondition)
 			} else {
 				require.NoError(t, err)
diff --git a/internal/praefect/datastore/repository_memory.go b/internal/praefect/datastore/repository_memory.go
index 212aa6deb..7c60c2bce 100644
--- a/internal/praefect/datastore/repository_memory.go
+++ b/internal/praefect/datastore/repository_memory.go
@@ -199,6 +199,16 @@ func (m *MemoryRepositoryStore) GetConsistentSecondaries(ctx context.Context, vi
 	return consistentSecondaries, nil
 }
 
+func (m *MemoryRepositoryStore) IsLatestGeneration(ctx context.Context, virtualStorage, relativePath, storage string) (bool, error) {
+	expected := m.getRepositoryGeneration(virtualStorage, relativePath)
+	if expected == GenerationUnknown {
+		return true, nil
+	}
+
+	actual := m.getStorageGeneration(virtualStorage, relativePath, storage)
+	return expected == actual, nil
+}
+
 func (m *MemoryRepositoryStore) getRepositoryGeneration(virtualStorage, relativePath string) int {
 	gen, ok := m.virtualStorageState[virtualStorage][relativePath]
 	if !ok {
diff --git a/internal/praefect/datastore/repository_memory_test.go b/internal/praefect/datastore/repository_memory_test.go
index 0277099f9..57623e05a 100644
--- a/internal/praefect/datastore/repository_memory_test.go
+++ b/internal/praefect/datastore/repository_memory_test.go
@@ -466,4 +466,27 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) {
 			require.Empty(t, secondaries)
 		})
 	})
+
+	t.Run("IsLatestGeneration", func(t *testing.T) {
+		rs, _ := newStore(t, nil)
+
+		latest, err := rs.IsLatestGeneration(ctx, vs, repo, "no-expected-record")
+		require.NoError(t, err)
+		require.True(t, latest)
+
+		require.NoError(t, rs.SetGeneration(ctx, vs, repo, "up-to-date", 1))
+		require.NoError(t, rs.SetGeneration(ctx, vs, repo, "outdated", 0))
+
+		latest, err = rs.IsLatestGeneration(ctx, vs, repo, "no-record")
+		require.NoError(t, err)
+		require.False(t, latest)
+
+		latest, err = rs.IsLatestGeneration(ctx, vs, repo, "outdated")
+		require.NoError(t, err)
+		require.False(t, latest)
+
+		latest, err = rs.IsLatestGeneration(ctx, vs, repo, "up-to-date")
+		require.NoError(t, err)
+		require.True(t, latest)
+	})
 }
diff --git a/internal/praefect/datastore/repository_postgres.go b/internal/praefect/datastore/repository_postgres.go
index 1427bde23..e4237f848 100644
--- a/internal/praefect/datastore/repository_postgres.go
+++ b/internal/praefect/datastore/repository_postgres.go
@@ -74,6 +74,9 @@ type RepositoryStore interface {
 	// GetConsistentSecondaries checks which secondaries are on the same generation as the primary and returns them.
 	// If the primary's generation is unknown, all secondaries are considered inconsistent.
 	GetConsistentSecondaries(ctx context.Context, virtualStorage, relativePath, primary string) (map[string]struct{}, error)
+	// IsLatestGeneration checks whether the repository is on the latest generation or not. If the repository does not
+	// have an expected generation, every storage is considered to be on the latest version.
+	IsLatestGeneration(ctx context.Context, virtualStorage, relativePath, storage string) (bool, error)
 }
 
 // PostgresRepositoryStore is a Postgres implementation of RepositoryStore.
@@ -353,3 +356,29 @@ WHERE storage = ANY($4::text[])
 
 	return consistentSecondaries, rows.Err()
 }
+
+func (rs *PostgresRepositoryStore) IsLatestGeneration(ctx context.Context, virtualStorage, relativePath, storage string) (bool, error) {
+	const q = `
+SELECT COALESCE(r.generation = sr.generation, false)
+FROM repositories AS r
+LEFT JOIN storage_repositories AS sr
+	ON sr.virtual_storage = r.virtual_storage
+	AND sr.relative_path = r.relative_path
+	AND sr.storage = $3
+WHERE r.virtual_storage = $1
+AND r.relative_path = $2
+`
+
+	var isLatest bool
+	if err := rs.db.QueryRowContext(ctx, q, virtualStorage, relativePath, storage).Scan(&isLatest); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			// if there is no record of the expected generation, we'll have to consider the storage
+			// up to date as this will be the case on repository creation
+			return true, nil
+		}
+
+		return false, err
+	}
+
+	return isLatest, nil
+}