diff options
| author | Pavlo Strokov <pstrokov@gitlab.com> | 2021-10-19 16:31:05 +0300 |
|---|---|---|
| committer | Pavlo Strokov <pstrokov@gitlab.com> | 2021-10-26 12:45:55 +0300 |
| commit | 4440c010033ab762621e3313c150987f552baf4e (patch) | |
| tree | e7a0e44f977062991f256f6f30d4f48dfa7ef521 | |
| parent | 109e213f5b1f5e6578ed184ef8aa56004f054621 (diff) | |
sql: Run migrations and tests on pre-populated database
To get better tests we should run test scenarios on top of
pre-populated database. It will give us more confidence
about our queries as well as highlight any hidden conditions
as triggers execution during run of the migration scripts, etc.
The population of the database is done via a migration script
that is injected into a common migration set during template
database creation. It inserts data into storage_repositories
and repositories tables at the moment as we had problems with
running migrations on them because of the triggers. The set
could be extended as well. We also could add additional
code checks to verify migrations were applied as expected.
Because of pre-existing data we don't want to remove we change
the way we do the testing. Now we rely on the transactions to
run the tests and get into initial database state once the test
is completed. If it is not possible to use a transaction (nested
transactions are not supported) we remove any created artifacts
after test completion.
We change repository_id hardcoded values to the value returned
by the ReserveRepositoryID method for particular virtual storage
and relative path as we can't predict auto-generated values for
the column anymore.
Part of: https://gitlab.com/gitlab-org/gitaly/-/issues/3820
18 files changed, 525 insertions, 386 deletions
diff --git a/cmd/praefect/subcmd_accept_dataloss_test.go b/cmd/praefect/subcmd_accept_dataloss_test.go index 55d235afc..b58a0f378 100644 --- a/cmd/praefect/subcmd_accept_dataloss_test.go +++ b/cmd/praefect/subcmd_accept_dataloss_test.go @@ -39,18 +39,20 @@ func TestAcceptDatalossSubcommand(t *testing.T) { rs := datastore.NewPostgresRepositoryStore(db, conf.StorageNames()) startingGenerations := map[string]int{st1: 1, st2: 0, st3: datastore.GenerationUnknown} - repoCreated := false + repositoryID := int64(0) for storage, generation := range startingGenerations { if generation == datastore.GenerationUnknown { continue } - if !repoCreated { - repoCreated = true - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, storage, nil, nil, false, false)) + if repositoryID == 0 { + var err error + repositoryID, err = rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, repositoryID, vs, repo, storage, nil, nil, false, false)) } - require.NoError(t, rs.SetGeneration(ctx, 1, storage, generation)) + require.NoError(t, rs.SetGeneration(ctx, repositoryID, storage, generation)) } ln, clean := listenAndServe(t, []svcRegistrar{registerPraefectInfoServer(info.NewServer(conf, rs, nil, nil, nil))}) @@ -145,7 +147,7 @@ func TestAcceptDatalossSubcommand(t *testing.T) { require.NoError(t, fs.Parse(tc.args)) tc.matchError(t, cmd.Exec(fs, conf)) for storage, expected := range tc.expectedGenerations { - actual, err := rs.GetGeneration(ctx, 1, storage) + actual, err := rs.GetGeneration(ctx, repositoryID, storage) require.NoError(t, err) require.Equal(t, expected, actual, storage) } diff --git a/cmd/praefect/subcmd_dataloss_test.go b/cmd/praefect/subcmd_dataloss_test.go index 07d7ad602..deb16746c 100644 --- a/cmd/praefect/subcmd_dataloss_test.go +++ b/cmd/praefect/subcmd_dataloss_test.go @@ -56,28 +56,28 @@ func TestDatalossSubcommand(t *testing.T) { ` INSERT INTO repositories (repository_id, virtual_storage, relative_path, "primary") VALUES - (1, 'virtual-storage-1', 'repository-1', 'gitaly-1'), - (2, 'virtual-storage-1', 'repository-2', 'gitaly-3') + (100000000, 'virtual-storage-1', 'repository-1', 'gitaly-1'), + (100000001, 'virtual-storage-1', 'repository-2', 'gitaly-3') `, ` INSERT INTO repository_assignments (repository_id, virtual_storage, relative_path, storage) VALUES - (1, 'virtual-storage-1', 'repository-1', 'gitaly-1'), - (1, 'virtual-storage-1', 'repository-1', 'gitaly-2'), - (2, 'virtual-storage-1', 'repository-2', 'gitaly-1'), - (2, 'virtual-storage-1', 'repository-2', 'gitaly-3') + (100000000, 'virtual-storage-1', 'repository-1', 'gitaly-1'), + (100000000, 'virtual-storage-1', 'repository-1', 'gitaly-2'), + (100000001, 'virtual-storage-1', 'repository-2', 'gitaly-1'), + (100000001, 'virtual-storage-1', 'repository-2', 'gitaly-3') `, } { _, err := tx.ExecContext(ctx, q) require.NoError(t, err) } - require.NoError(t, gs.SetGeneration(ctx, 1, "gitaly-1", 1)) - require.NoError(t, gs.SetGeneration(ctx, 1, "gitaly-2", 0)) - require.NoError(t, gs.SetGeneration(ctx, 1, "gitaly-3", 0)) + require.NoError(t, gs.SetGeneration(ctx, 100000000, "gitaly-1", 1)) + require.NoError(t, gs.SetGeneration(ctx, 100000000, "gitaly-2", 0)) + require.NoError(t, gs.SetGeneration(ctx, 100000000, "gitaly-3", 0)) - require.NoError(t, gs.SetGeneration(ctx, 2, "gitaly-2", 1)) - require.NoError(t, gs.SetGeneration(ctx, 2, "gitaly-3", 0)) + require.NoError(t, gs.SetGeneration(ctx, 100000001, "gitaly-2", 1)) + require.NoError(t, gs.SetGeneration(ctx, 100000001, "gitaly-3", 0)) ln, clean := listenAndServe(t, []svcRegistrar{ registerPraefectInfoServer(info.NewServer(cfg, gs, nil, nil, nil)), diff --git a/cmd/praefect/subcmd_set_replication_factor_test.go b/cmd/praefect/subcmd_set_replication_factor_test.go index 6e33847b4..8369d436d 100644 --- a/cmd/praefect/subcmd_set_replication_factor_test.go +++ b/cmd/praefect/subcmd_set_replication_factor_test.go @@ -92,8 +92,11 @@ func TestSetReplicationFactorSubcommand(t *testing.T) { } // create a repository record + rs := datastore.NewPostgresRepositoryStore(tx, nil) + id, err := rs.ReserveRepositoryID(ctx, "virtual-storage", "relative-path") + require.NoError(t, err) require.NoError(t, - datastore.NewPostgresRepositoryStore(tx, nil).CreateRepository(ctx, 1, "virtual-storage", "relative-path", "primary", nil, nil, false, false), + rs.CreateRepository(ctx, id, "virtual-storage", "relative-path", "primary", nil, nil, false, false), ) ln, clean := listenAndServe(t, []svcRegistrar{registerPraefectInfoServer( @@ -105,7 +108,7 @@ func TestSetReplicationFactorSubcommand(t *testing.T) { cmd := &setReplicationFactorSubcommand{stdout: stdout} fs := cmd.FlagSet() require.NoError(t, fs.Parse(tc.args)) - err := cmd.Exec(fs, config.Config{ + err = cmd.Exec(fs, config.Config{ SocketPath: ln.Addr().String(), }) testassert.GrpcEqualErr(t, tc.error, err) diff --git a/internal/praefect/checks_test.go b/internal/praefect/checks_test.go index 9e6a6f1b5..f8adc31f7 100644 --- a/internal/praefect/checks_test.go +++ b/internal/praefect/checks_test.go @@ -55,6 +55,9 @@ func TestPraefectMigrations_success(t *testing.T) { t.Run(tc.desc, func(t *testing.T) { var cfg config.Config db := glsql.NewDB(t) + // Migration that adds artificial data to the database needs to be dropped first + _, err := db.Exec(`DELETE FROM ` + migrations.MigrationTableName + " WHERE id = '20200921170400_artificial_repositories'") + require.NoError(t, err) cfg.DB = glsql.GetDBConfig(t, db.Name) require.NoError(t, tc.prepare(cfg)) diff --git a/internal/praefect/coordinator_pg_test.go b/internal/praefect/coordinator_pg_test.go index 53b645666..9dc5db242 100644 --- a/internal/praefect/coordinator_pg_test.go +++ b/internal/praefect/coordinator_pg_test.go @@ -10,7 +10,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "gitlab.com/gitlab-org/gitaly/v14/internal/praefect/commonerr" "gitlab.com/gitlab-org/gitaly/v14/internal/praefect/config" "gitlab.com/gitlab-org/gitaly/v14/internal/praefect/datastore" "gitlab.com/gitlab-org/gitaly/v14/internal/praefect/datastore/glsql" @@ -159,7 +158,6 @@ func TestStreamDirectorMutator_Transaction(t *testing.T) { for _, tc := range testcases { t.Run(tc.desc, func(t *testing.T) { db.Truncate(t, "replication_queue_job_lock", "replication_queue", "replication_queue_lock") - db.SequenceReset(t) storageNodes := make([]*config.Node, 0, len(tc.nodes)) for i := range tc.nodes { @@ -201,25 +199,25 @@ func TestStreamDirectorMutator_Transaction(t *testing.T) { // set up the generations prior to transaction rs := datastore.NewPostgresRepositoryStore(tx, conf.StorageNames()) - repoCreated := false + var id int64 + var err error for i, n := range tc.nodes { if n.generation == datastore.GenerationUnknown { continue } - if !repoCreated { - repoCreated = true - require.NoError(t, rs.CreateRepository(ctx, 1, repo.StorageName, repo.RelativePath, storageNodes[i].Storage, nil, nil, true, false)) + if id == 0 { + id, err = rs.ReserveRepositoryID(ctx, repo.StorageName, repo.RelativePath) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, repo.StorageName, repo.RelativePath, storageNodes[i].Storage, nil, nil, true, false)) defer func() { - repositoryStore := datastore.NewPostgresRepositoryStore(db, nil) - _, _, err := repositoryStore.DeleteRepository(ctx, repo.StorageName, repo.RelativePath) - if !errors.As(err, &commonerr.RepositoryNotFoundError{}) { - require.NoError(t, err) - } + repoStore := datastore.NewPostgresRepositoryStore(db, nil) + _, _, err := repoStore.DeleteRepository(ctx, repo.StorageName, repo.RelativePath) + require.NoError(t, err) }() } - require.NoError(t, rs.SetGeneration(ctx, 1, storageNodes[i].Storage, n.generation)) + require.NoError(t, rs.SetGeneration(ctx, id, storageNodes[i].Storage, n.generation)) } testhelper.SetHealthyNodes(t, ctx, tx, map[string]map[string][]string{"praefect": conf.StorageNames()}) @@ -307,7 +305,7 @@ func TestStreamDirectorMutator_Transaction(t *testing.T) { } if tc.concurrentWrite { - require.NoError(t, rs.SetGeneration(ctx, 1, "non-participating-storage", 2)) + require.NoError(t, rs.SetGeneration(ctx, id, "non-participating-storage", 2)) } err = streamParams.RequestFinalizer() @@ -321,7 +319,7 @@ func TestStreamDirectorMutator_Transaction(t *testing.T) { // Nodes that did not successfully commit or did not participate should remain on their // existing generation. for i, n := range tc.nodes { - gen, err := rs.GetGeneration(ctx, 1, storageNodes[i].Storage) + gen, err := rs.GetGeneration(ctx, id, storageNodes[i].Storage) require.NoError(t, err) require.Equal(t, n.expectedGeneration, gen, "node %d has wrong generation", i) } diff --git a/internal/praefect/coordinator_test.go b/internal/praefect/coordinator_test.go index a7a7561cd..0dbc6b839 100644 --- a/internal/praefect/coordinator_test.go +++ b/internal/praefect/coordinator_test.go @@ -194,8 +194,12 @@ func TestStreamDirectorMutator(t *testing.T) { rs := datastore.NewPostgresRepositoryStore(tx, conf.StorageNames()) + var id int64 + var err error if tc.repositoryExists { - require.NoError(t, rs.CreateRepository(ctx, 1, targetRepo.StorageName, targetRepo.RelativePath, primaryNode.Storage, []string{secondaryNode.Storage}, nil, true, true)) + id, err = rs.ReserveRepositoryID(ctx, targetRepo.StorageName, targetRepo.RelativePath) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, targetRepo.StorageName, targetRepo.RelativePath, primaryNode.Storage, []string{secondaryNode.Storage}, nil, true, true)) } testhelper.SetHealthyNodes(t, ctx, tx, map[string]map[string][]string{"praefect": conf.StorageNames()}) @@ -273,7 +277,7 @@ func TestStreamDirectorMutator(t *testing.T) { CreatedAt: events[0].CreatedAt, UpdatedAt: events[0].UpdatedAt, Job: datastore.ReplicationJob{ - RepositoryID: 1, + RepositoryID: id, Change: datastore.UpdateRepo, VirtualStorage: conf.VirtualStorages[0].Name, RelativePath: targetRepo.RelativePath, diff --git a/internal/praefect/datastore/assignment_test.go b/internal/praefect/datastore/assignment_test.go index 12409a6e3..599d739ef 100644 --- a/internal/praefect/datastore/assignment_test.go +++ b/internal/praefect/datastore/assignment_test.go @@ -92,7 +92,7 @@ func TestAssignmentStore_GetHostAssignments(t *testing.T) { require.NoError(t, rs.CreateRepository(ctx, repositoryID, assignment.virtualStorage, assignment.relativePath, assignment.storage, nil, nil, false, false)) } - _, err = db.ExecContext(ctx, ` + _, err = tx.ExecContext(ctx, ` INSERT INTO repository_assignments (repository_id, virtual_storage, relative_path, storage) VALUES ($1, $2, $3, $4) `, repositoryID, assignment.virtualStorage, assignment.relativePath, assignment.storage) @@ -133,6 +133,7 @@ func TestAssignmentStore_SetReplicationFactor(t *testing.T) { } db := glsql.NewDB(t) + const repositoryID = 1e10 for _, tc := range []struct { desc string @@ -219,16 +220,16 @@ func TestAssignmentStore_SetReplicationFactor(t *testing.T) { if !tc.nonExistentRepository { _, err := tx.ExecContext(ctx, ` INSERT INTO repositories (virtual_storage, relative_path, "primary", repository_id) - VALUES ('virtual-storage', 'relative-path', 'primary', 1000) - `) + VALUES ('virtual-storage', 'relative-path', 'primary', $1) + `, repositoryID) require.NoError(t, err) } for _, storage := range tc.existingAssignments { _, err := tx.ExecContext(ctx, ` INSERT INTO repository_assignments (virtual_storage, relative_path, storage, repository_id) - VALUES ('virtual-storage', 'relative-path', $1, 1000) - `, storage) + VALUES ('virtual-storage', 'relative-path', $1, $2) + `, storage, repositoryID) require.NoError(t, err) } @@ -242,7 +243,7 @@ func TestAssignmentStore_SetReplicationFactor(t *testing.T) { tc.requireStorages(t, setStorages) - assignedStorages, err := store.GetHostAssignments(ctx, "virtual-storage", 1) + assignedStorages, err := store.GetHostAssignments(ctx, "virtual-storage", repositoryID) require.NoError(t, err) sort.Strings(assignedStorages) @@ -252,8 +253,8 @@ func TestAssignmentStore_SetReplicationFactor(t *testing.T) { require.NoError(t, tx.QueryRowContext(ctx, ` SELECT array_agg(storage) FROM repository_assignments - WHERE COALESCE(repository_id != 1000, true) - `).Scan(&storagesWithIncorrectRepositoryID)) + WHERE COALESCE(repository_id != $1, true) + `, repositoryID).Scan(&storagesWithIncorrectRepositoryID)) require.Empty(t, storagesWithIncorrectRepositoryID) }) } diff --git a/internal/praefect/datastore/glsql/testing.go b/internal/praefect/datastore/glsql/testing.go index aa180fd71..e1d37ec91 100644 --- a/internal/praefect/datastore/glsql/testing.go +++ b/internal/praefect/datastore/glsql/testing.go @@ -235,7 +235,7 @@ func initPraefectTestDB(t testing.TB, database string) *sql.DB { require.NoErrorf(t, templateDB.Close(), "release connection to the %q database", templateDBConf.DBName) }() - if _, err := Migrate(templateDB, false, migrations.All()); err != nil { + if _, err := Migrate(templateDB, false, append(migrations.All(), testDataMigrations()...)); err != nil { // If database has unknown migration we try to re-create template database with // current migration. It may be caused by other code changes done in another branch. if pErr := (*migrate.PlanError)(nil); errors.As(err, &pErr) { @@ -251,7 +251,7 @@ func initPraefectTestDB(t testing.TB, database string) *sql.DB { defer func() { require.NoErrorf(t, remigrateTemplateDB.Close(), "release connection to the %q database", templateDBConf.DBName) }() - _, err = Migrate(remigrateTemplateDB, false, migrations.All()) + _, err = Migrate(remigrateTemplateDB, false, append(migrations.All(), testDataMigrations()...)) require.NoErrorf(t, err, "failed to run database migration on %q", praefectTemplateDatabase) } else { require.NoErrorf(t, err, "failed to run database migration on %q", praefectTemplateDatabase) @@ -350,3 +350,64 @@ func getDatabaseEnvironment(t testing.TB) map[string]string { return databaseEnv } + +// testDataMigrations function should be used to include additional migration scripts into the +// existing set of migrations. All migrations applied in order. The ordering is done via numeric +// prefix of the migration. To insert migration in between some existing migrations you should +// include it with the prefix that will be greater that the lowest and lesser than the upper one. +func testDataMigrations() []*migrate.Migration { + return []*migrate.Migration{ + { + // Applied after 20200921170311_repositories_primary_column + Id: "20200921170400_artificial_repositories", + Up: []string{ + // random_hex_string generates a random HEX string with requested length. + `-- +migrate StatementBegin + CREATE OR REPLACE FUNCTION random_hex_string(length INTEGER) RETURNS TEXT AS + $$ + DECLARE + chars TEXT[] := '{0,1,2,3,4,5,6,7,8,9,a,b,c,d,e,f}'; + result TEXT := ''; + i INTEGER := 0; + BEGIN + IF length < 0 THEN + RAISE EXCEPTION 'Given length cannot be less than 0'; + END IF; + FOR i IN 1..length LOOP + result := result || chars[1+RANDOM()*(ARRAY_LENGTH(chars, 1)-1)]; + END LOOP; + RETURN result; + END; + $$ LANGUAGE plpgsql`, + + // repository_relative_path generates a relative path for the repository. + // It has a standard format: @hashed/ab/cd/abcd000000000000000000000000000000000000000000000000000000000000.git + `-- +migrate StatementBegin + CREATE OR REPLACE FUNCTION repository_relative_path() RETURNS TEXT AS + $$ + DECLARE + result TEXT; + BEGIN + SELECT CONCAT_WS('/', '@hashed', SUBSTR(path, 1, 2), SUBSTR(path, 3, 2), path || '.git') INTO result + FROM (SELECT random_hex_string(64) AS path) t; + RETURN result; + END; + $$ LANGUAGE plpgsql + -- +migrate StatementEnd`, + + // Populate database tables with artificial data. + `INSERT INTO storage_repositories (virtual_storage, relative_path, storage, generation) + SELECT virtual_storages.name, repositories.relative_path, storages.name, 1 + FROM (SELECT UNNEST(ARRAY['artificial-praefect-1','artificial-praefect-2']) AS name) virtual_storages + CROSS JOIN + (SELECT UNNEST(ARRAY['artificial-gitaly-1','artificial-gitaly-2','artificial-gitaly-3']) AS name) storages + CROSS JOIN + (SELECT generate_series(1,300), repository_relative_path() AS relative_path) repositories`, + + `INSERT INTO repositories (virtual_storage, relative_path, generation) + SELECT DISTINCT virtual_storage, relative_path, 1 + FROM storage_repositories`, + }, + }, + } +} diff --git a/internal/praefect/datastore/listener_postgres_test.go b/internal/praefect/datastore/listener_postgres_test.go index 53f1cc403..7dc5c99a8 100644 --- a/internal/praefect/datastore/listener_postgres_test.go +++ b/internal/praefect/datastore/listener_postgres_test.go @@ -377,7 +377,7 @@ func TestPostgresListener_Listen_repositories_delete(t *testing.T) { "repositories_updates", func(t *testing.T) { _, err := db.DB.Exec(` - INSERT INTO repositories + INSERT INTO repositories (virtual_storage, relative_path, generation) VALUES ('praefect-1', '/path/to/repo/1', 1), ('praefect-1', '/path/to/repo/2', 1), ('praefect-1', '/path/to/repo/3', 0), @@ -385,7 +385,7 @@ func TestPostgresListener_Listen_repositories_delete(t *testing.T) { require.NoError(t, err) }, func(t *testing.T) { - _, err := db.DB.Exec(`DELETE FROM repositories WHERE generation > 0`) + _, err := db.DB.Exec(`DELETE FROM repositories WHERE generation > 0 AND virtual_storage LIKE 'praefect%'`) require.NoError(t, err) }, func(t *testing.T, n glsql.Notification) { @@ -435,11 +435,14 @@ func TestPostgresListener_Listen_storage_repositories_update(t *testing.T) { db.Name, channel, func(t *testing.T) { - _, err := db.DB.Exec(`INSERT INTO storage_repositories VALUES ('praefect-1', '/path/to/repo', 'gitaly-1', 0)`) + _, err := db.DB.Exec( + `INSERT INTO storage_repositories(virtual_storage, relative_path, storage, generation) + VALUES ('praefect-1', '/path/to/repo', 'gitaly-1', 0)`, + ) require.NoError(t, err) }, func(t *testing.T) { - _, err := db.DB.Exec(`UPDATE storage_repositories SET generation = generation + 1`) + _, err := db.DB.Exec(`UPDATE storage_repositories SET generation = generation + 1 WHERE virtual_storage = 'praefect-1'`) require.NoError(t, err) }, func(t *testing.T, n glsql.Notification) { @@ -461,7 +464,7 @@ func TestPostgresListener_Listen_storage_empty_notification(t *testing.T) { channel, func(t *testing.T) {}, func(t *testing.T) { - _, err := db.DB.Exec(`UPDATE storage_repositories SET generation = 1`) + _, err := db.DB.Exec(`UPDATE storage_repositories SET generation = 1 WHERE repository_id = -1`) require.NoError(t, err) }, nil, // no notification events expected @@ -486,7 +489,7 @@ func TestPostgresListener_Listen_storage_repositories_delete(t *testing.T) { require.NoError(t, err) }, func(t *testing.T) { - _, err := db.DB.Exec(`DELETE FROM storage_repositories`) + _, err := db.DB.Exec(`DELETE FROM storage_repositories WHERE virtual_storage = 'praefect-1'`) require.NoError(t, err) }, func(t *testing.T, n glsql.Notification) { @@ -497,6 +500,7 @@ func TestPostgresListener_Listen_storage_repositories_delete(t *testing.T) { } func testListener(t *testing.T, dbName, channel string, setup func(t *testing.T), trigger func(t *testing.T), verifier func(t *testing.T, notification glsql.Notification)) { + t.Helper() setup(t) readyChan := make(chan struct{}) diff --git a/internal/praefect/datastore/repository_store_test.go b/internal/praefect/datastore/repository_store_test.go index 1fb0c4444..9781f195c 100644 --- a/internal/praefect/datastore/repository_store_test.go +++ b/internal/praefect/datastore/repository_store_test.go @@ -52,8 +52,7 @@ LEFT JOIN ( FROM repository_assignments GROUP BY repository_id, virtual_storage, relative_path ) AS repository_assignments USING (repository_id, virtual_storage, relative_path) - - `) +WHERE virtual_storage NOT LIKE 'artificial%'`) require.NoError(t, err) defer rows.Close() @@ -85,7 +84,7 @@ LEFT JOIN ( requireStorageState := func(t testing.TB, ctx context.Context, exp storageState) { rows, err := db.QueryContext(ctx, ` SELECT repository_id, virtual_storage, relative_path, storage, generation -FROM storage_repositories +FROM storage_repositories WHERE virtual_storage NOT LIKE 'artificial%' `) require.NoError(t, err) defer rows.Close() @@ -142,7 +141,7 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { first call second call error error - state storageState + state func(id int64) storageState }{ { desc: "both successful", @@ -154,13 +153,15 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { primary: "primary", secondaries: []string{"secondary"}, }, - state: storageState{ - "virtual-storage": { - "relative-path": { - "primary": {repositoryID: 1, generation: 2}, - "secondary": {repositoryID: 1, generation: 2}, + state: func(id int64) storageState { + return storageState{ + "virtual-storage": { + "relative-path": { + "primary": {repositoryID: id, generation: 2}, + "secondary": {repositoryID: id, generation: 2}, + }, }, - }, + } }, }, { @@ -172,13 +173,15 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { primary: "primary", secondaries: []string{"secondary"}, }, - state: storageState{ - "virtual-storage": { - "relative-path": { - "primary": {repositoryID: 1, generation: 2}, - "secondary": {repositoryID: 1, generation: 0}, + state: func(id int64) storageState { + return storageState{ + "virtual-storage": { + "relative-path": { + "primary": {repositoryID: id, generation: 2}, + "secondary": {repositoryID: id, generation: 0}, + }, }, - }, + } }, }, { @@ -190,13 +193,15 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { primary: "secondary", }, error: errWriteToOutdatedNodes, - state: storageState{ - "virtual-storage": { - "relative-path": { - "primary": {repositoryID: 1, generation: 1}, - "secondary": {repositoryID: 1, generation: 0}, + state: func(id int64) storageState { + return storageState{ + "virtual-storage": { + "relative-path": { + "primary": {repositoryID: id, generation: 1}, + "secondary": {repositoryID: id, generation: 0}, + }, }, - }, + } }, }, } { @@ -205,7 +210,9 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { defer cancel() globalRepositoryStore := NewPostgresRepositoryStore(db, nil) - require.NoError(t, globalRepositoryStore.CreateRepository(ctx, 1, "virtual-storage", "relative-path", "primary", []string{"secondary"}, nil, false, false)) + id, err := globalRepositoryStore.ReserveRepositoryID(ctx, "virtual-storage", "relative-path") + require.NoError(t, err) + require.NoError(t, globalRepositoryStore.CreateRepository(ctx, id, "virtual-storage", "relative-path", "primary", []string{"secondary"}, nil, false, false)) defer func() { _, _, err := globalRepositoryStore.DeleteRepository(ctx, "virtual-storage", "relative-path") require.NoError(t, err) @@ -214,7 +221,7 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { firstTx := db.Begin(t) secondTx := db.Begin(t) - err := NewPostgresRepositoryStore(firstTx, nil).IncrementGeneration(ctx, 1, tc.first.primary, tc.first.secondaries) + err = NewPostgresRepositoryStore(firstTx, nil).IncrementGeneration(ctx, id, tc.first.primary, tc.first.secondaries) require.NoError(t, err) go func() { @@ -222,13 +229,13 @@ func TestRepositoryStore_incrementGenerationConcurrently(t *testing.T) { firstTx.Commit(t) }() - err = NewPostgresRepositoryStore(secondTx, nil).IncrementGeneration(ctx, 1, tc.second.primary, tc.second.secondaries) + err = NewPostgresRepositoryStore(secondTx, nil).IncrementGeneration(ctx, id, tc.second.primary, tc.second.secondaries) require.Equal(t, tc.error, err) secondTx.Commit(t) requireState(t, ctx, db, - virtualStorageState{"virtual-storage": {"relative-path": {repositoryID: 1, replicaPath: "relative-path"}}}, - tc.state, + virtualStorageState{"virtual-storage": {"relative-path": {repositoryID: id, replicaPath: "relative-path"}}}, + tc.state(id), ) }) } @@ -277,7 +284,7 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { rs, requireState := newStore(t, nil) require.Equal(t, - rs.IncrementGeneration(ctx, 1, "primary", []string{"secondary-1"}), + rs.IncrementGeneration(ctx, 1e10, "primary", []string{"secondary-1"}), commonerr.ErrRepositoryNotFound, ) requireState(t, ctx, @@ -288,26 +295,27 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("write to outdated nodes", func(t *testing.T) { rs, requireState := newStore(t, nil) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "latest-node", []string{"outdated-primary", "outdated-secondary"}, nil, false, false)) - require.NoError(t, rs.SetGeneration(ctx, 1, "latest-node", 1)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "latest-node", []string{"outdated-primary", "outdated-secondary"}, nil, false, false)) + require.NoError(t, rs.SetGeneration(ctx, id, "latest-node", 1)) require.Equal(t, - rs.IncrementGeneration(ctx, 1, "outdated-primary", []string{"outdated-secondary"}), + rs.IncrementGeneration(ctx, id, "outdated-primary", []string{"outdated-secondary"}), errWriteToOutdatedNodes, ) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "latest-node": {repositoryID: 1, generation: 1}, - "outdated-primary": {repositoryID: 1, generation: 0}, - "outdated-secondary": {repositoryID: 1, generation: 0}, + "latest-node": {repositoryID: id, generation: 1}, + "outdated-primary": {repositoryID: id, generation: 0}, + "outdated-secondary": {repositoryID: id, generation: 0}, }, }, }, @@ -317,82 +325,86 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("increments generation for up to date nodes", func(t *testing.T) { rs, requireState := newStore(t, nil) - for id, pair := range []struct{ virtualStorage, relativePath string }{ + var ids []int64 + for _, pair := range []struct{ virtualStorage, relativePath string }{ {vs, repo}, // create records that don't get modified to ensure the query is correctly scoped by virtual storage // and relative path {vs, "other-relative-path"}, {"other-virtual-storage", repo}, } { - require.NoError(t, rs.CreateRepository(ctx, int64(id+1), pair.virtualStorage, pair.relativePath, "primary", []string{"up-to-date-secondary", "outdated-secondary"}, nil, false, false)) + id, err := rs.ReserveRepositoryID(ctx, pair.virtualStorage, pair.relativePath) + require.NoError(t, err) + ids = append(ids, id) + require.NoError(t, rs.CreateRepository(ctx, id, pair.virtualStorage, pair.relativePath, "primary", []string{"up-to-date-secondary", "outdated-secondary"}, nil, false, false)) } - require.NoError(t, rs.IncrementGeneration(ctx, 1, "primary", []string{"up-to-date-secondary"})) + require.NoError(t, rs.IncrementGeneration(ctx, ids[0], "primary", []string{"up-to-date-secondary"})) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, - "other-relative-path": {repositoryID: 2, replicaPath: "other-relative-path"}, + "repository-1": {repositoryID: ids[0], replicaPath: "repository-1"}, + "other-relative-path": {repositoryID: ids[1], replicaPath: "other-relative-path"}, }, "other-virtual-storage": { - "repository-1": {repositoryID: 3, replicaPath: "repository-1"}, + "repository-1": {repositoryID: ids[2], replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "primary": {repositoryID: 1, generation: 1}, - "up-to-date-secondary": {repositoryID: 1, generation: 1}, - "outdated-secondary": {repositoryID: 1, generation: 0}, + "primary": {repositoryID: ids[0], generation: 1}, + "up-to-date-secondary": {repositoryID: ids[0], generation: 1}, + "outdated-secondary": {repositoryID: ids[0], generation: 0}, }, "other-relative-path": { - "primary": {repositoryID: 2}, - "up-to-date-secondary": {repositoryID: 2}, - "outdated-secondary": {repositoryID: 2}, + "primary": {repositoryID: ids[1]}, + "up-to-date-secondary": {repositoryID: ids[1]}, + "outdated-secondary": {repositoryID: ids[1]}, }, }, "other-virtual-storage": { "repository-1": { - "primary": {repositoryID: 3}, - "up-to-date-secondary": {repositoryID: 3}, - "outdated-secondary": {repositoryID: 3}, + "primary": {repositoryID: ids[2]}, + "up-to-date-secondary": {repositoryID: ids[2]}, + "outdated-secondary": {repositoryID: ids[2]}, }, }, }, ) - require.NoError(t, rs.IncrementGeneration(ctx, 1, "primary", []string{ + require.NoError(t, rs.IncrementGeneration(ctx, ids[0], "primary", []string{ "up-to-date-secondary", "outdated-secondary", "non-existing-secondary", })) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, - "other-relative-path": {repositoryID: 2, replicaPath: "other-relative-path"}, + "repository-1": {repositoryID: ids[0], replicaPath: "repository-1"}, + "other-relative-path": {repositoryID: ids[1], replicaPath: "other-relative-path"}, }, "other-virtual-storage": { - "repository-1": {repositoryID: 3, replicaPath: "repository-1"}, + "repository-1": {repositoryID: ids[2], replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "primary": {repositoryID: 1, generation: 2}, - "up-to-date-secondary": {repositoryID: 1, generation: 2}, - "outdated-secondary": {repositoryID: 1, generation: 0}, + "primary": {repositoryID: ids[0], generation: 2}, + "up-to-date-secondary": {repositoryID: ids[0], generation: 2}, + "outdated-secondary": {repositoryID: ids[0], generation: 0}, }, "other-relative-path": { - "primary": {repositoryID: 2}, - "up-to-date-secondary": {repositoryID: 2}, - "outdated-secondary": {repositoryID: 2}, + "primary": {repositoryID: ids[1]}, + "up-to-date-secondary": {repositoryID: ids[1]}, + "outdated-secondary": {repositoryID: ids[1]}, }, }, "other-virtual-storage": { "repository-1": { - "primary": {repositoryID: 3}, - "up-to-date-secondary": {repositoryID: 3}, - "outdated-secondary": {repositoryID: 3}, + "primary": {repositoryID: ids[2]}, + "up-to-date-secondary": {repositoryID: ids[2]}, + "outdated-secondary": {repositoryID: ids[2]}, }, }, }, @@ -403,18 +415,19 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("SetGeneration", func(t *testing.T) { t.Run("creates a record for the replica", func(t *testing.T) { rs, requireState := newStore(t, nil) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, stor, nil, nil, false, false)) - require.NoError(t, rs.SetGeneration(ctx, 1, "storage-2", 0)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, stor, nil, nil, false, false)) + require.NoError(t, rs.SetGeneration(ctx, id, "storage-2", 0)) requireState(t, ctx, virtualStorageState{"virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }}, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1, generation: 0}, - "storage-2": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id, generation: 0}, + "storage-2": {repositoryID: id, generation: 0}, }, }, }, @@ -423,20 +436,21 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("updates existing record", func(t *testing.T) { rs, requireState := newStore(t, nil) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "storage-1", nil, nil, false, false)) - require.NoError(t, rs.SetGeneration(ctx, 1, stor, 1)) - require.NoError(t, rs.SetGeneration(ctx, 1, stor, 0)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "storage-1", nil, nil, false, false)) + require.NoError(t, rs.SetGeneration(ctx, id, stor, 1)) + require.NoError(t, rs.SetGeneration(ctx, id, stor, 0)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id, generation: 0}, }, }, }, @@ -456,19 +470,20 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("sets an existing replica as the latest", func(t *testing.T) { rs, requireState := newStore(t, nil) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "storage-1", []string{"storage-2"}, nil, false, false)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "storage-1", []string{"storage-2"}, nil, false, false)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1, generation: 0}, - "storage-2": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id, generation: 0}, + "storage-2": {repositoryID: id, generation: 0}, }, }, }, @@ -478,14 +493,14 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1, generation: 1}, - "storage-2": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id, generation: 1}, + "storage-2": {repositoryID: id, generation: 0}, }, }, }, @@ -494,18 +509,19 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("sets a new replica as the latest", func(t *testing.T) { rs, requireState := newStore(t, nil) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "storage-1", nil, nil, false, false)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "storage-1", nil, nil, false, false)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id, generation: 0}, }, }, }, @@ -515,14 +531,14 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1, generation: 0}, - "storage-2": {repositoryID: 1, generation: 1}, + "storage-1": {repositoryID: id, generation: 0}, + "storage-2": {repositoryID: id, generation: 1}, }, }, }, @@ -533,13 +549,16 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("GetGeneration", func(t *testing.T) { rs, _ := newStore(t, nil) - generation, err := rs.GetGeneration(ctx, 1, stor) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + + generation, err := rs.GetGeneration(ctx, id, stor) require.NoError(t, err) require.Equal(t, GenerationUnknown, generation) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, stor, nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, stor, nil, nil, false, false)) - generation, err = rs.GetGeneration(ctx, 1, stor) + generation, err = rs.GetGeneration(ctx, id, stor) require.NoError(t, err) require.Equal(t, 0, generation) }) @@ -547,48 +566,54 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("GetReplicatedGeneration", func(t *testing.T) { t.Run("no previous record allowed", func(t *testing.T) { rs, _ := newStore(t, nil) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) - gen, err := rs.GetReplicatedGeneration(ctx, 1, "source", "target") + gen, err := rs.GetReplicatedGeneration(ctx, id, "source", "target") require.NoError(t, err) require.Equal(t, GenerationUnknown, gen) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "source", nil, nil, false, false)) - gen, err = rs.GetReplicatedGeneration(ctx, 1, "source", "target") + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "source", nil, nil, false, false)) + gen, err = rs.GetReplicatedGeneration(ctx, id, "source", "target") require.NoError(t, err) require.Equal(t, 0, gen) }) t.Run("upgrade allowed", func(t *testing.T) { rs, _ := newStore(t, nil) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "source", nil, nil, false, false)) - require.NoError(t, rs.IncrementGeneration(ctx, 1, "source", nil)) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "source", nil, nil, false, false)) + require.NoError(t, rs.IncrementGeneration(ctx, id, "source", nil)) - gen, err := rs.GetReplicatedGeneration(ctx, 1, "source", "target") + gen, err := rs.GetReplicatedGeneration(ctx, id, "source", "target") require.NoError(t, err) require.Equal(t, 1, gen) - require.NoError(t, rs.SetGeneration(ctx, 1, "target", 0)) - gen, err = rs.GetReplicatedGeneration(ctx, 1, "source", "target") + require.NoError(t, rs.SetGeneration(ctx, id, "target", 0)) + gen, err = rs.GetReplicatedGeneration(ctx, id, "source", "target") require.NoError(t, err) require.Equal(t, 1, gen) }) t.Run("downgrade prevented", func(t *testing.T) { rs, _ := newStore(t, nil) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "target", nil, nil, false, false)) - require.NoError(t, rs.IncrementGeneration(ctx, 1, "target", nil)) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "target", nil, nil, false, false)) + require.NoError(t, rs.IncrementGeneration(ctx, id, "target", nil)) - _, err := rs.GetReplicatedGeneration(ctx, 1, "source", "target") + _, err = rs.GetReplicatedGeneration(ctx, id, "source", "target") require.Equal(t, DowngradeAttemptedError{"target", 1, GenerationUnknown}, err) - require.NoError(t, rs.SetGeneration(ctx, 1, "source", 1)) - _, err = rs.GetReplicatedGeneration(ctx, 1, "source", "target") + require.NoError(t, rs.SetGeneration(ctx, id, "source", 1)) + _, err = rs.GetReplicatedGeneration(ctx, id, "source", "target") require.Equal(t, DowngradeAttemptedError{"target", 1, 1}, err) - require.NoError(t, rs.SetGeneration(ctx, 1, "source", 0)) - _, err = rs.GetReplicatedGeneration(ctx, 1, "source", "target") + require.NoError(t, rs.SetGeneration(ctx, id, "source", 0)) + _, err = rs.GetReplicatedGeneration(ctx, id, "source", "target") require.Equal(t, DowngradeAttemptedError{"target", 1, 0}, err) }) }) @@ -657,26 +682,28 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { } { t.Run(tc.desc, func(t *testing.T) { rs, requireState := newStore(t, nil) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "primary", tc.updatedSecondaries, tc.outdatedSecondaries, tc.storePrimary, tc.storeAssignments)) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "primary", tc.updatedSecondaries, tc.outdatedSecondaries, tc.storePrimary, tc.storeAssignments)) expectedStorageState := storageState{ vs: { repo: { - "primary": {repositoryID: 1, generation: 0}, + "primary": {repositoryID: id, generation: 0}, }, }, } for _, updatedSecondary := range tc.updatedSecondaries { - expectedStorageState[vs][repo][updatedSecondary] = replicaRecord{repositoryID: 1, generation: 0} + expectedStorageState[vs][repo][updatedSecondary] = replicaRecord{repositoryID: id, generation: 0} } requireState(t, ctx, virtualStorageState{ vs: { repo: { - repositoryID: 1, + repositoryID: id, replicaPath: repo, primary: tc.expectedPrimary, assignments: tc.expectedAssignments, @@ -691,21 +718,25 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("conflict due to virtual storage and relative path", func(t *testing.T) { rs, _ := newStore(t, nil) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, stor, nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, stor, nil, nil, false, false)) require.Equal(t, RepositoryExistsError{vs, repo, stor}, - rs.CreateRepository(ctx, 2, vs, repo, stor, nil, nil, false, false), + rs.CreateRepository(ctx, -1, vs, repo, stor, nil, nil, false, false), ) }) t.Run("conflict due to repository id", func(t *testing.T) { rs, _ := newStore(t, nil) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) - require.NoError(t, rs.CreateRepository(ctx, 1, "virtual-storage-1", "relative-path-1", "storage-1", nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id, "virtual-storage-1", "relative-path-1", "storage-1", nil, nil, false, false)) require.Equal(t, - fmt.Errorf("repository id 1 already in use"), - rs.CreateRepository(ctx, 1, "virtual-storage-2", "relative-path-2", "storage-2", nil, nil, false, false), + fmt.Errorf("repository id %d already in use", id), + rs.CreateRepository(ctx, id, "virtual-storage-2", "relative-path-2", "storage-2", nil, nil, false, false), ) }) }) @@ -722,34 +753,40 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("delete existing", func(t *testing.T) { rs, requireState := newStore(t, nil) + id1, err := rs.ReserveRepositoryID(ctx, "virtual-storage-1", "repository-1") + require.NoError(t, err) + id2, err := rs.ReserveRepositoryID(ctx, "virtual-storage-2", "repository-2") + require.NoError(t, err) + id3, err := rs.ReserveRepositoryID(ctx, "virtual-storage-3", "repository-3") + require.NoError(t, err) - require.NoError(t, rs.CreateRepository(ctx, 1, "virtual-storage-1", "repository-1", "storage-1", nil, nil, false, false)) - require.NoError(t, rs.CreateRepository(ctx, 2, "virtual-storage-2", "repository-1", "storage-1", []string{"storage-2"}, nil, false, false)) - require.NoError(t, rs.CreateRepository(ctx, 3, "virtual-storage-2", "repository-2", "storage-1", nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id1, "virtual-storage-1", "repository-1", "storage-1", nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id2, "virtual-storage-2", "repository-1", "storage-1", []string{"storage-2"}, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id3, "virtual-storage-2", "repository-2", "storage-1", nil, nil, false, false)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id1, replicaPath: "repository-1"}, }, "virtual-storage-2": { - "repository-1": {repositoryID: 2, replicaPath: "repository-1"}, - "repository-2": {repositoryID: 3, replicaPath: "repository-2"}, + "repository-1": {repositoryID: id2, replicaPath: "repository-1"}, + "repository-2": {repositoryID: id3, replicaPath: "repository-2"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1}, + "storage-1": {repositoryID: id1}, }, }, "virtual-storage-2": { "repository-1": { - "storage-1": {repositoryID: 2}, - "storage-2": {repositoryID: 2}, + "storage-1": {repositoryID: id2}, + "storage-2": {repositoryID: id2}, }, "repository-2": { - "storage-1": {repositoryID: 3}, + "storage-1": {repositoryID: id3}, }, }, }, @@ -763,21 +800,21 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id1, replicaPath: "repository-1"}, }, "virtual-storage-2": { - "repository-2": {repositoryID: 3, replicaPath: "repository-2"}, + "repository-2": {repositoryID: id3, replicaPath: "repository-2"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "storage-1": {repositoryID: 1}, + "storage-1": {repositoryID: id1}, }, }, "virtual-storage-2": { "repository-2": { - "storage-1": {repositoryID: 3}, + "storage-1": {repositoryID: id3}, }, }, }, @@ -793,33 +830,39 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { }) t.Run("delete existing", func(t *testing.T) { - require.NoError(t, rs.CreateRepository(ctx, 1, "virtual-storage-1", "relative-path-1", "storage-1", []string{"storage-2"}, nil, false, false)) - require.NoError(t, rs.CreateRepository(ctx, 2, "virtual-storage-1", "relative-path-2", "storage-1", nil, nil, false, false)) - require.NoError(t, rs.CreateRepository(ctx, 3, "virtual-storage-2", "relative-path-1", "storage-1", nil, nil, false, false)) + id1, err := rs.ReserveRepositoryID(ctx, "virtual-storage-1", "relative-path-1") + require.NoError(t, err) + id2, err := rs.ReserveRepositoryID(ctx, "virtual-storage-2", "relative-path-2") + require.NoError(t, err) + id3, err := rs.ReserveRepositoryID(ctx, "virtual-storage-3", "relative-path-3") + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id1, "virtual-storage-1", "relative-path-1", "storage-1", []string{"storage-2"}, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id2, "virtual-storage-1", "relative-path-2", "storage-1", nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id3, "virtual-storage-2", "relative-path-1", "storage-1", nil, nil, false, false)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "relative-path-1": {repositoryID: 1, replicaPath: "relative-path-1"}, - "relative-path-2": {repositoryID: 2, replicaPath: "relative-path-2"}, + "relative-path-1": {repositoryID: id1, replicaPath: "relative-path-1"}, + "relative-path-2": {repositoryID: id2, replicaPath: "relative-path-2"}, }, "virtual-storage-2": { - "relative-path-1": {repositoryID: 3, replicaPath: "relative-path-1"}, + "relative-path-1": {repositoryID: id3, replicaPath: "relative-path-1"}, }, }, storageState{ "virtual-storage-1": { "relative-path-1": { - "storage-1": {repositoryID: 1, generation: 0}, - "storage-2": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id1, generation: 0}, + "storage-2": {repositoryID: id1, generation: 0}, }, "relative-path-2": { - "storage-1": {repositoryID: 2, generation: 0}, + "storage-1": {repositoryID: id2, generation: 0}, }, }, "virtual-storage-2": { "relative-path-1": { - "storage-1": {repositoryID: 3, generation: 0}, + "storage-1": {repositoryID: id3, generation: 0}, }, }, }, @@ -830,25 +873,25 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "relative-path-1": {repositoryID: 1, replicaPath: "relative-path-1"}, - "relative-path-2": {repositoryID: 2, replicaPath: "relative-path-2"}, + "relative-path-1": {repositoryID: id1, replicaPath: "relative-path-1"}, + "relative-path-2": {repositoryID: id2, replicaPath: "relative-path-2"}, }, "virtual-storage-2": { - "relative-path-1": {repositoryID: 3, replicaPath: "relative-path-1"}, + "relative-path-1": {repositoryID: id3, replicaPath: "relative-path-1"}, }, }, storageState{ "virtual-storage-1": { "relative-path-1": { - "storage-2": {repositoryID: 1, generation: 0}, + "storage-2": {repositoryID: id1, generation: 0}, }, "relative-path-2": { - "storage-1": {repositoryID: 2, generation: 0}, + "storage-1": {repositoryID: id2, generation: 0}, }, }, "virtual-storage-2": { "relative-path-1": { - "storage-1": {repositoryID: 3, generation: 0}, + "storage-1": {repositoryID: id3, generation: 0}, }, }, }, @@ -868,25 +911,28 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("rename existing", func(t *testing.T) { rs, requireState := newStore(t, nil) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, "renamed-all", "storage-1", nil, nil, false, false)) - require.NoError(t, rs.CreateRepository(ctx, 2, vs, "renamed-some", "storage-1", []string{"storage-2"}, nil, false, false)) + id1, err := rs.ReserveRepositoryID(ctx, vs, "renamed-all") + require.NoError(t, err) + id2, err := rs.ReserveRepositoryID(ctx, vs, "renamed-some") + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id1, vs, "renamed-all", "storage-1", nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, id2, vs, "renamed-some", "storage-1", []string{"storage-2"}, nil, false, false)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "renamed-all": {repositoryID: 1, replicaPath: "renamed-all"}, - "renamed-some": {repositoryID: 2, replicaPath: "renamed-some"}, + "renamed-all": {repositoryID: id1, replicaPath: "renamed-all"}, + "renamed-some": {repositoryID: id2, replicaPath: "renamed-some"}, }, }, storageState{ "virtual-storage-1": { "renamed-all": { - "storage-1": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id1, generation: 0}, }, "renamed-some": { - "storage-1": {repositoryID: 2, generation: 0}, - "storage-2": {repositoryID: 2, generation: 0}, + "storage-1": {repositoryID: id2, generation: 0}, + "storage-2": {repositoryID: id2, generation: 0}, }, }, }, @@ -898,20 +944,20 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "renamed-all-new": {repositoryID: 1, replicaPath: "renamed-all-new"}, - "renamed-some-new": {repositoryID: 2, replicaPath: "renamed-some-new"}, + "renamed-all-new": {repositoryID: id1, replicaPath: "renamed-all-new"}, + "renamed-some-new": {repositoryID: id2, replicaPath: "renamed-some-new"}, }, }, storageState{ "virtual-storage-1": { "renamed-all-new": { - "storage-1": {repositoryID: 1, generation: 0}, + "storage-1": {repositoryID: id1, generation: 0}, }, "renamed-some-new": { - "storage-1": {repositoryID: 2, generation: 0}, + "storage-1": {repositoryID: id2, generation: 0}, }, "renamed-some": { - "storage-2": {repositoryID: 2, generation: 0}, + "storage-2": {repositoryID: id2, generation: 0}, }, }, }, @@ -930,27 +976,28 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.Empty(t, replicaPath) require.Empty(t, secondaries) - replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, 1) + replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, 1e10) require.Equal(t, commonerr.ErrRepositoryNotFound, err) require.Empty(t, replicaPath) require.Empty(t, secondaries) }) - - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "primary", []string{"consistent-secondary"}, nil, false, false)) - require.NoError(t, rs.IncrementGeneration(ctx, 1, "primary", []string{"consistent-secondary"})) - require.NoError(t, rs.SetGeneration(ctx, 1, "inconsistent-secondary", 0)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "primary", []string{"consistent-secondary"}, nil, false, false)) + require.NoError(t, rs.IncrementGeneration(ctx, id, "primary", []string{"consistent-secondary"})) + require.NoError(t, rs.SetGeneration(ctx, id, "inconsistent-secondary", 0)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "primary": {repositoryID: 1, generation: 1}, - "consistent-secondary": {repositoryID: 1, generation: 1}, - "inconsistent-secondary": {repositoryID: 1, generation: 0}, + "primary": {repositoryID: id, generation: 1}, + "consistent-secondary": {repositoryID: id, generation: 1}, + "inconsistent-secondary": {repositoryID: id, generation: 0}, }, }, }, @@ -962,13 +1009,13 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.Equal(t, map[string]struct{}{"primary": {}, "consistent-secondary": {}}, secondaries) require.Equal(t, repo, replicaPath) - replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, 1) + replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, id) require.NoError(t, err) require.Equal(t, map[string]struct{}{"primary": {}, "consistent-secondary": {}}, secondaries) require.Equal(t, repo, replicaPath) }) - require.NoError(t, rs.SetGeneration(ctx, 1, "primary", 0)) + require.NoError(t, rs.SetGeneration(ctx, id, "primary", 0)) t.Run("outdated primary", func(t *testing.T) { replicaPath, secondaries, err := rs.GetConsistentStorages(ctx, vs, repo) @@ -976,28 +1023,28 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.Equal(t, map[string]struct{}{"consistent-secondary": {}}, secondaries) require.Equal(t, repo, replicaPath) - replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, 1) + replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, id) require.NoError(t, err) require.Equal(t, map[string]struct{}{"consistent-secondary": {}}, secondaries) require.Equal(t, repo, replicaPath) }) t.Run("storage with highest generation is not configured", func(t *testing.T) { - require.NoError(t, rs.SetGeneration(ctx, 1, "unknown", 2)) - require.NoError(t, rs.SetGeneration(ctx, 1, "primary", 1)) + require.NoError(t, rs.SetGeneration(ctx, id, "unknown", 2)) + require.NoError(t, rs.SetGeneration(ctx, id, "primary", 1)) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "unknown": {repositoryID: 1, generation: 2}, - "primary": {repositoryID: 1, generation: 1}, - "consistent-secondary": {repositoryID: 1, generation: 1}, - "inconsistent-secondary": {repositoryID: 1, generation: 0}, + "unknown": {repositoryID: id, generation: 2}, + "primary": {repositoryID: id, generation: 1}, + "consistent-secondary": {repositoryID: id, generation: 1}, + "inconsistent-secondary": {repositoryID: id, generation: 0}, }, }, }, @@ -1008,7 +1055,7 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.Equal(t, map[string]struct{}{"unknown": {}}, secondaries) require.Equal(t, repo, replicaPath) - replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, 1) + replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, id) require.NoError(t, err) require.Equal(t, map[string]struct{}{"unknown": {}}, secondaries) require.Equal(t, repo, replicaPath) @@ -1024,7 +1071,7 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.Empty(t, secondaries) require.Empty(t, replicaPath) - replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, 1) + replicaPath, secondaries, err = rs.GetConsistentStoragesByRepositoryID(ctx, id) require.Equal(t, commonerr.ErrRepositoryNotFound, err) require.Empty(t, secondaries) require.Empty(t, replicaPath) @@ -1034,25 +1081,29 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("DeleteInvalidRepository", func(t *testing.T) { t.Run("only replica", func(t *testing.T) { rs, requireState := newStore(t, nil) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "invalid-storage", nil, nil, false, false)) - require.NoError(t, rs.DeleteInvalidRepository(ctx, 1, "invalid-storage")) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "invalid-storage", nil, nil, false, false)) + require.NoError(t, rs.DeleteInvalidRepository(ctx, id, "invalid-storage")) requireState(t, ctx, virtualStorageState{}, storageState{}) }) t.Run("another replica", func(t *testing.T) { rs, requireState := newStore(t, nil) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, "invalid-storage", []string{"other-storage"}, nil, false, false)) - require.NoError(t, rs.DeleteInvalidRepository(ctx, 1, "invalid-storage")) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, "invalid-storage", []string{"other-storage"}, nil, false, false)) + require.NoError(t, rs.DeleteInvalidRepository(ctx, id, "invalid-storage")) requireState(t, ctx, virtualStorageState{ "virtual-storage-1": { - "repository-1": {repositoryID: 1, replicaPath: "repository-1"}, + "repository-1": {repositoryID: id, replicaPath: "repository-1"}, }, }, storageState{ "virtual-storage-1": { "repository-1": { - "other-storage": {repositoryID: 1, generation: 0}, + "other-storage": {repositoryID: id, generation: 0}, }, }, }, @@ -1067,7 +1118,9 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.NoError(t, err) require.False(t, exists) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, stor, nil, nil, false, false)) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, stor, nil, nil, false, false)) exists, err = rs.RepositoryExists(ctx, vs, repo) require.NoError(t, err) require.True(t, exists) @@ -1082,23 +1135,22 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { t.Run("ReserveRepositoryID", func(t *testing.T) { rs, _ := newStore(t, nil) - id, err := rs.ReserveRepositoryID(ctx, vs, repo) + idInitial, err := rs.ReserveRepositoryID(ctx, vs, repo) require.NoError(t, err) - require.Equal(t, int64(1), id) - id, err = rs.ReserveRepositoryID(ctx, vs, repo) + idNext, err := rs.ReserveRepositoryID(ctx, vs, repo) require.NoError(t, err) - require.Equal(t, int64(2), id) + require.NotEqual(t, idNext, idInitial) - require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, stor, nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, idNext, vs, repo, stor, nil, nil, false, false)) - id, err = rs.ReserveRepositoryID(ctx, vs, repo) + id, err := rs.ReserveRepositoryID(ctx, vs, repo) require.Equal(t, commonerr.ErrRepositoryAlreadyExists, err) require.Equal(t, int64(0), id) id, err = rs.ReserveRepositoryID(ctx, vs, repo+"-2") require.NoError(t, err) - require.Equal(t, int64(3), id) + require.NotEqual(t, idNext, id) }) t.Run("GetRepositoryID", func(t *testing.T) { @@ -1108,11 +1160,13 @@ func testRepositoryStore(t *testing.T, newStore repositoryStoreFactory) { require.Equal(t, commonerr.NewRepositoryNotFoundError(vs, repo), err) require.Equal(t, int64(0), id) - require.NoError(t, rs.CreateRepository(ctx, 1, vs, repo, stor, nil, nil, false, false)) + id, err = rs.ReserveRepositoryID(ctx, vs, repo) + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, vs, repo, stor, nil, nil, false, false)) - id, err = rs.GetRepositoryID(ctx, vs, repo) + idReturned, err := rs.GetRepositoryID(ctx, vs, repo) require.Nil(t, err) - require.Equal(t, int64(1), id) + require.Equal(t, id, idReturned) }) } diff --git a/internal/praefect/datastore/storage_cleanup_test.go b/internal/praefect/datastore/storage_cleanup_test.go index 1beb61af1..4ff18a00a 100644 --- a/internal/praefect/datastore/storage_cleanup_test.go +++ b/internal/praefect/datastore/storage_cleanup_test.go @@ -207,8 +207,21 @@ func TestStorageCleanup_Exists(t *testing.T) { db := glsql.NewDB(t) repoStore := NewPostgresRepositoryStore(db.DB, nil) - require.NoError(t, repoStore.CreateRepository(ctx, 0, "vs", "p/1", "g1", []string{"g2", "g3"}, nil, false, false)) - require.NoError(t, repoStore.CreateRepository(ctx, 1, "vs", "p/2", "g1", []string{"g2", "g3"}, nil, false, false)) + const ( + virtualStorage = "vs" + relativePath1 = "p/1" + relativePath2 = "p/2" + storage1 = "g1" + storage2 = "g2" + storage3 = "g3" + ) + id1, err := repoStore.ReserveRepositoryID(ctx, virtualStorage, relativePath1) + require.NoError(t, err) + id2, err := repoStore.ReserveRepositoryID(ctx, virtualStorage, relativePath2) + require.NoError(t, err) + + require.NoError(t, repoStore.CreateRepository(ctx, id1, virtualStorage, relativePath1, storage1, []string{storage2, storage3}, nil, false, false)) + require.NoError(t, repoStore.CreateRepository(ctx, id2, virtualStorage, relativePath2, storage1, []string{storage2, storage3}, nil, false, false)) storageCleanup := NewStorageCleanup(db.DB) for _, tc := range []struct { @@ -220,57 +233,57 @@ func TestStorageCleanup_Exists(t *testing.T) { }{ { desc: "multiple doesn't exist", - virtualStorage: "vs", - storage: "g1", - relativeReplicaPaths: []string{"p/1", "p/2", "path/x", "path/y"}, + virtualStorage: virtualStorage, + storage: storage1, + relativeReplicaPaths: []string{relativePath1, relativePath2, "path/x", "path/y"}, out: []RepositoryClusterPath{ - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "g1"}, RelativeReplicaPath: "path/x"}, - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "g1"}, RelativeReplicaPath: "path/y"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: storage1}, RelativeReplicaPath: "path/x"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: storage1}, RelativeReplicaPath: "path/y"}, }, }, { desc: "duplicates", - virtualStorage: "vs", - storage: "g1", - relativeReplicaPaths: []string{"p/1", "path/x", "path/x"}, + virtualStorage: virtualStorage, + storage: storage1, + relativeReplicaPaths: []string{relativePath1, "path/x", "path/x"}, out: []RepositoryClusterPath{ - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "g1"}, RelativeReplicaPath: "path/x"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: storage1}, RelativeReplicaPath: "path/x"}, }, }, { desc: "all exist", - virtualStorage: "vs", - storage: "g1", - relativeReplicaPaths: []string{"p/1", "p/2"}, + virtualStorage: virtualStorage, + storage: storage1, + relativeReplicaPaths: []string{relativePath1, relativePath2}, out: nil, }, { desc: "all doesn't exist", - virtualStorage: "vs", - storage: "g1", + virtualStorage: virtualStorage, + storage: storage1, relativeReplicaPaths: []string{"path/x", "path/y", "path/z"}, out: []RepositoryClusterPath{ - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "g1"}, RelativeReplicaPath: "path/x"}, - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "g1"}, RelativeReplicaPath: "path/y"}, - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "g1"}, RelativeReplicaPath: "path/z"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: storage1}, RelativeReplicaPath: "path/x"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: storage1}, RelativeReplicaPath: "path/y"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: storage1}, RelativeReplicaPath: "path/z"}, }, }, { desc: "doesn't exist because of storage", - virtualStorage: "vs", + virtualStorage: virtualStorage, storage: "stub", relativeReplicaPaths: []string{"path/x"}, out: []RepositoryClusterPath{ - {ClusterPath: ClusterPath{VirtualStorage: "vs", Storage: "stub"}, RelativeReplicaPath: "path/x"}, + {ClusterPath: ClusterPath{VirtualStorage: virtualStorage, Storage: "stub"}, RelativeReplicaPath: "path/x"}, }, }, { desc: "doesn't exist because of virtual storage", virtualStorage: "stub", - storage: "g1", + storage: storage1, relativeReplicaPaths: []string{"path/x"}, out: []RepositoryClusterPath{ - {ClusterPath: ClusterPath{VirtualStorage: "stub", Storage: "g1"}, RelativeReplicaPath: "path/x"}, + {ClusterPath: ClusterPath{VirtualStorage: "stub", Storage: storage1}, RelativeReplicaPath: "path/x"}, }, }, } { diff --git a/internal/praefect/info_service_test.go b/internal/praefect/info_service_test.go index a49284124..e45bfc73a 100644 --- a/internal/praefect/info_service_test.go +++ b/internal/praefect/info_service_test.go @@ -78,8 +78,10 @@ func TestInfoService_RepositoryReplicas(t *testing.T) { elector := nodes.NewPerRepositoryElector(tx) conns := nodeSet.Connections() rs := datastore.NewPostgresRepositoryStore(tx, conf.StorageNames()) + id, err := rs.ReserveRepositoryID(ctx, virtualStorage, testRepo.GetRelativePath()) + require.NoError(t, err) require.NoError(t, - rs.CreateRepository(ctx, 1, virtualStorage, testRepo.GetRelativePath(), "g-1", []string{"g-2", "g-3"}, nil, true, false), + rs.CreateRepository(ctx, id, virtualStorage, testRepo.GetRelativePath(), "g-1", []string{"g-2", "g-3"}, nil, true, false), ) cc, _, cleanup := runPraefectServer(t, ctx, conf, buildOptions{ diff --git a/internal/praefect/nodes/health_manager_test.go b/internal/praefect/nodes/health_manager_test.go index b5c44aafa..a36bb6759 100644 --- a/internal/praefect/nodes/health_manager_test.go +++ b/internal/praefect/nodes/health_manager_test.go @@ -142,7 +142,7 @@ func TestHealthManager(t *testing.T) { }, }, { - After: failoverTimeout, + After: failoverTimeout + time.Millisecond, PraefectName: "praefect-1", LocalStatus: LocalStatus{ "virtual-storage-1": { @@ -178,7 +178,7 @@ func TestHealthManager(t *testing.T) { HealthConsensus: map[string][]string{}, }, { - After: activePraefectTimeout, + After: activePraefectTimeout + time.Millisecond, PraefectName: "praefect-3", LocalStatus: LocalStatus{ "virtual-storage-1": { @@ -456,7 +456,7 @@ func TestHealthManager(t *testing.T) { }, }, { - After: failoverTimeout, + After: failoverTimeout + time.Millisecond, PraefectName: "praefect-1", LocalStatus: LocalStatus{ "virtual-storage-1": { diff --git a/internal/praefect/nodes/per_repository_test.go b/internal/praefect/nodes/per_repository_test.go index 184d31614..426cf384d 100644 --- a/internal/praefect/nodes/per_repository_test.go +++ b/internal/praefect/nodes/per_repository_test.go @@ -485,29 +485,32 @@ func TestPerRepositoryElector(t *testing.T) { }, } { t.Run(tc.desc, func(t *testing.T) { + var repositoryID int64 rs := datastore.NewPostgresRepositoryStore(db, nil) for virtualStorage, relativePaths := range tc.state { for relativePath, storages := range relativePaths { - repositoryID, err := rs.ReserveRepositoryID(ctx, virtualStorage, relativePath) + reservedRepositoryID, err := rs.ReserveRepositoryID(ctx, virtualStorage, relativePath) require.NoError(t, err) repoCreated := false for storage, record := range storages { if !repoCreated { repoCreated = true - require.NoError(t, rs.CreateRepository(ctx, repositoryID, virtualStorage, relativePath, storage, nil, nil, false, false)) + require.NoError(t, rs.CreateRepository(ctx, reservedRepositoryID, virtualStorage, relativePath, storage, nil, nil, false, false)) defer func(virtualStorage, relativePath string) { _, _, err = rs.DeleteRepository(ctx, virtualStorage, relativePath) require.NoError(t, err) }(virtualStorage, relativePath) + repositoryID = reservedRepositoryID } - require.NoError(t, rs.SetGeneration(ctx, repositoryID, storage, record.generation)) + require.NoError(t, rs.SetGeneration(ctx, reservedRepositoryID, storage, record.generation)) if record.assigned { _, err := db.ExecContext(ctx, ` - INSERT INTO repository_assignments VALUES ($1, $2, $3, $4) - `, virtualStorage, relativePath, storage, repositoryID) + INSERT INTO repository_assignments (virtual_storage, relative_path, storage, repository_id) + VALUES ($1, $2, $3, $4) + `, virtualStorage, relativePath, storage, reservedRepositoryID) require.NoError(t, err) } } @@ -534,7 +537,6 @@ func TestPerRepositoryElector(t *testing.T) { } previousPrimary := "" - const repositoryID int64 = 1 for _, step := range tc.steps { runElection := func(tx *glsql.TxWrapper) (string, *logrus.Entry) { diff --git a/internal/praefect/reconciler/reconciler_test.go b/internal/praefect/reconciler/reconciler_test.go index cf769462d..3d54fb316 100644 --- a/internal/praefect/reconciler/reconciler_test.go +++ b/internal/praefect/reconciler/reconciler_test.go @@ -3,9 +3,7 @@ package reconciler import ( "context" "database/sql" - "errors" "fmt" - "sort" "testing" "github.com/lib/pq" @@ -32,9 +30,11 @@ func TestReconciler(t *testing.T) { type repositories map[string]map[string]map[string]storageRecord type existingJobs []datastore.ReplicationEvent - // Jobs is a set of jobs that the reconciliation run might produce. Only one of the job - // sets is expected from a run at a time. - type jobs [][]datastore.ReplicationJob + type jobs []datastore.ReplicationJob + type jobsCondition struct { + oneOf bool + jobs jobs + } type storages map[string][]string configuredStorages := storages{ @@ -86,12 +86,12 @@ func TestReconciler(t *testing.T) { db := glsql.NewDB(t) for _, tc := range []struct { - desc string - healthyStorages storages - repositories repositories - existingJobs existingJobs - deletedRepositories map[string][]string - reconciliationJobs jobs + desc string + healthyStorages storages + repositories repositories + existingJobs existingJobs + deletedRepositories map[string][]string + reconciliationJobCondition jobsCondition }{ { desc: "no repositories", @@ -126,8 +126,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -160,20 +160,22 @@ func TestReconciler(t *testing.T) { return repos }(), - reconciliationJobs: func() jobs { - var generated []datastore.ReplicationJob - for i := 0; i < 2*logBatchSize+1; i++ { - generated = append(generated, datastore.ReplicationJob{ - Change: datastore.UpdateRepo, - VirtualStorage: "virtual-storage-1", - RelativePath: fmt.Sprintf("relative-path-%d", i), - SourceNodeStorage: "storage-1", - TargetNodeStorage: "storage-2", - }) - } + reconciliationJobCondition: jobsCondition{ + jobs: func() jobs { + var generated jobs + for i := 0; i < 2*logBatchSize+1; i++ { + generated = append(generated, datastore.ReplicationJob{ + Change: datastore.UpdateRepo, + VirtualStorage: "virtual-storage-1", + RelativePath: fmt.Sprintf("relative-path-%d", i), + SourceNodeStorage: "storage-1", + TargetNodeStorage: "storage-2", + }) + } - return jobs{generated} - }(), + return generated + }(), + }, }, { desc: "no healthy source to reconcile from", @@ -203,8 +205,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -226,8 +228,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -261,8 +263,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -320,8 +322,8 @@ func TestReconciler(t *testing.T) { TargetNodeStorage: "storage-1", }, ), - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -357,8 +359,8 @@ func TestReconciler(t *testing.T) { TargetNodeStorage: "storage-2", }, ), - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -404,8 +406,8 @@ func TestReconciler(t *testing.T) { TargetNodeStorage: "storage-3", }, ), - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -442,8 +444,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -473,8 +475,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -504,8 +506,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.DeleteReplica, VirtualStorage: "virtual-storage-1", @@ -527,17 +529,15 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + oneOf: true, + jobs: jobs{ { Change: datastore.DeleteReplica, VirtualStorage: "virtual-storage-1", RelativePath: "relative-path-1", TargetNodeStorage: "storage-2", - }, - }, - { - { + }, { Change: datastore.DeleteReplica, VirtualStorage: "virtual-storage-1", RelativePath: "relative-path-1", @@ -569,8 +569,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.DeleteReplica, VirtualStorage: "virtual-storage-1", @@ -591,8 +591,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -626,8 +626,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.UpdateRepo, VirtualStorage: "virtual-storage-1", @@ -823,8 +823,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.DeleteReplica, VirtualStorage: "virtual-storage-1", @@ -1011,8 +1011,8 @@ func TestReconciler(t *testing.T) { }, }, }, - reconciliationJobs: jobs{ - { + reconciliationJobCondition: jobsCondition{ + jobs: jobs{ { Change: datastore.DeleteReplica, VirtualStorage: "virtual-storage-1", @@ -1028,19 +1028,15 @@ func TestReconciler(t *testing.T) { defer cancel() db.Truncate(t, "replication_queue") - db.SequenceReset(t) // set up the repository generation records expected by the test case rs := datastore.NewPostgresRepositoryStore(db, configuredStorages) for virtualStorage, relativePaths := range tc.repositories { for relativePath, storages := range relativePaths { var repositoryID int64 - repoCreated := false for storage, repo := range storages { if repo.generation >= 0 { - if !repoCreated { - repoCreated = true - + if repositoryID == 0 { var err error repositoryID, err = rs.ReserveRepositoryID(ctx, virtualStorage, relativePath) require.NoError(t, err) @@ -1049,9 +1045,7 @@ func TestReconciler(t *testing.T) { defer func(virtualStorage, relativePath string) { _, _, err := rs.DeleteRepository(ctx, virtualStorage, relativePath) - if !errors.As(err, &commonerr.RepositoryNotFoundError{}) { - require.NoError(t, err) - } + require.NoError(t, err) }(virtualStorage, relativePath) } @@ -1061,10 +1055,12 @@ func TestReconciler(t *testing.T) { for storage, repo := range storages { if repo.assigned { - _, err := db.ExecContext(ctx, ` - INSERT INTO repository_assignments (repository_id, virtual_storage, relative_path, storage) - VALUES ($1, $2, $3, $4) - `, repositoryID, virtualStorage, relativePath, storage) + _, err := db.ExecContext( + ctx, + `INSERT INTO repository_assignments (repository_id, virtual_storage, relative_path, storage) + VALUES ($1, $2, $3, $4)`, + repositoryID, virtualStorage, relativePath, storage, + ) require.NoError(t, err) } } @@ -1131,15 +1127,13 @@ func TestReconciler(t *testing.T) { } // Fill the expected reconciliation jobs with generated repository IDs. - for _, jobs := range tc.reconciliationJobs { - for i, job := range jobs { - id, err := rs.GetRepositoryID(ctx, job.VirtualStorage, job.RelativePath) - if err != nil { - require.Equal(t, commonerr.NewRepositoryNotFoundError(job.VirtualStorage, job.RelativePath), err) - } - - jobs[i].RepositoryID = id + for i, job := range tc.reconciliationJobCondition.jobs { + id, err := rs.GetRepositoryID(ctx, job.VirtualStorage, job.RelativePath) + if err != nil { + require.Equal(t, commonerr.NewRepositoryNotFoundError(job.VirtualStorage, job.RelativePath), err) } + + tc.reconciliationJobCondition.jobs[i].RepositoryID = id } // run reconcile in two concurrent transactions to ensure everything works @@ -1171,7 +1165,7 @@ func TestReconciler(t *testing.T) { require.NoError(t, err) defer rows.Close() - actualJobs := make([]datastore.ReplicationJob, 0, len(tc.reconciliationJobs)) + actualJobs := make(jobs, 0, len(tc.reconciliationJobCondition.jobs)) for rows.Next() { var job datastore.ReplicationJob var meta datastore.Params @@ -1181,22 +1175,11 @@ func TestReconciler(t *testing.T) { } require.NoError(t, rows.Err()) - - expectedJobs := tc.reconciliationJobs - if expectedJobs == nil { - // If the test case defined there are no jobs to be produced, just set an empty slice so the - // require.Contains matches the empty set. - expectedJobs = jobs{{}} - } - - // Sort the jobs so the require.Contains works below, the order of the produced jobs is not important. - for _, jobs := range append(expectedJobs, actualJobs) { - sort.Slice(jobs, func(i, j int) bool { - return jobs[i].VirtualStorage+jobs[i].RelativePath < jobs[j].VirtualStorage+jobs[j].RelativePath - }) + if tc.reconciliationJobCondition.oneOf { + require.Subset(t, tc.reconciliationJobCondition.jobs, actualJobs) + } else { + require.ElementsMatch(t, tc.reconciliationJobCondition.jobs, actualJobs) } - - require.Contains(t, expectedJobs, actualJobs) }) } } diff --git a/internal/praefect/replicator_pg_test.go b/internal/praefect/replicator_pg_test.go index 7f79191a6..7e1d3bc1e 100644 --- a/internal/praefect/replicator_pg_test.go +++ b/internal/praefect/replicator_pg_test.go @@ -51,7 +51,9 @@ func TestReplicatorInvalidSourceRepository(t *testing.T) { rs := datastore.NewPostgresRepositoryStore(glsql.NewDB(t), nil) - require.NoError(t, rs.CreateRepository(ctx, 1, "virtual-storage-1", "relative-path-1", "gitaly-1", nil, nil, true, false)) + id, err := rs.ReserveRepositoryID(ctx, "virtual-storage-1", "relative-path-1") + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, "virtual-storage-1", "relative-path-1", "gitaly-1", nil, nil, true, false)) exists, err := rs.RepositoryExists(ctx, "virtual-storage-1", "relative-path-1") require.NoError(t, err) @@ -60,7 +62,7 @@ func TestReplicatorInvalidSourceRepository(t *testing.T) { r := &defaultReplicator{rs: rs, log: testhelper.DiscardTestLogger(t)} require.NoError(t, r.Replicate(ctx, datastore.ReplicationEvent{ Job: datastore.ReplicationJob{ - RepositoryID: 1, + RepositoryID: id, VirtualStorage: "virtual-storage-1", RelativePath: "relative-path-1", SourceNodeStorage: "gitaly-1", @@ -92,7 +94,9 @@ func TestReplicatorDestroy(t *testing.T) { ctx, cancel := testhelper.Context() defer cancel() - require.NoError(t, rs.CreateRepository(ctx, 1, "virtual-storage-1", "relative-path-1", "storage-1", []string{"storage-2"}, nil, false, false)) + id, err := rs.ReserveRepositoryID(ctx, "virtual-storage-1", "relative-path-1") + require.NoError(t, err) + require.NoError(t, rs.CreateRepository(ctx, id, "virtual-storage-1", "relative-path-1", "storage-1", []string{"storage-2"}, nil, false, false)) ln, err := net.Listen("tcp", "localhost:0") require.NoError(t, err) diff --git a/internal/praefect/repocleaner/repository_test.go b/internal/praefect/repocleaner/repository_test.go index 4cbed8cae..a78803b3a 100644 --- a/internal/praefect/repocleaner/repository_test.go +++ b/internal/praefect/repocleaner/repository_test.go @@ -89,7 +89,7 @@ func TestRunner_Run(t *testing.T) { defer cancel() repoStore := datastore.NewPostgresRepositoryStore(db.DB, nil) - for i, set := range []struct { + for _, set := range []struct { relativePath string primary string secondaries []string @@ -110,7 +110,9 @@ func TestRunner_Run(t *testing.T) { secondaries: []string{storage2, storage3}, }, } { - require.NoError(t, repoStore.CreateRepository(ctx, int64(i), conf.VirtualStorages[0].Name, set.relativePath, set.primary, set.secondaries, nil, false, false)) + id, err := repoStore.ReserveRepositoryID(ctx, conf.VirtualStorages[0].Name, set.relativePath) + require.NoError(t, err) + require.NoError(t, repoStore.CreateRepository(ctx, id, conf.VirtualStorages[0].Name, set.relativePath, set.primary, set.secondaries, nil, false, false)) } logger, loggerHook := test.NewNullLogger() diff --git a/internal/praefect/router_per_repository_test.go b/internal/praefect/router_per_repository_test.go index dd2be1ecc..c23abea2c 100644 --- a/internal/praefect/router_per_repository_test.go +++ b/internal/praefect/router_per_repository_test.go @@ -445,6 +445,8 @@ func TestPerRepositoryRouter_RouteRepositoryCreation(t *testing.T) { return actual.Secondaries[i].Storage < actual.Secondaries[j].Storage }) sort.Strings(actual.ReplicationTargets) + // Hard code it here as there is no way we can predict the value of the sequence + actual.RepositoryID = 1 require.Contains(t, expected, actual) } } @@ -601,14 +603,15 @@ func TestPerRepositoryRouter_RouteRepositoryCreation(t *testing.T) { ctx, cancel := testhelper.Context() defer cancel() - db.SequenceReset(t) tx := db.Begin(t) defer tx.Rollback(t) rs := datastore.NewPostgresRepositoryStore(tx, nil) if tc.repositoryExists { + id, err := rs.ReserveRepositoryID(ctx, "virtual-storage-1", "relative-path") + require.NoError(t, err) require.NoError(t, - rs.CreateRepository(ctx, 1, "virtual-storage-1", relativePath, "primary", nil, nil, true, true), + rs.CreateRepository(ctx, id, "virtual-storage-1", relativePath, "primary", nil, nil, true, true), ) } |