Add Database/Schema support to SQL Server and PostgreSQL persisters

docs/testing-persistence.md

@@ -8,6 +8,15 @@ ServiceControl supports multiple persistence types
 
 All persistence test projects can be run with `dotnet test` against the corresponding test project in `src/`.
 
+## Test isolation on the SQL persisters
+
+The SQL Server and PostgreSQL suites give each test its own **schema** in one shared database, named `sc_test_<guid>` for persistence tests and `sc_at_<guid>` for acceptance tests, and drop it on teardown. This uses the same `Database/Schema` setting that is offered to customers, so every run exercises that feature.
+
+Two consequences are worth knowing:
+
+* The connection string is used exactly as it is given, so **the database it names must already exist**. The test containers create one; a server you point the environment variable at will not.
+* A test failure that leaves a process behind can leave a schema behind with it. `SELECT nspname FROM pg_namespace WHERE nspname LIKE 'sc\_%'` and `SELECT name FROM sys.schemas WHERE name LIKE 'sc[_]%'` will find any strays.
+
 ## RavenDB
 
 RavenDB persistence tests start an embedded RavenDB instance for the duration of the test run.
@@ -23,7 +32,7 @@ Build that image locally before running SQL Server persistence tests:
 docker buildx build --platform=linux/amd64 --tag particular/servicecontrol-testing-sqlserver:latest ./src/Scripts/Docker/servicecontrol-testing-sqlserver
 ```
 
-If you want to use an existing SQL Server instance instead of a test container, set the `ServiceControl_Persistence_SqlServer_ConnectionString` environment variable to a valid SQL Server connection string.
+If you want to use an existing SQL Server instance instead of a test container, set the `ServiceControl_Persistence_SqlServer_ConnectionString` environment variable to a valid SQL Server connection string. It must name a database that exists, not `master`, because the tests create their schemas in whatever database it points at. The test container creates a `ServiceControlTests` database for this.
 
 ## PostgreSQL
 
@@ -35,4 +44,4 @@ If you want to use an existing PostgreSQL instance instead of a test container,
 ServiceControl_Persistence_PostgreSql_ConnectionString
 ```
 
-to a valid PostgreSQL connection string.
+to a valid PostgreSQL connection string. It must name a database that exists, because the tests create their schemas in whatever database it points at. The test container creates a `servicecontroltests` database for this, so that test schemas do not end up in the `postgres` maintenance database.

src/ServiceControl.AcceptanceTests.PostgreSql/AcceptanceTestStorageConfiguration.cs

@@ -1,10 +1,10 @@
 namespace ServiceControl.AcceptanceTests.PostgreSql;
 
 using System;
+using System.Collections.Concurrent;
 using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
-using Npgsql;
 using ServiceBus.Management.Infrastructure.Settings;
 using ServiceControl.AcceptanceTests.TestSupport;
 using ServiceControl.Persistence.EFCore.Abstractions;
@@ -17,53 +17,39 @@ public class AcceptanceTestStorageConfiguration : IAcceptanceTestStorageConfigur
 
     public async Task CustomizeSettings(Settings settings, CancellationToken cancellationToken = default)
     {
-        databaseName = $"sc_at_{Guid.NewGuid():n}";
-        serverConnectionString = await PostgreSqlSharedContainer.GetConnectionStringAsync(cancellationToken).ConfigureAwait(false);
+        var schema = $"sc_at_{Guid.NewGuid():n}";
+        var bodyStoragePath = Directory.CreateTempSubdirectory("sc_at_bodies_").FullName;
 
-        var connectionStringBuilder = new NpgsqlConnectionStringBuilder(serverConnectionString)
-        {
-            Database = databaseName
-        };
+        connectionString = await PostgreSqlSharedContainer.GetConnectionStringAsync(cancellationToken).ConfigureAwait(false);
+        await TestSchema.Create(connectionString, schema, cancellationToken).ConfigureAwait(false);
 
-        bodyStoragePath = Directory.CreateTempSubdirectory("sc_at_bodies_").FullName;
+        // A test that runs more than one scenario comes back through here, and the runner cleans up
+        // after each one. Recording everything created, rather than keeping only the most recent,
+        // is what stops the earlier schema being stranded in the shared database.
+        schemas.Add(schema);
+        bodyStoragePaths.Add(bodyStoragePath);
 
         settings.PersisterSpecificSettings = new PostgreSqlPersisterSettings
         {
-            ConnectionString = connectionStringBuilder.ConnectionString,
+            ConnectionString = connectionString,
+            Schema = schema,
             ErrorRetentionPeriod = TimeSpan.FromDays(10),
             BodyStorage = new FileSystemBodyStorageSettings { StoragePath = bodyStoragePath }
         };
     }
 
     public async Task Cleanup(CancellationToken cancellationToken = default)
     {
-        if (Interlocked.Exchange(ref cleanupStarted, 1) != 0)
-        {
-            return;
-        }
-
         try
         {
-            if (serverConnectionString == null || databaseName == null)
+            while (schemas.TryTake(out var schema))
             {
-                return;
-            }
-
-            var connection = new NpgsqlConnection(serverConnectionString);
-            await using (connection.ConfigureAwait(false))
-            {
-                await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
-                var command = connection.CreateCommand();
-                await using (command.ConfigureAwait(false))
-                {
-                    command.CommandText = $"DROP DATABASE IF EXISTS \"{databaseName}\" WITH (FORCE)";
-                    await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
-                }
+                await TestSchema.Drop(connectionString, schema, cancellationToken).ConfigureAwait(false);
             }
         }
         finally
         {
-            if (bodyStoragePath != null)
+            while (bodyStoragePaths.TryTake(out var bodyStoragePath))
             {
                 try
                 {
@@ -91,8 +77,7 @@ public void Dispose()
         }
     }
 
-    string serverConnectionString;
-    string databaseName;
-    string bodyStoragePath;
-    int cleanupStarted;
-}
+    readonly ConcurrentBag<string> schemas = [];
+    readonly ConcurrentBag<string> bodyStoragePaths = [];
+    string connectionString;
+}

src/ServiceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj

@@ -32,6 +32,7 @@
   <ItemGroup>
     <Compile Include="..\ServiceControl.AcceptanceTests\**\*.cs" LinkBase="Shared" />
     <Compile Include="..\ServiceControl.Persistence.Tests.PostgreSql\PostgreSqlSharedContainer.cs" />
+    <Compile Include="..\ServiceControl.Persistence.Tests.PostgreSql\TestSchema.cs" />
     <Compile Include="..\ServiceControl.Persistence.Tests.PostgreSql\StopSharedPostgreSql.cs" />
     <Compile Include="..\ServiceControl.UnitTests\NUnitParallelRunnerSettings.cs" />
 

src/ServiceControl.AcceptanceTests.SqlServer/AcceptanceTestStorageConfiguration.cs

@@ -1,10 +1,10 @@
 namespace ServiceControl.AcceptanceTests.SqlServer;
 
 using System;
+using System.Collections.Concurrent;
 using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
-using Microsoft.Data.SqlClient;
 using ServiceBus.Management.Infrastructure.Settings;
 using ServiceControl.AcceptanceTests.TestSupport;
 using ServiceControl.Persistence.EFCore.Abstractions;
@@ -17,59 +17,39 @@ public class AcceptanceTestStorageConfiguration : IAcceptanceTestStorageConfigur
 
     public async Task CustomizeSettings(Settings settings, CancellationToken cancellationToken = default)
     {
-        databaseName = $"sc_at_{Guid.NewGuid():n}";
-        serverConnectionString = await SqlServerSharedContainer.GetConnectionStringAsync(cancellationToken).ConfigureAwait(false);
+        var schema = $"sc_at_{Guid.NewGuid():n}";
+        var bodyStoragePath = Directory.CreateTempSubdirectory("sc_at_bodies_").FullName;
 
-        var connectionStringBuilder = new SqlConnectionStringBuilder(serverConnectionString)
-        {
-            InitialCatalog = databaseName
-        };
+        connectionString = await SqlServerSharedContainer.GetConnectionStringAsync(cancellationToken).ConfigureAwait(false);
+        await TestSchema.Create(connectionString, schema, cancellationToken).ConfigureAwait(false);
 
-        bodyStoragePath = Directory.CreateTempSubdirectory("sc_at_bodies_").FullName;
+        // A test that runs more than one scenario comes back through here, and the runner cleans up
+        // after each one. Recording everything created, rather than keeping only the most recent,
+        // is what stops the earlier schema being stranded in the shared database.
+        schemas.Add(schema);
+        bodyStoragePaths.Add(bodyStoragePath);
 
         settings.PersisterSpecificSettings = new SqlServerPersisterSettings
         {
-            ConnectionString = connectionStringBuilder.ConnectionString,
+            ConnectionString = connectionString,
+            Schema = schema,
             ErrorRetentionPeriod = TimeSpan.FromDays(10),
             BodyStorage = new FileSystemBodyStorageSettings { StoragePath = bodyStoragePath }
         };
     }
 
     public async Task Cleanup(CancellationToken cancellationToken = default)
     {
-        if (Interlocked.Exchange(ref cleanupStarted, 1) != 0)
-        {
-            return;
-        }
-
         try
         {
-            if (serverConnectionString == null || databaseName == null)
+            while (schemas.TryTake(out var schema))
             {
-                return;
-            }
-
-            var connection = new SqlConnection(serverConnectionString);
-            await using (connection.ConfigureAwait(false))
-            {
-                await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
-                var command = connection.CreateCommand();
-                await using (command.ConfigureAwait(false))
-                {
-                    command.CommandText = $"""
-                        IF DB_ID('{databaseName}') IS NOT NULL
-                        BEGIN
-                            ALTER DATABASE [{databaseName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE;
-                            DROP DATABASE [{databaseName}];
-                        END
-                        """;
-                    await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
-                }
+                await TestSchema.Drop(connectionString, schema, cancellationToken).ConfigureAwait(false);
             }
         }
         finally
         {
-            if (bodyStoragePath != null)
+            while (bodyStoragePaths.TryTake(out var bodyStoragePath))
             {
                 try
                 {
@@ -97,8 +77,7 @@ public void Dispose()
         }
     }
 
-    string serverConnectionString;
-    string databaseName;
-    string bodyStoragePath;
-    int cleanupStarted;
+    readonly ConcurrentBag<string> schemas = [];
+    readonly ConcurrentBag<string> bodyStoragePaths = [];
+    string connectionString;
 }

src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj

@@ -32,6 +32,7 @@
   <ItemGroup>
     <Compile Include="..\ServiceControl.AcceptanceTests\**\*.cs" LinkBase="Shared" />
     <Compile Include="..\ServiceControl.Persistence.Tests.SqlServer\SqlServerSharedContainer.cs" />
+    <Compile Include="..\ServiceControl.Persistence.Tests.SqlServer\TestSchema.cs" />
     <Compile Include="..\ServiceControl.Persistence.Tests.SqlServer\StopSharedSqlServer.cs" />
     <Compile Include="..\ServiceControl.UnitTests\NUnitParallelRunnerSettings.cs" />
 

src/ServiceControl.Persistence.EFCore.PostgreSql/FullTextSearchSql.cs

@@ -1,5 +1,7 @@
 namespace ServiceControl.Persistence.EFCore.PostgreSql;
 
+using Microsoft.EntityFrameworkCore.Migrations.Operations;
+
 /// <summary>
 /// Full text search DDL for the failed messages table. EF Core cannot model a GIN index over an
 /// expression, so it is applied by the AddFullTextSearch migration. The statements live here, and
@@ -9,6 +11,7 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql;
 static class FullTextSearchSql
 {
     const string IndexName = "ix_failed_messages_full_text";
+    const string TableName = "failed_messages";
 
     // 'simple' rather than 'english': message and header content is technical, stemming and
     // stopword removal do more harm than good.
@@ -26,7 +29,34 @@ static class FullTextSearchSql
     public const string IndexedExpression =
         $"""to_tsvector('{Configuration}', headers_json || ' ' || COALESCE(body_text, '') || ' ' || replace(replace(COALESCE(message_type, ''), '.', ' '), '+', ' '))""";
 
-    public const string Up = $"CREATE INDEX {IndexName} ON failed_messages USING GIN ({IndexedExpression})";
+    public static readonly string Up = CreateIndexSql(null);
+
+    public static readonly string Down = DropIndexSql(null);
+
+    /// <summary>
+    /// Re-renders the statement the migration carries, this time with the configured schema in it.
+    /// Anything else is left alone: EF Core builds the migrations history table's own SQL through
+    /// the same generator, already pointed at the right schema. MigrationSqlIsSchemaAwareTests is
+    /// what catches a statement of ours that should have been listed here.
+    /// </summary>
+    public static MigrationOperation Rewrite(SqlOperation operation, string schema) =>
+        operation.Sql switch
+        {
+            var sql when sql == Up => WithSql(operation, CreateIndexSql(schema)),
+            var sql when sql == Down => WithSql(operation, DropIndexSql(schema)),
+            _ => operation
+        };
+
+    public static bool IsHandled(string sql) => sql == Up || sql == Down;
+
+    static string CreateIndexSql(string? schema) =>
+        $"CREATE INDEX {IndexName} ON {Qualify(schema, TableName)} USING GIN ({IndexedExpression})";
+
+    // An index belongs to its table's schema, so it is the index that gets qualified here.
+    static string DropIndexSql(string? schema) => $"DROP INDEX IF EXISTS {Qualify(schema, IndexName)}";
+
+    static string Qualify(string? schema, string name) => schema is null ? name : $"\"{schema}\".{name}";
 
-    public const string Down = $"DROP INDEX IF EXISTS {IndexName}";
+    static SqlOperation WithSql(SqlOperation operation, string sql) =>
+        new() { Sql = sql, SuppressTransaction = operation.SuppressTransaction };
 }

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDatabaseMigrator.cs

@@ -14,10 +14,31 @@ public async Task ApplyMigrations(CancellationToken cancellationToken = default)
         var previousTimeout = dbContext.Database.GetCommandTimeout();
         dbContext.Database.SetCommandTimeout(EFPersisterSettings.MigrationCommandTimeout);
 
+        await RequireSchema(cancellationToken);
         await dbContext.Database.MigrateAsync(cancellationToken);
 
         dbContext.Database.SetCommandTimeout(previousTimeout);
 
         logger.LogInformation("PostgreSQL database migration completed");
     }
+
+    // EF Core would create the schema on its way to creating the migrations history table, which
+    // would turn a misspelled Database/Schema into a silently empty instance rather than an error.
+    async Task RequireSchema(CancellationToken cancellationToken)
+    {
+        if (dbContext.Schema is null)
+        {
+            return;
+        }
+
+        var exists = await dbContext.Database
+            .SqlQueryRaw<int>("""SELECT CASE WHEN EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = {0}) THEN 1 ELSE 0 END AS "Value" """, dbContext.Schema)
+            .SingleAsync(cancellationToken);
+
+        if (exists == 0)
+        {
+            throw new InvalidOperationException(
+                $"The configured schema '{dbContext.Schema}' does not exist in the database. ServiceControl does not create schemas, the same way it does not create the database. Create the schema, grant the configured user rights on it, and run setup again.");
+        }
+    }
 }

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs

@@ -4,9 +4,16 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Storage;
 using ServiceControl.Persistence.EFCore.DbContexts;
+using ServiceControl.Persistence.EFCore.Infrastructure;
 
 abstract class PostgreSqlDialect
 {
+    /// <summary>
+    /// The delimited, schema qualified name of the table an entity is mapped to. Every statement
+    /// below names its target this way so that a configured schema reaches the raw SQL too.
+    /// </summary>
+    protected static string Table<TEntity>(ServiceControlDbContext dbContext) => SchemaQualifiedTableName.For<TEntity>(dbContext);
+
     protected static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumerable<object?[]> rows, CancellationToken cancellationToken = default)
     {
         await using var command = dbContext.Database.GetDbConnection().CreateCommand();