diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index d28a7f9ceb..0ca0c404c2 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -12,6 +12,7 @@ + diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Infrastructure/KnownEndpointsReconciler.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Infrastructure/KnownEndpointsReconciler.cs new file mode 100644 index 0000000000..666312ee06 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Infrastructure/KnownEndpointsReconciler.cs @@ -0,0 +1,46 @@ +namespace ServiceControl.Persistence.EFCore.PostgreSql.Infrastructure; + +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class KnownEndpointsReconciler( + ILogger logger, + TimeProvider timeProvider, + IServiceScopeFactory serviceScopeFactory) + : InsertOnlyTableReconciler( + logger, timeProvider, serviceScopeFactory, nameof(KnownEndpointsReconciler)) +{ + protected override Task ReconcileBatch(ServiceControlDbContext dbContext, CancellationToken stoppingToken) => + ReconcileBatch(dbContext, BatchSize, stoppingToken); + + // Static so tests can execute a batch deterministically without the background service's timer loop. + // Must be called within an active transaction because of the pg_try_advisory_xact_lock. + internal static async Task ReconcileBatch(ServiceControlDbContext dbContext, int batchSize, CancellationToken cancellationToken) + { + var sql = """ + WITH lock_check AS ( + SELECT pg_try_advisory_xact_lock(hashtext('known_endpoints_sync')) AS acquired + ), + batch AS ( + SELECT ctid FROM "known_endpoints_insert_only" + WHERE (SELECT acquired FROM lock_check) + LIMIT @batchSize + ), + ins AS ( + INSERT INTO "known_endpoints" ("id", "name", "host_id", "host", "monitored") + SELECT DISTINCT ON ("known_endpoint_id") "known_endpoint_id", "name", "host_id", "host", FALSE + FROM "known_endpoints_insert_only" + WHERE ctid IN (SELECT ctid FROM batch) + ON CONFLICT ("id") DO NOTHING + ) + DELETE FROM "known_endpoints_insert_only" + WHERE ctid IN (SELECT ctid FROM batch); + """; + + var rowsAffected = await dbContext.Database.ExecuteSqlRawAsync(sql, [new Npgsql.NpgsqlParameter("@batchSize", batchSize)], cancellationToken); + return rowsAffected; + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260720230745_Initial.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260720230745_Initial.Designer.cs new file mode 100644 index 0000000000..1a1383b3e2 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260720230745_Initial.Designer.cs @@ -0,0 +1,93 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; +using ServiceControl.Persistence.EFCore.PostgreSql; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + [DbContext(typeof(PostgreSqlServiceControlDbContext))] + [Migration("20260720230745_Initial")] + partial class Initial + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Host") + .IsRequired() + .HasColumnType("text") + .HasColumnName("host"); + + b.Property("HostId") + .HasColumnType("uuid") + .HasColumnName("host_id"); + + b.Property("Monitored") + .HasColumnType("boolean") + .HasColumnName("monitored"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text") + .HasColumnName("name"); + + b.HasKey("Id"); + + b.ToTable("known_endpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointInsertOnlyEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Host") + .IsRequired() + .HasColumnType("text") + .HasColumnName("host"); + + b.Property("HostId") + .HasColumnType("uuid") + .HasColumnName("host_id"); + + b.Property("KnownEndpointId") + .HasColumnType("uuid") + .HasColumnName("known_endpoint_id"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text") + .HasColumnName("name"); + + b.HasKey("Id"); + + b.HasIndex("KnownEndpointId"); + + b.ToTable("known_endpoints_insert_only", (string)null); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260720230745_Initial.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260720230745_Initial.cs new file mode 100644 index 0000000000..d2caaa189a --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260720230745_Initial.cs @@ -0,0 +1,62 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class Initial : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "known_endpoints", + columns: table => new + { + id = table.Column(type: "uuid", nullable: false), + name = table.Column(type: "text", nullable: false), + host_id = table.Column(type: "uuid", nullable: false), + host = table.Column(type: "text", nullable: false), + monitored = table.Column(type: "boolean", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_known_endpoints", x => x.id); + }); + + migrationBuilder.CreateTable( + name: "known_endpoints_insert_only", + columns: table => new + { + id = table.Column(type: "bigint", nullable: false) + .Annotation("Npgsql:ValueGenerationStrategy", NpgsqlValueGenerationStrategy.IdentityByDefaultColumn), + known_endpoint_id = table.Column(type: "uuid", nullable: false), + name = table.Column(type: "text", nullable: false), + host_id = table.Column(type: "uuid", nullable: false), + host = table.Column(type: "text", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_known_endpoints_insert_only", x => x.id); + }); + + migrationBuilder.CreateIndex( + name: "IX_known_endpoints_insert_only_known_endpoint_id", + table: "known_endpoints_insert_only", + column: "known_endpoint_id"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "known_endpoints"); + + migrationBuilder.DropTable( + name: "known_endpoints_insert_only"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs new file mode 100644 index 0000000000..83b17187a4 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -0,0 +1,90 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; +using ServiceControl.Persistence.EFCore.PostgreSql; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + [DbContext(typeof(PostgreSqlServiceControlDbContext))] + partial class PostgreSqlServiceControlDbContextModelSnapshot : ModelSnapshot + { + protected override void BuildModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Host") + .IsRequired() + .HasColumnType("text") + .HasColumnName("host"); + + b.Property("HostId") + .HasColumnType("uuid") + .HasColumnName("host_id"); + + b.Property("Monitored") + .HasColumnType("boolean") + .HasColumnName("monitored"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text") + .HasColumnName("name"); + + b.HasKey("Id"); + + b.ToTable("known_endpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointInsertOnlyEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Host") + .IsRequired() + .HasColumnType("text") + .HasColumnName("host"); + + b.Property("HostId") + .HasColumnType("uuid") + .HasColumnName("host_id"); + + b.Property("KnownEndpointId") + .HasColumnType("uuid") + .HasColumnName("known_endpoint_id"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text") + .HasColumnName("name"); + + b.HasKey("Id"); + + b.HasIndex("KnownEndpointId"); + + b.ToTable("known_endpoints_insert_only", (string)null); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs index bba7a9506a..ea32ae9788 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs @@ -4,6 +4,7 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql; using Microsoft.Extensions.DependencyInjection; using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.PostgreSql.Infrastructure; class PostgreSqlPersistence(PostgreSqlPersisterSettings settings) : BasePersistence, IPersistence { @@ -12,6 +13,8 @@ public void AddPersistence(IServiceCollection services) RegisterSettings(services); ConfigureDbContext(services); RegisterDataStores(services); + + services.AddHostedService(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlServiceControlDbContext.cs index 3f7baa435e..c62a5c78e1 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlServiceControlDbContext.cs @@ -5,4 +5,11 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql; public class PostgreSqlServiceControlDbContext(DbContextOptions options) : ServiceControlDbContext(options) { + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + base.OnConfiguring(optionsBuilder); + + // Use snake_case naming convention for PostgreSQL + optionsBuilder.UseSnakeCaseNamingConvention(); + } } diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/ServiceControl.Persistence.EFCore.PostgreSql.csproj b/src/ServiceControl.Persistence.EFCore.PostgreSql/ServiceControl.Persistence.EFCore.PostgreSql.csproj index a20e270c74..52f0f4c5b5 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/ServiceControl.Persistence.EFCore.PostgreSql.csproj +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/ServiceControl.Persistence.EFCore.PostgreSql.csproj @@ -8,13 +8,21 @@ true - - + + + + + + + + @@ -22,6 +30,7 @@ + diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Infrastructure/KnownEndpointsReconciler.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Infrastructure/KnownEndpointsReconciler.cs new file mode 100644 index 0000000000..727ecb8e83 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Infrastructure/KnownEndpointsReconciler.cs @@ -0,0 +1,64 @@ +namespace ServiceControl.Persistence.EFCore.SqlServer.Infrastructure; + +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class KnownEndpointsReconciler( + ILogger logger, + TimeProvider timeProvider, + IServiceScopeFactory serviceScopeFactory) + : InsertOnlyTableReconciler( + logger, timeProvider, serviceScopeFactory, nameof(KnownEndpointsReconciler)) +{ + protected override Task ReconcileBatch(ServiceControlDbContext dbContext, CancellationToken stoppingToken) => + ReconcileBatch(dbContext, BatchSize, stoppingToken); + + // Static so tests can execute a batch deterministically without the background service's timer loop. + // Must be called within an active transaction because of the sp_getapplock Transaction lock owner. + internal static async Task ReconcileBatch(ServiceControlDbContext dbContext, int batchSize, CancellationToken cancellationToken) + { + var sql = """ + DECLARE @lockResult INT; + EXEC @lockResult = sp_getapplock @Resource = 'known_endpoints_sync', @LockMode = 'Exclusive', @LockOwner = 'Transaction', @LockTimeout = 0; + IF @lockResult < 0 + BEGIN + SELECT 0; + RETURN; + END; + + DECLARE @deleted TABLE ( + KnownEndpointId UNIQUEIDENTIFIER, + Name NVARCHAR(MAX), + HostId UNIQUEIDENTIFIER, + Host NVARCHAR(MAX) + ); + + DELETE TOP (@batchSize) FROM KnownEndpointsInsertOnly + OUTPUT DELETED.KnownEndpointId, DELETED.Name, DELETED.HostId, DELETED.Host + INTO @deleted; + + WITH ranked AS ( + SELECT KnownEndpointId, Name, HostId, Host, + ROW_NUMBER() OVER (PARTITION BY KnownEndpointId ORDER BY (SELECT NULL)) AS rn + FROM @deleted + ), + aggregated AS ( + SELECT KnownEndpointId, Name, HostId, Host + FROM ranked + WHERE rn = 1 + ) + MERGE INTO KnownEndpoints WITH (HOLDLOCK) AS target + USING aggregated AS source + ON target.Id = source.KnownEndpointId + WHEN NOT MATCHED THEN + INSERT (Id, Name, HostId, Host, Monitored) + VALUES (source.KnownEndpointId, source.Name, source.HostId, source.Host, 0); + """; + + var rowsAffected = await dbContext.Database.ExecuteSqlRawAsync(sql, [new Microsoft.Data.SqlClient.SqlParameter("@batchSize", batchSize)], cancellationToken); + return rowsAffected; + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260720230811_Initial.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260720230811_Initial.Designer.cs new file mode 100644 index 0000000000..5901a2770e --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260720230811_Initial.Designer.cs @@ -0,0 +1,83 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using ServiceControl.Persistence.EFCore.SqlServer; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + [DbContext(typeof(SqlServerServiceControlDbContext))] + [Migration("20260720230811_Initial")] + partial class Initial + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 128); + + SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Host") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("HostId") + .HasColumnType("uniqueidentifier"); + + b.Property("Monitored") + .HasColumnType("bit"); + + b.Property("Name") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("Id"); + + b.ToTable("KnownEndpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointInsertOnlyEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("Host") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("HostId") + .HasColumnType("uniqueidentifier"); + + b.Property("KnownEndpointId") + .HasColumnType("uniqueidentifier"); + + b.Property("Name") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("Id"); + + b.HasIndex("KnownEndpointId"); + + b.ToTable("KnownEndpointsInsertOnly", (string)null); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260720230811_Initial.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260720230811_Initial.cs new file mode 100644 index 0000000000..d1af2ee2e1 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260720230811_Initial.cs @@ -0,0 +1,61 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class Initial : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "KnownEndpoints", + columns: table => new + { + Id = table.Column(type: "uniqueidentifier", nullable: false), + Name = table.Column(type: "nvarchar(max)", nullable: false), + HostId = table.Column(type: "uniqueidentifier", nullable: false), + Host = table.Column(type: "nvarchar(max)", nullable: false), + Monitored = table.Column(type: "bit", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_KnownEndpoints", x => x.Id); + }); + + migrationBuilder.CreateTable( + name: "KnownEndpointsInsertOnly", + columns: table => new + { + Id = table.Column(type: "bigint", nullable: false) + .Annotation("SqlServer:Identity", "1, 1"), + KnownEndpointId = table.Column(type: "uniqueidentifier", nullable: false), + Name = table.Column(type: "nvarchar(max)", nullable: false), + HostId = table.Column(type: "uniqueidentifier", nullable: false), + Host = table.Column(type: "nvarchar(max)", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_KnownEndpointsInsertOnly", x => x.Id); + }); + + migrationBuilder.CreateIndex( + name: "IX_KnownEndpointsInsertOnly_KnownEndpointId", + table: "KnownEndpointsInsertOnly", + column: "KnownEndpointId"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "KnownEndpoints"); + + migrationBuilder.DropTable( + name: "KnownEndpointsInsertOnly"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs new file mode 100644 index 0000000000..df880e2f92 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -0,0 +1,80 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using ServiceControl.Persistence.EFCore.SqlServer; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + [DbContext(typeof(SqlServerServiceControlDbContext))] + partial class SqlServerServiceControlDbContextModelSnapshot : ModelSnapshot + { + protected override void BuildModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 128); + + SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Host") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("HostId") + .HasColumnType("uniqueidentifier"); + + b.Property("Monitored") + .HasColumnType("bit"); + + b.Property("Name") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("Id"); + + b.ToTable("KnownEndpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointInsertOnlyEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("Host") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("HostId") + .HasColumnType("uniqueidentifier"); + + b.Property("KnownEndpointId") + .HasColumnType("uniqueidentifier"); + + b.Property("Name") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("Id"); + + b.HasIndex("KnownEndpointId"); + + b.ToTable("KnownEndpointsInsertOnly", (string)null); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/ServiceControl.Persistence.EFCore.SqlServer.csproj b/src/ServiceControl.Persistence.EFCore.SqlServer/ServiceControl.Persistence.EFCore.SqlServer.csproj index 9346aa13f2..55d741d09e 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/ServiceControl.Persistence.EFCore.SqlServer.csproj +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/ServiceControl.Persistence.EFCore.SqlServer.csproj @@ -8,13 +8,21 @@ true - - + + + + + + + + diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs index 171f08a837..2882099881 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs @@ -4,6 +4,7 @@ namespace ServiceControl.Persistence.EFCore.SqlServer; using Microsoft.Extensions.DependencyInjection; using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.SqlServer.Infrastructure; class SqlServerPersistence(SqlServerPersisterSettings settings) : BasePersistence, IPersistence { @@ -12,6 +13,8 @@ public void AddPersistence(IServiceCollection services) RegisterSettings(services); ConfigureDbContext(services); RegisterDataStores(services); + + services.AddHostedService(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index dccd2abcbb..8a6642c313 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -6,6 +6,7 @@ namespace ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Operations.BodyStorage; using ServiceControl.Persistence.EFCore.Implementation; using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.MessageRedirects; using ServiceControl.Persistence.Recoverability; using ServiceControl.Persistence.UnitOfWork; @@ -47,5 +48,7 @@ protected static void RegisterDataStores(IServiceCollection services) services.AddSingleton(); services.AddSingleton(); + + services.AddSingleton(); } } diff --git a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs index 83f28e3310..6a7e0a6fc9 100644 --- a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs @@ -1,7 +1,24 @@ namespace ServiceControl.Persistence.EFCore.DbContexts; using Microsoft.EntityFrameworkCore; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.EntityConfigurations; public abstract class ServiceControlDbContext(DbContextOptions options) : DbContext(options) { + public DbSet KnownEndpoints { get; set; } + public DbSet KnownEndpointsInsertOnly { get; set; } + + protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) + { + optionsBuilder.EnableDetailedErrors(); + } + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + base.OnModelCreating(modelBuilder); + + modelBuilder.ApplyConfiguration(new KnownEndpointConfiguration()); + modelBuilder.ApplyConfiguration(new KnownEndpointInsertOnlyConfiguration()); + } } diff --git a/src/ServiceControl.Persistence.EFCore/Entities/KnownEndpointEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/KnownEndpointEntity.cs new file mode 100644 index 0000000000..5f2154b7e2 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/KnownEndpointEntity.cs @@ -0,0 +1,13 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public class KnownEndpointEntity +{ + public Guid Id { get; set; } + public required string Name { get; set; } + + public Guid HostId { get; set; } + + public required string Host { get; set; } + + public bool Monitored { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/Entities/KnownEndpointInsertOnlyEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/KnownEndpointInsertOnlyEntity.cs new file mode 100644 index 0000000000..00f4274407 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/KnownEndpointInsertOnlyEntity.cs @@ -0,0 +1,13 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public class KnownEndpointInsertOnlyEntity +{ + public long Id { get; set; } + public Guid KnownEndpointId { get; set; } + + public string? Name { get; set; } + + public Guid HostId { get; set; } + + public string? Host { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/KnownEndpointConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/KnownEndpointConfiguration.cs new file mode 100644 index 0000000000..f82cd1bbe5 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/KnownEndpointConfiguration.cs @@ -0,0 +1,19 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class KnownEndpointConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("KnownEndpoints"); + builder.HasKey(e => e.Id); + builder.Property(e => e.Id).ValueGeneratedNever(); + builder.Property(e => e.Name).IsRequired(); + builder.Property(e => e.HostId).IsRequired(); + builder.Property(e => e.Host).IsRequired(); + builder.Property(e => e.Monitored).IsRequired(); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/KnownEndpointInsertOnlyConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/KnownEndpointInsertOnlyConfiguration.cs new file mode 100644 index 0000000000..ba3ce9f578 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/KnownEndpointInsertOnlyConfiguration.cs @@ -0,0 +1,21 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class KnownEndpointInsertOnlyConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("KnownEndpointsInsertOnly"); + builder.HasKey(e => e.Id); + builder.Property(e => e.Id).ValueGeneratedOnAdd(); + builder.Property(e => e.KnownEndpointId).IsRequired(); + builder.Property(e => e.Name).IsRequired(); + builder.Property(e => e.HostId).IsRequired(); + builder.Property(e => e.Host).IsRequired(); + + builder.HasIndex(e => e.KnownEndpointId); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs b/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs new file mode 100644 index 0000000000..157450c54b --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs @@ -0,0 +1,39 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using System; +using System.Threading.Tasks; +using DbContexts; +using Microsoft.Extensions.DependencyInjection; + +/// +/// Base class for data stores that provides helper methods to simplify scope and DbContext management +/// +public abstract class DataStoreBase(IServiceScopeFactory scopeFactory) +{ + protected readonly IServiceScopeFactory scopeFactory = scopeFactory; + + /// + /// Executes an operation with a scoped DbContext, returning a result + /// + protected async Task ExecuteWithDbContext(Func> operation) + { + await using var scope = scopeFactory.CreateAsyncScope();// Use CreateAsyncScope for async disposal + var dbContext = scope.ServiceProvider.GetRequiredService(); + return await operation(dbContext); + } + + /// + /// Executes an operation with a scoped DbContext, without returning a result + /// + protected async Task ExecuteWithDbContext(Func operation) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + await operation(dbContext); + } + + /// + /// Creates a scope for operations that need to manage their own scope lifecycle (e.g., managers) + /// + protected IServiceScope CreateScope() => scopeFactory.CreateAsyncScope(); +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FakeBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FakeBodyStoragePersistence.cs new file mode 100644 index 0000000000..7aa2199020 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FakeBodyStoragePersistence.cs @@ -0,0 +1,10 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using ServiceControl.Persistence.EFCore.Infrastructure; + +public class FakeBodyStoragePersistence : IBodyStoragePersistence +{ + public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task ReadBody(string bodyId, DateTime createdOn, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task WriteBody(string bodyId, DateTime createdOn, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) => throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs index d6ce70598f..1212338c56 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs @@ -1,24 +1,155 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; using ServiceControl.Operations; +using ServiceControl.Persistence.EFCore.Entities; -public class MonitoringDataStore : IMonitoringDataStore +public class MonitoringDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IMonitoringDataStore { - public Task CreateIfNotExists(EndpointDetails endpoint) => - throw new NotImplementedException(); + public Task CreateIfNotExists(EndpointDetails endpoint) + { + return ExecuteWithDbContext(async dbContext => + { + var id = endpoint.GetDeterministicId(); - public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring) => - throw new NotImplementedException(); + var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id); + if (exists) + { + return; + } - public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored) => - throw new NotImplementedException(); + var knownEndpoint = new KnownEndpointEntity + { + Id = id, + Name = endpoint.Name, + HostId = endpoint.HostId, + Host = endpoint.Host, + Monitored = false + }; - public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring) => - throw new NotImplementedException(); + dbContext.KnownEndpoints.Add(knownEndpoint); - public Task Delete(Guid endpointId) => - throw new NotImplementedException(); + try + { + await dbContext.SaveChangesAsync(); + } + catch (DbUpdateException) + { + // A concurrent insert with the same deterministic id may have won the race. + // Anything else is a genuine failure. + dbContext.Entry(knownEndpoint).State = EntityState.Detached; + if (!await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id)) + { + throw; + } + } + }); + } - public Task> GetAllKnownEndpoints() => - throw new NotImplementedException(); + public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring) + { + return ExecuteWithDbContext(async dbContext => + { + var id = endpoint.GetDeterministicId(); + + var knownEndpoint = await dbContext.KnownEndpoints.FirstOrDefaultAsync(e => e.Id == id); + + if (knownEndpoint == null) + { + knownEndpoint = new KnownEndpointEntity + { + Id = id, + Name = endpoint.Name, + HostId = endpoint.HostId, + Host = endpoint.Host, + Monitored = true + }; + dbContext.KnownEndpoints.Add(knownEndpoint); + + try + { + await dbContext.SaveChangesAsync(); + } + catch (DbUpdateException) + { + // A concurrent insert with the same deterministic id may have won the race; + // fall back to the update path. Rethrow anything else. + dbContext.Entry(knownEndpoint).State = EntityState.Detached; + var monitored = endpointInstanceMonitoring.IsMonitored(id); + var updated = await dbContext.KnownEndpoints + .Where(e => e.Id == id) + .ExecuteUpdateAsync(setters => setters.SetProperty(e => e.Monitored, monitored)); + if (updated == 0) + { + throw; + } + } + } + else + { + knownEndpoint.Monitored = endpointInstanceMonitoring.IsMonitored(id); + await dbContext.SaveChangesAsync(); + } + }); + } + + public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored) + { + return ExecuteWithDbContext(async dbContext => + { + var id = endpoint.GetDeterministicId(); + + await dbContext.KnownEndpoints + .Where(e => e.Id == id) + .ExecuteUpdateAsync(setters => setters.SetProperty(e => e.Monitored, isMonitored)); + }); + } + + public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring) + { + return ExecuteWithDbContext(async dbContext => + { + await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable()) + { + var endpointDetails = new EndpointDetails + { + Name = endpoint.Name, + HostId = endpoint.HostId, + Host = endpoint.Host + }; + + endpointInstanceMonitoring.DetectEndpointFromPersistentStore(endpointDetails, endpoint.Monitored); + } + }); + } + + public Task Delete(Guid endpointId) + { + return ExecuteWithDbContext(async dbContext => + { + await dbContext.KnownEndpoints + .Where(e => e.Id == endpointId) + .ExecuteDeleteAsync(); + }); + } + + public Task> GetAllKnownEndpoints() + { + return ExecuteWithDbContext>(async dbContext => + await dbContext.KnownEndpoints + .AsNoTracking() + .Select(e => new KnownEndpoint + { + EndpointDetails = new EndpointDetails + { + Name = e.Name, + HostId = e.HostId, + Host = e.Host + }, + HostDisplayName = e.Host, + Monitored = e.Monitored + }) + .ToListAsync()); + } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs index 33d7ae477f..3ead245bb1 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs @@ -1,6 +1,5 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; -using Microsoft.Extensions.DependencyInjection; using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.DbContexts; using ServiceControl.Persistence.EFCore.Infrastructure; @@ -9,19 +8,19 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; public class EFIngestionUnitOfWork : IIngestionUnitOfWork { readonly ServiceControlDbContext dbContext; - readonly AsyncServiceScope scope; + readonly IAsyncDisposable scope; - public EFIngestionUnitOfWork(AsyncServiceScope scope, ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings) + public EFIngestionUnitOfWork(IAsyncDisposable scope, ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings) { this.scope = scope; this.dbContext = dbContext; + Recoverability = new EFRecoverabilityIngestionUnitOfWork(dbContext, storagePersistence, settings); + Monitoring = new EFMonitoringIngestionUnitOfWork(dbContext); } - public IMonitoringIngestionUnitOfWork Monitoring => - throw new NotImplementedException(); + public IMonitoringIngestionUnitOfWork Monitoring { get; } - public IRecoverabilityIngestionUnitOfWork Recoverability => - throw new NotImplementedException(); + public IRecoverabilityIngestionUnitOfWork Recoverability { get; } public Task Complete(CancellationToken cancellationToken) => dbContext.SaveChangesAsync(cancellationToken); @@ -33,3 +32,4 @@ public async ValueTask DisposeAsync() GC.SuppressFinalize(this); } } + diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFMonitoringIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFMonitoringIngestionUnitOfWork.cs index af7080b6a5..b37f07d634 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFMonitoringIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFMonitoringIngestionUnitOfWork.cs @@ -1,9 +1,25 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; using ServiceControl.Persistence.UnitOfWork; -public class EFMonitoringIngestionUnitOfWork : IMonitoringIngestionUnitOfWork +public class EFMonitoringIngestionUnitOfWork(ServiceControlDbContext dbContext) : IMonitoringIngestionUnitOfWork { - public Task RecordKnownEndpoint(KnownEndpoint knownEndpoint) => - throw new NotImplementedException(); + readonly ServiceControlDbContext dbContext = dbContext; + + public Task RecordKnownEndpoint(KnownEndpoint knownEndpoint) + { + var entity = new KnownEndpointInsertOnlyEntity + { + KnownEndpointId = knownEndpoint.EndpointDetails.GetDeterministicId(), + Name = knownEndpoint.EndpointDetails.Name, + HostId = knownEndpoint.EndpointDetails.HostId, + Host = knownEndpoint.EndpointDetails.Host + }; + + dbContext.KnownEndpointsInsertOnly.Add(entity); + + return Task.CompletedTask; + } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs index 7e7e93f63e..5eb0989b71 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs @@ -3,8 +3,16 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; using NServiceBus.Transport; using ServiceControl.MessageFailures; using ServiceControl.Persistence.UnitOfWork; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using ServiceControl.Persistence.EFCore.Abstractions; -public class EFRecoverabilityIngestionUnitOfWork : IRecoverabilityIngestionUnitOfWork +#pragma warning disable CS9113 // Parameter is unread. +public class EFRecoverabilityIngestionUnitOfWork(ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings) : IRecoverabilityIngestionUnitOfWork +#pragma warning restore CS9113 // Parameter is unread. { public Task RecordFailedProcessingAttempt(MessageContext context, FailedMessage.ProcessingAttempt processingAttempt, diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs new file mode 100644 index 0000000000..643b146219 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs @@ -0,0 +1,80 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using ServiceControl.Persistence.EFCore.DbContexts; + +public abstract class InsertOnlyTableReconciler( + ILogger logger, + TimeProvider timeProvider, + IServiceScopeFactory serviceScopeFactory, + string serviceName) : BackgroundService +{ + protected const int BatchSize = 1000; + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + logger.LogInformation("Starting {ServiceName}", serviceName); + + try + { + await Task.Delay(TimeSpan.FromSeconds(30), timeProvider, stoppingToken); + + using PeriodicTimer timer = new(TimeSpan.FromSeconds(30), timeProvider); + + do + { + try + { + await Reconcile(pace: true, stoppingToken); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "Error during {ServiceName} reconciliation", serviceName); + } + } while (await timer.WaitForNextTickAsync(stoppingToken)); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + logger.LogInformation("Stopping {ServiceName}", serviceName); + } + } + + // Drains all pending rows immediately, bypassing the background timer and the inter-batch pacing. + // Intended for tests that need reconciled data visible without waiting for the timer loop. + public Task ReconcileNow(CancellationToken cancellationToken = default) => Reconcile(pace: false, cancellationToken); + + async Task Reconcile(bool pace, CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + using var scope = serviceScopeFactory.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + // A retrying execution strategy (EnableRetryOnFailure) rejects user-initiated transactions + // unless the whole unit is wrapped so it can be retried as one. + var strategy = dbContext.Database.CreateExecutionStrategy(); + var rowsAffected = await strategy.ExecuteAsync(dbContext, async (context, ct) => + { + await using var transaction = await context.Database.BeginTransactionAsync(ct); + var rows = await ReconcileBatch(context, ct); + await transaction.CommitAsync(ct); + return rows; + }, cancellationToken); + + if (rowsAffected < BatchSize) + { + break; + } + + if (pace) + { + await Task.Delay(TimeSpan.FromSeconds(2), timeProvider, cancellationToken); + } + } + } + + protected abstract Task ReconcileBatch(ServiceControlDbContext dbContext, CancellationToken stoppingToken); +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs index 2d58946d9b..531eba1f58 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs @@ -4,8 +4,8 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; public class MessageBodyFileResult { - public Stream Stream { get; set; } = null!; - public string ContentType { get; set; } = null!; - public int BodySize { get; set; } - public string Etag { get; set; } = null!; + public required Stream Stream { get; set; } + public required string ContentType { get; set; } + public required int BodySize { get; set; } + public required string Etag { get; set; } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests.InMemory/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.InMemory/PersistenceTestsContext.cs index 8e66c39775..2b17a1fa7e 100644 --- a/src/ServiceControl.Persistence.Tests.InMemory/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.InMemory/PersistenceTestsContext.cs @@ -23,7 +23,5 @@ public Task Setup(IHostApplicationBuilder hostBuilder) public Task TearDown() => Task.CompletedTask; - public void CompleteDatabaseOperation() - { - } + public Task CompleteDatabaseOperation() => Task.CompletedTask; } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/KnownEndpointsReconcilerTests.cs b/src/ServiceControl.Persistence.Tests.PostgreSql/KnownEndpointsReconcilerTests.cs new file mode 100644 index 0000000000..a1529d19b1 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/KnownEndpointsReconcilerTests.cs @@ -0,0 +1,156 @@ +namespace ServiceControl.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.EntityFrameworkCore; + using Microsoft.Extensions.DependencyInjection; + using NUnit.Framework; + using ServiceControl.Persistence.EFCore.DbContexts; + using ServiceControl.Persistence.EFCore.Entities; + using ServiceControl.Persistence.EFCore.PostgreSql.Infrastructure; + + [TestFixture] + class KnownEndpointsReconcilerTests : PersistenceTestBase + { + [Test] + public async Task Moves_pending_rows_into_known_endpoints() + { + var row1 = NewInsertOnlyRow("Endpoint1"); + var row2 = NewInsertOnlyRow("Endpoint2"); + await SeedInsertOnlyRows(row1, row2); + + await RunReconcileBatch(); + + var ids = new[] { row1.KnownEndpointId, row2.KnownEndpointId }; + var knownEndpoints = await GetKnownEndpoints(ids); + Assert.That(knownEndpoints, Has.Count.EqualTo(2)); + + var endpoint1 = knownEndpoints.Single(e => e.Id == row1.KnownEndpointId); + using (Assert.EnterMultipleScope()) + { + Assert.That(endpoint1.Name, Is.EqualTo(row1.Name)); + Assert.That(endpoint1.HostId, Is.EqualTo(row1.HostId)); + Assert.That(endpoint1.Host, Is.EqualTo(row1.Host)); + Assert.That(endpoint1.Monitored, Is.False, "Endpoints detected during ingestion should not be monitored by default"); + } + + Assert.That(await CountInsertOnlyRows(ids), Is.Zero, "Reconciled rows should be removed from the insert-only table"); + } + + [Test] + public async Task Duplicate_pending_rows_produce_single_endpoint() + { + var row = NewInsertOnlyRow("Endpoint1"); + var duplicate = NewInsertOnlyRow("Endpoint1"); + duplicate.KnownEndpointId = row.KnownEndpointId; + await SeedInsertOnlyRows(row, duplicate); + + await RunReconcileBatch(); + + var ids = new[] { row.KnownEndpointId }; + using (Assert.EnterMultipleScope()) + { + Assert.That(await GetKnownEndpoints(ids), Has.Count.EqualTo(1)); + Assert.That(await CountInsertOnlyRows(ids), Is.Zero); + } + } + + [Test] + public async Task Existing_endpoints_are_not_modified() + { + var row = NewInsertOnlyRow("Endpoint1"); + await SeedKnownEndpoint(new KnownEndpointEntity + { + Id = row.KnownEndpointId, + Name = row.Name, + HostId = row.HostId, + Host = row.Host, + Monitored = true + }); + await SeedInsertOnlyRows(row); + + await RunReconcileBatch(); + + var ids = new[] { row.KnownEndpointId }; + var knownEndpoints = await GetKnownEndpoints(ids); + using (Assert.EnterMultipleScope()) + { + Assert.That(knownEndpoints, Has.Count.EqualTo(1)); + Assert.That(knownEndpoints[0].Monitored, Is.True, "Reconciliation should not overwrite existing endpoints"); + Assert.That(await CountInsertOnlyRows(ids), Is.Zero); + } + } + + [Test] + public async Task Reconciles_in_batches() + { + var rows = new[] { NewInsertOnlyRow("Endpoint1"), NewInsertOnlyRow("Endpoint2"), NewInsertOnlyRow("Endpoint3") }; + await SeedInsertOnlyRows(rows); + var ids = rows.Select(r => r.KnownEndpointId).ToArray(); + + await RunReconcileBatch(batchSize: 2); + Assert.That(await CountInsertOnlyRows(ids), Is.EqualTo(1), "Only one batch should be processed per call"); + + await RunReconcileBatch(batchSize: 2); + using (Assert.EnterMultipleScope()) + { + Assert.That(await CountInsertOnlyRows(ids), Is.Zero); + Assert.That(await GetKnownEndpoints(ids), Has.Count.EqualTo(3)); + } + } + + static KnownEndpointInsertOnlyEntity NewInsertOnlyRow(string name) => new() + { + KnownEndpointId = Guid.NewGuid(), + Name = name, + HostId = Guid.NewGuid(), + Host = "Host1" + }; + + async Task SeedInsertOnlyRows(params KnownEndpointInsertOnlyEntity[] rows) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + dbContext.KnownEndpointsInsertOnly.AddRange(rows); + await dbContext.SaveChangesAsync(); + } + + async Task SeedKnownEndpoint(KnownEndpointEntity entity) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + dbContext.KnownEndpoints.Add(entity); + await dbContext.SaveChangesAsync(); + } + + async Task RunReconcileBatch(int batchSize = 1000) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + var strategy = dbContext.Database.CreateExecutionStrategy(); + await strategy.ExecuteAsync(async () => + { + await using var transaction = await dbContext.Database.BeginTransactionAsync(); + await KnownEndpointsReconciler.ReconcileBatch(dbContext, batchSize, CancellationToken.None); + await transaction.CommitAsync(); + }); + } + + async Task> GetKnownEndpoints(IReadOnlyCollection ids) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + return await dbContext.KnownEndpoints.AsNoTracking().Where(e => ids.Contains(e.Id)).ToListAsync(); + } + + async Task CountInsertOnlyRows(IReadOnlyCollection ids) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + return await dbContext.KnownEndpointsInsertOnly.CountAsync(e => ids.Contains(e.KnownEndpointId)); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs index 091e910ece..a104f188eb 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs @@ -1,36 +1,73 @@ // ReSharper disable once CheckNamespace namespace ServiceControl.Persistence.Tests; +using System; +using System.Linq; using System.Threading.Tasks; using EFCore.PostgreSql; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Time.Testing; +using Npgsql; +using ServiceControl.Persistence.EFCore.Infrastructure; public class PersistenceTestsContext : IPersistenceTestsContext { + IHost host; + string databaseName; + + public FakeTimeProvider FakeTime { get; } = new(); + public async Task Setup(IHostApplicationBuilder hostBuilder) { + databaseName = $"sc_test_{Guid.NewGuid():n}"; + + var connectionStringBuilder = new NpgsqlConnectionStringBuilder(await PostgreSqlSharedContainer.GetConnectionStringAsync()) + { + Database = databaseName + }; + PersistenceSettings = new PostgreSqlPersisterSettings { - ConnectionString = await PostgreSqlSharedContainer.GetConnectionStringAsync() + ConnectionString = connectionStringBuilder.ConnectionString }; var persistence = new PostgreSqlPersistenceConfiguration().Create(PersistenceSettings); persistence.AddPersistence(hostBuilder.Services); persistence.AddInstaller(hostBuilder.Services); + + hostBuilder.Services.AddSingleton(FakeTime); } public async Task PostSetup(IHost host) { + this.host = host; + using var scope = host.Services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); - await db.Database.EnsureCreatedAsync(); + await db.Database.MigrateAsync(); } - public Task TearDown() => Task.CompletedTask; + public async Task TearDown() + { + await using var connection = new NpgsqlConnection(await PostgreSqlSharedContainer.GetConnectionStringAsync()); + await connection.OpenAsync(); + await using var command = connection.CreateCommand(); + command.CommandText = $"DROP DATABASE IF EXISTS \"{databaseName}\" WITH (FORCE)"; + await command.ExecuteNonQueryAsync(); + } - public void CompleteDatabaseOperation() { } + // Drain every insert-only reconciler so that ingested data is visible to the data stores, + // without waiting for the reconciler background services' timers. + public async Task CompleteDatabaseOperation() + { + foreach (var reconciler in host.Services.GetServices().OfType()) + { + await reconciler.ReconcileNow(); + } + } public PersistenceSettings PersistenceSettings { get; set; } diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj index 82155803ad..c2d8b611a2 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj @@ -14,6 +14,7 @@ + diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs index 67b12d36b5..1b89d399b1 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs @@ -41,7 +41,7 @@ public async Task SingleMessageMarkedAsArchiveShouldExpire() await uow.Complete(TestContext.CurrentContext.CancellationToken); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var error = await GetAllMessages(); @@ -71,7 +71,7 @@ public async Task AllMessagesInUnArchivedGroupShouldNotExpire() await uow.Complete(TestContext.CurrentContext.CancellationToken); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var error = await GetAllMessages(); @@ -85,7 +85,7 @@ public async Task AllMessagesInUnArchivedGroupShouldNotExpire() // Let ArchivedGroupsViewIndex catch up with the archive, or the unarchive silently // no-ops ("No messages to unarchive") and message B wrongly expires. - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await ArchiveMessages.UnarchiveAllInGroup(groupIdB); @@ -107,7 +107,7 @@ public async Task AllMessagesInArchivedGroupShouldExpire() await uow.Complete(TestContext.CurrentContext.CancellationToken); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var error = await GetAllMessages(); @@ -130,7 +130,7 @@ public async Task SingleMessageMarkedAsResolvedShouldExpire() await uow.Complete(TestContext.CurrentContext.CancellationToken); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var errors = await GetAllMessages(); @@ -153,7 +153,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() await uow.Complete(TestContext.CurrentContext.CancellationToken); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var errors = await GetAllMessages(); @@ -191,7 +191,7 @@ public async Task EventLogItemShouldExpire() await ErrorStore.StoreEventLogItem(new EventLogItem()); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var (logItems, _, _) = await EventLogDataStore.GetEventLogItems(new PagingInfo(1, 1)); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.RavenDB/PersistenceTestsContext.cs index e8ba73939c..63e455105c 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/PersistenceTestsContext.cs @@ -47,7 +47,7 @@ public async Task PostSetup(IHost host) DocumentStore = await host.Services.GetRequiredService().GetDocumentStore(); SessionProvider = host.Services.GetRequiredService(); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); } public async Task TearDown() => await embeddedServer.DeleteDatabase(databaseName); @@ -59,7 +59,11 @@ public async Task PostSetup(IHost host) public IRavenSessionProvider SessionProvider { get; private set; } - public void CompleteDatabaseOperation() => DocumentStore.WaitForIndexing(); + public Task CompleteDatabaseOperation() + { + DocumentStore.WaitForIndexing(); + return Task.CompletedTask; + } [Conditional("DEBUG")] public void BlockToInspectDatabase() diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs index 5ef3d88d91..95bf8d2cc0 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs @@ -59,7 +59,7 @@ public async Task GetStore() await Task.Yield(); await GenerateAndSaveFailedMessage(); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); store = ServiceProvider.GetRequiredService(); } diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/KnownEndpointsReconcilerTests.cs b/src/ServiceControl.Persistence.Tests.SqlServer/KnownEndpointsReconcilerTests.cs new file mode 100644 index 0000000000..c9aa851b9d --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.SqlServer/KnownEndpointsReconcilerTests.cs @@ -0,0 +1,158 @@ +namespace ServiceControl.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.EntityFrameworkCore; + using Microsoft.Extensions.DependencyInjection; + using NUnit.Framework; + using ServiceControl.Persistence.EFCore.DbContexts; + using ServiceControl.Persistence.EFCore.Entities; + using ServiceControl.Persistence.EFCore.SqlServer.Infrastructure; + + // The SqlServer/PostgreSql persistence test suites share one database/container across all + // tests (no per-test isolation), so assertions here are always scoped to the specific + // KnownEndpointId values each test seeds, rather than the table's total row count. + class KnownEndpointsReconcilerTests : PersistenceTestBase + { + [Test] + public async Task Moves_pending_rows_into_known_endpoints() + { + var row1 = NewInsertOnlyRow("Endpoint1"); + var row2 = NewInsertOnlyRow("Endpoint2"); + await SeedInsertOnlyRows(row1, row2); + + await RunReconcileBatch(); + + var ids = new[] { row1.KnownEndpointId, row2.KnownEndpointId }; + var knownEndpoints = await GetKnownEndpoints(ids); + Assert.That(knownEndpoints, Has.Count.EqualTo(2)); + + var endpoint1 = knownEndpoints.Single(e => e.Id == row1.KnownEndpointId); + using (Assert.EnterMultipleScope()) + { + Assert.That(endpoint1.Name, Is.EqualTo(row1.Name)); + Assert.That(endpoint1.HostId, Is.EqualTo(row1.HostId)); + Assert.That(endpoint1.Host, Is.EqualTo(row1.Host)); + Assert.That(endpoint1.Monitored, Is.False, "Endpoints detected during ingestion should not be monitored by default"); + } + + Assert.That(await CountInsertOnlyRows(ids), Is.Zero, "Reconciled rows should be removed from the insert-only table"); + } + + [Test] + public async Task Duplicate_pending_rows_produce_single_endpoint() + { + var row = NewInsertOnlyRow("Endpoint1"); + var duplicate = NewInsertOnlyRow("Endpoint1"); + duplicate.KnownEndpointId = row.KnownEndpointId; + await SeedInsertOnlyRows(row, duplicate); + + await RunReconcileBatch(); + + var ids = new[] { row.KnownEndpointId }; + using (Assert.EnterMultipleScope()) + { + Assert.That(await GetKnownEndpoints(ids), Has.Count.EqualTo(1)); + Assert.That(await CountInsertOnlyRows(ids), Is.Zero); + } + } + + [Test] + public async Task Existing_endpoints_are_not_modified() + { + var row = NewInsertOnlyRow("Endpoint1"); + await SeedKnownEndpoint(new KnownEndpointEntity + { + Id = row.KnownEndpointId, + Name = row.Name, + HostId = row.HostId, + Host = row.Host, + Monitored = true + }); + await SeedInsertOnlyRows(row); + + await RunReconcileBatch(); + + var ids = new[] { row.KnownEndpointId }; + var knownEndpoints = await GetKnownEndpoints(ids); + using (Assert.EnterMultipleScope()) + { + Assert.That(knownEndpoints, Has.Count.EqualTo(1)); + Assert.That(knownEndpoints[0].Monitored, Is.True, "Reconciliation should not overwrite existing endpoints"); + Assert.That(await CountInsertOnlyRows(ids), Is.Zero); + } + } + + [Test] + public async Task Reconciles_in_batches() + { + var rows = new[] { NewInsertOnlyRow("Endpoint1"), NewInsertOnlyRow("Endpoint2"), NewInsertOnlyRow("Endpoint3") }; + await SeedInsertOnlyRows(rows); + var ids = rows.Select(r => r.KnownEndpointId).ToArray(); + + await RunReconcileBatch(batchSize: 2); + Assert.That(await CountInsertOnlyRows(ids), Is.EqualTo(1), "Only one batch should be processed per call"); + + await RunReconcileBatch(batchSize: 2); + using (Assert.EnterMultipleScope()) + { + Assert.That(await CountInsertOnlyRows(ids), Is.Zero); + Assert.That(await GetKnownEndpoints(ids), Has.Count.EqualTo(3)); + } + } + + static KnownEndpointInsertOnlyEntity NewInsertOnlyRow(string name) => new() + { + KnownEndpointId = Guid.NewGuid(), + Name = name, + HostId = Guid.NewGuid(), + Host = "Host1" + }; + + async Task SeedInsertOnlyRows(params KnownEndpointInsertOnlyEntity[] rows) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + dbContext.KnownEndpointsInsertOnly.AddRange(rows); + await dbContext.SaveChangesAsync(); + } + + async Task SeedKnownEndpoint(KnownEndpointEntity entity) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + dbContext.KnownEndpoints.Add(entity); + await dbContext.SaveChangesAsync(); + } + + async Task RunReconcileBatch(int batchSize = 1000) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + var strategy = dbContext.Database.CreateExecutionStrategy(); + await strategy.ExecuteAsync(async () => + { + await using var transaction = await dbContext.Database.BeginTransactionAsync(); + await KnownEndpointsReconciler.ReconcileBatch(dbContext, batchSize, CancellationToken.None); + await transaction.CommitAsync(); + }); + } + + async Task> GetKnownEndpoints(IReadOnlyCollection ids) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + return await dbContext.KnownEndpoints.AsNoTracking().Where(e => ids.Contains(e.Id)).ToListAsync(); + } + + async Task CountInsertOnlyRows(IReadOnlyCollection ids) + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + return await dbContext.KnownEndpointsInsertOnly.CountAsync(e => ids.Contains(e.KnownEndpointId)); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs index 2d3a6b49d3..9af5dc6de0 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs @@ -1,36 +1,79 @@ // ReSharper disable once CheckNamespace namespace ServiceControl.Persistence.Tests; +using System; +using System.Linq; using System.Threading.Tasks; using EFCore.SqlServer; +using Microsoft.Data.SqlClient; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Time.Testing; +using ServiceControl.Persistence.EFCore.Infrastructure; public class PersistenceTestsContext : IPersistenceTestsContext { + IHost host; + string databaseName; + + public FakeTimeProvider FakeTime { get; } = new(); + public async Task Setup(IHostApplicationBuilder hostBuilder) { + databaseName = $"sc_test_{Guid.NewGuid():n}"; + + var connectionStringBuilder = new SqlConnectionStringBuilder(await SqlServerSharedContainer.GetConnectionStringAsync()) + { + InitialCatalog = databaseName + }; + PersistenceSettings = new SqlServerPersisterSettings { - ConnectionString = await SqlServerSharedContainer.GetConnectionStringAsync() + ConnectionString = connectionStringBuilder.ConnectionString }; var persistence = new SqlServerPersistenceConfiguration().Create(PersistenceSettings); persistence.AddPersistence(hostBuilder.Services); persistence.AddInstaller(hostBuilder.Services); + + hostBuilder.Services.AddSingleton(FakeTime); } public async Task PostSetup(IHost host) { + this.host = host; + using var scope = host.Services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); - await db.Database.EnsureCreatedAsync(); + await db.Database.MigrateAsync(); } - public Task TearDown() => Task.CompletedTask; + public async Task TearDown() + { + await using var connection = new SqlConnection(await SqlServerSharedContainer.GetConnectionStringAsync()); + await connection.OpenAsync(); + await using var command = connection.CreateCommand(); + 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(); + } - public void CompleteDatabaseOperation() { } + // Drain every insert-only reconciler so that ingested data is visible to the data stores, + // without waiting for the reconciler background services' timers. + public async Task CompleteDatabaseOperation() + { + foreach (var reconciler in host.Services.GetServices().OfType()) + { + await reconciler.ReconcileNow(); + } + } public PersistenceSettings PersistenceSettings { get; set; } diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj index 8bb8ad37bb..bfd404dd6b 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj +++ b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj @@ -14,6 +14,7 @@ + diff --git a/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs index de9d11235e..bc365e74f2 100644 --- a/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs +++ b/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs @@ -73,7 +73,7 @@ async Task RunTest(Func, string> getIdToQuery) await uow.Complete(cancellationSource.Token); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var fetchById = getIdToQuery(headers); diff --git a/src/ServiceControl.Persistence.Tests/CustomChecksDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/CustomChecksDataStoreTests.cs index d8d5af9903..04818e0417 100644 --- a/src/ServiceControl.Persistence.Tests/CustomChecksDataStoreTests.cs +++ b/src/ServiceControl.Persistence.Tests/CustomChecksDataStoreTests.cs @@ -30,7 +30,7 @@ public async Task CustomChecks_load_from_data_store() var status = await CustomChecks.UpdateCustomCheckStatus(checkDetails); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var stats = await CustomChecks.GetStats(new PagingInfo()); Assert.That(stats.Results, Has.Count.EqualTo(1)); @@ -61,7 +61,7 @@ public async Task Storing_failed_custom_checks_returns_unchanged() var statusInitial = await CustomChecks.UpdateCustomCheckStatus(checkDetails); var statusUpdate = await CustomChecks.UpdateCustomCheckStatus(checkDetails); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); using (Assert.EnterMultipleScope()) { @@ -89,7 +89,7 @@ public async Task Retrieving_custom_checks_by_status() var _ = await CustomChecks.UpdateCustomCheckStatus(checkDetails); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var stats = await CustomChecks.GetStats(new PagingInfo(), "pass"); @@ -116,11 +116,11 @@ public async Task Should_delete_custom_checks() var _ = await CustomChecks.UpdateCustomCheckStatus(checkDetails); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await CustomChecks.DeleteCustomCheck(checkId); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var storedChecks = await CustomChecks.GetStats(new PagingInfo()); var check = storedChecks.Results.Where(c => c.Id == checkId.ToString()).ToList(); diff --git a/src/ServiceControl.Persistence.Tests/IPersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests/IPersistenceTestsContext.cs index 5009a117a8..bf59a57835 100644 --- a/src/ServiceControl.Persistence.Tests/IPersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests/IPersistenceTestsContext.cs @@ -11,7 +11,7 @@ public interface IPersistenceTestsContext Task TearDown(); - void CompleteDatabaseOperation(); + Task CompleteDatabaseOperation(); PersistenceSettings PersistenceSettings { get; } diff --git a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs index 7e8a7d3fce..51e796fd83 100644 --- a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs +++ b/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs @@ -18,7 +18,7 @@ public async Task Endpoints_load_from_dataStore_into_monitor() var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; await MonitoringDataStore.CreateIfNotExists(endpoint1); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host), Is.EqualTo(1)); @@ -32,7 +32,7 @@ public async Task Endpoints_added_more_than_once_are_treated_as_same_endpoint() await MonitoringDataStore.CreateIfNotExists(endpoint1); await MonitoringDataStore.CreateIfNotExists(endpoint1); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host), Is.EqualTo(1)); @@ -46,7 +46,7 @@ public async Task Updating_existing_endpoint_does_not_create_new_ones() await MonitoringDataStore.CreateIfNotExists(endpoint1); await MonitoringDataStore.CreateOrUpdate(endpoint1, endpointInstanceMonitoring); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host), Is.EqualTo(1)); @@ -61,7 +61,7 @@ public async Task Endpoint_is_created_if_doesnt_exist() await MonitoringDataStore.CreateIfNotExists(endpoint1); await MonitoringDataStore.CreateIfNotExists(endpoint2); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host || w.HostDisplayName == endpoint2.Host), Is.EqualTo(2)); @@ -76,7 +76,7 @@ public async Task Endpoint_is_created_if_doesnt_exist_on_update() await MonitoringDataStore.CreateIfNotExists(endpoint1); await MonitoringDataStore.CreateOrUpdate(endpoint2, endpointInstanceMonitoring); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host || w.HostDisplayName == endpoint2.Host), Is.EqualTo(2)); @@ -89,14 +89,14 @@ public async Task Endpoint_is_updated() var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; await MonitoringDataStore.CreateIfNotExists(endpoint1); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.IsMonitored(endpointInstanceMonitoring.GetEndpoints()[0].Id), Is.False); await MonitoringDataStore.UpdateEndpointMonitoring(endpoint1, true); endpointInstanceMonitoring = new EndpointInstanceMonitoring(new FakeDomainEvents(), NullLogger.Instance); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.IsMonitored(endpointInstanceMonitoring.GetEndpoints()[0].Id), Is.True); } @@ -109,7 +109,7 @@ public async Task Endpoint_is_deleted() var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; await MonitoringDataStore.CreateIfNotExists(endpoint1); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host), Is.EqualTo(1)); @@ -117,11 +117,85 @@ public async Task Endpoint_is_deleted() endpointInstanceMonitoring = new EndpointInstanceMonitoring(new FakeDomainEvents(), NullLogger.Instance); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await MonitoringDataStore.WarmupMonitoringFromPersistence(endpointInstanceMonitoring); Assert.That(endpointInstanceMonitoring.GetKnownEndpoints().Count(w => w.HostDisplayName == endpoint1.Host), Is.EqualTo(0)); } + [Test] + public async Task GetAllKnownEndpoints_returns_created_endpoints() + { + var endpointInstanceMonitoring = new EndpointInstanceMonitoring(new FakeDomainEvents(), NullLogger.Instance); + var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; + var endpoint2 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host2", Name = "Name2" }; + await MonitoringDataStore.CreateIfNotExists(endpoint1); + await MonitoringDataStore.CreateOrUpdate(endpoint2, endpointInstanceMonitoring); + + await CompleteDatabaseOperation(); + var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints(); + + var known1 = knownEndpoints.Single(e => e.EndpointDetails.HostId == endpoint1.HostId); + var known2 = knownEndpoints.Single(e => e.EndpointDetails.HostId == endpoint2.HostId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(known1.EndpointDetails.Name, Is.EqualTo(endpoint1.Name)); + Assert.That(known1.EndpointDetails.Host, Is.EqualTo(endpoint1.Host)); + Assert.That(known1.HostDisplayName, Is.EqualTo(endpoint1.Host)); + Assert.That(known1.Monitored, Is.False, "Endpoints created via CreateIfNotExists should not be monitored"); + Assert.That(known2.Monitored, Is.True, "Endpoints created via CreateOrUpdate should be monitored"); + } + } + + [Test] + public async Task Delete_removes_only_target_endpoint() + { + var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; + var endpoint2 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host2", Name = "Name2" }; + await MonitoringDataStore.CreateIfNotExists(endpoint1); + await MonitoringDataStore.CreateIfNotExists(endpoint2); + + await MonitoringDataStore.Delete(endpoint1.GetDeterministicId()); + + await CompleteDatabaseOperation(); + var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(knownEndpoints.Any(e => e.EndpointDetails.HostId == endpoint1.HostId), Is.False); + Assert.That(knownEndpoints.Any(e => e.EndpointDetails.HostId == endpoint2.HostId), Is.True); + } + } + + [Test] + public async Task Concurrent_creates_result_in_single_endpoint() + { + var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; + + await Task.WhenAll(Enumerable.Range(0, 20) + .Select(_ => Task.Run(() => MonitoringDataStore.CreateIfNotExists(endpoint1)))); + + await CompleteDatabaseOperation(); + var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints(); + + Assert.That(knownEndpoints.Count(e => e.EndpointDetails.HostId == endpoint1.HostId), Is.EqualTo(1)); + } + + [Test] + public async Task Concurrent_create_or_updates_result_in_single_endpoint() + { + var endpointInstanceMonitoring = new EndpointInstanceMonitoring(new FakeDomainEvents(), NullLogger.Instance); + var endpoint1 = new EndpointDetails() { HostId = Guid.NewGuid(), Host = "Host1", Name = "Name1" }; + + await Task.WhenAll(Enumerable.Range(0, 20) + .Select(_ => Task.Run(() => MonitoringDataStore.CreateOrUpdate(endpoint1, endpointInstanceMonitoring)))); + + await CompleteDatabaseOperation(); + var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints(); + + Assert.That(knownEndpoints.Count(e => e.EndpointDetails.HostId == endpoint1.HostId), Is.EqualTo(1)); + } + [Test] public async Task Unit_of_work_detects_endpoint() { @@ -138,7 +212,7 @@ public async Task Unit_of_work_detects_endpoint() await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); } - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints(); diff --git a/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs b/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs index b0fd60e9ac..e4175e2c4a 100644 --- a/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs +++ b/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs @@ -72,7 +72,7 @@ public async Task TearDown() protected Action RegisterServices { get; set; } = _ => { }; - protected void CompleteDatabaseOperation() => PersistenceTestsContext.CompleteDatabaseOperation(); + protected Task CompleteDatabaseOperation() => PersistenceTestsContext.CompleteDatabaseOperation(); protected static async Task WaitUntil(Func> conditionChecker, string condition, TimeSpan timeout = default) { diff --git a/src/ServiceControl.Persistence.Tests/RetryStateTests.cs b/src/ServiceControl.Persistence.Tests/RetryStateTests.cs index bc87e50e34..406388eda7 100644 --- a/src/ServiceControl.Persistence.Tests/RetryStateTests.cs +++ b/src/ServiceControl.Persistence.Tests/RetryStateTests.cs @@ -50,7 +50,7 @@ public async Task When_a_group_is_prepared_and_SC_is_started_the_group_is_marked var orphanage = new AdoptOrphanBatchesFromPreviousSessionHostedService(documentManager, new AsyncTimer(), NullLogger.Instance); await orphanage.AdoptOrphanedBatchesAsync(); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var status = retryManager.GetStatusForRetryOperation("Test-group", RetryType.FailureGroup); Assert.That(status.Failed, Is.True); @@ -104,7 +104,7 @@ public async Task When_a_group_is_prepared_with_three_batches_and_SC_is_restarte NullLogger.Instance); // Needs index RetryBatches_ByStatus_ReduceInitialBatchSize - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await processor.ProcessBatches(); // mark ready @@ -219,7 +219,7 @@ public async Task When_a_group_has_one_batch_out_of_two_forwarded_the_status_is_ var processor = new RetryProcessor(RetryBatchesStore, domainEvents, new TestReturnToSenderDequeuer(returnToSender, ErrorStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()), retryManager, new Lazy(() => sender), new RecordingMessageActionAuditLog(), NullLogger.Instance); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await processor.ProcessBatches(); // mark ready await processor.ProcessBatches(); @@ -255,11 +255,11 @@ public async Task When_a_selection_is_staged_each_message_is_audited_as_a_batch( }).ToArray(); await ErrorStore.StoreFailedMessagesForTestsOnly(messages); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var gateway = new CustomRetriesGateway(true, RetryStore, retryManager); await gateway.StartRetryForMessageSelection(ids, user, operationId); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var audit = new RecordingMessageActionAuditLog(); var sender = new TestSender(); @@ -346,19 +346,19 @@ async Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryMa // Needs index FailedMessages_ByGroup // Needs index FailedMessages_UniqueMessageIdAndTimeOfFailures - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); var documentManager = new CustomRetryDocumentManager(progressToStaged, RetryStore, retryManager); var gateway = new CustomRetriesGateway(progressToStaged, RetryStore, retryManager); gateway.EnqueueRetryForFailureGroup(new RetriesGateway.RetryForFailureGroup(groupId, "Test-Context", groupType: null, DateTime.UtcNow, initiatedBy, operationId)); - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); await gateway.ProcessNextBulkRetry(); // Wait for indexes to catch up - CompleteDatabaseOperation(); + await CompleteDatabaseOperation(); } class CustomRetriesGateway : RetriesGateway