diff --git a/Directory.Build.props b/Directory.Build.props index 6c5699a2..3d151ee9 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,7 +1,7 @@ 0.1.1 - beta.11 + beta.12 $(VersionPrefix) $(VersionPrefix)-$(VersionSuffix) diff --git a/src/Helpdesk.API/Endpoints/ServiceLink/ServiceLinkEndpoints.cs b/src/Helpdesk.API/Endpoints/ServiceLink/ServiceLinkEndpoints.cs index b9da3007..c64d2b9a 100644 --- a/src/Helpdesk.API/Endpoints/ServiceLink/ServiceLinkEndpoints.cs +++ b/src/Helpdesk.API/Endpoints/ServiceLink/ServiceLinkEndpoints.cs @@ -124,7 +124,8 @@ private static async Task Respond(HttpContext http, Func> ac { return Results.Problem(statusCode: error.StatusCode, title: error.Message, extensions: new Dictionary { ["code"] = error.Code }); } catch (Exception error) when (error is JsonException or DecoderFallbackException or ArgumentException or FormatException) { return Results.Problem(statusCode: 400, title: "The service-link request is invalid.", extensions: new Dictionary { ["code"] = "invalid-request" }); } - catch (Exception error) when (error is DbUpdateException or ServiceClientConflictException or IntegrationProviderConfigurationConflictException) + catch (Exception error) when (error is DbUpdateException or ServiceClientConflictException or IntegrationProviderConfigurationConflictException || + ServiceLinkDatabaseConflict.IsAbortedTransaction(error)) { return Results.Problem(statusCode: 409, title: "The durable registration or profile changed. Refresh the current state.", extensions: new Dictionary { ["code"] = "service-link-conflict" }); } catch (CryptographicException) { return Results.Problem(statusCode: 503, title: "The protected link state is unavailable. Restore the shared key ring.", extensions: new Dictionary { ["code"] = "protected-state-unavailable" }); } diff --git a/src/Helpdesk.Infrastructure/Orchestration/OrchestrationInternalClient.cs b/src/Helpdesk.Infrastructure/Orchestration/OrchestrationInternalClient.cs index d8b3e4bc..3ea56480 100644 --- a/src/Helpdesk.Infrastructure/Orchestration/OrchestrationInternalClient.cs +++ b/src/Helpdesk.Infrastructure/Orchestration/OrchestrationInternalClient.cs @@ -88,6 +88,8 @@ public async Task IngestAsync( options: new JsonSerializerOptions(JsonSerializerDefaults.General)) }; + if (!string.IsNullOrEmpty(requestPayload.CorrelationId)) + request.Headers.Add("X-Correlation-Id", requestPayload.CorrelationId); await AttachAuthHeaderAsync(settings, request, cancellationToken); var client = _httpClientFactory.CreateClient(settings.ServiceLink is null ? "OrchestrationInternalApi" : ServiceLinkOutboundNetwork.BusinessClientName); diff --git a/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkDatabaseConflict.cs b/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkDatabaseConflict.cs new file mode 100644 index 00000000..c1ba73ad --- /dev/null +++ b/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkDatabaseConflict.cs @@ -0,0 +1,24 @@ +using Microsoft.EntityFrameworkCore; +using Npgsql; + +namespace Helpdesk.Infrastructure.ServiceLink; + +/// Database errors that prove PostgreSQL aborted this transaction. +public static class ServiceLinkDatabaseConflict +{ + public static bool IsAbortedTransaction(Exception error) + { + // EF's nonretrying Npgsql strategy can wrap a SaveChanges serialization + // failure in InvalidOperationException -> DbUpdateException. Recognize + // only those known wrappers and the two explicit transaction-abort codes. + for (var depth = 0; depth < 4; depth++) + { + if (error is PostgresException postgres) + return postgres.SqlState is PostgresErrorCodes.SerializationFailure or PostgresErrorCodes.DeadlockDetected; + if (error is not (InvalidOperationException or DbUpdateException) || error.InnerException is null) + return false; + error = error.InnerException; + } + return false; + } +} diff --git a/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkWorker.cs b/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkWorker.cs index be7ab9a2..5ca42c74 100644 --- a/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkWorker.cs +++ b/src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkWorker.cs @@ -78,8 +78,13 @@ private async Task ProgressWithRetry(ServiceLinkAttempt a, C for (var retry = 0; ; retry++) { try { await Progress(a, ct); return a; } - catch (DbUpdateConcurrencyException) when (retry < 3) + catch (Exception error) when (retry < 3 && + (error is DbUpdateConcurrencyException || !ct.IsCancellationRequested && ServiceLinkDatabaseConflict.IsAbortedTransaction(error))) { + // Progress uses the existing protected handoff and durable operation journals. + // Only a known database abort may replay that recovery; an unknown commit, + // transport failure, or arbitrary transient error must leave this scope. + if (db.Database.CurrentTransaction is not null || System.Transactions.Transaction.Current is not null) throw; var attemptId = a.AttemptId; db.ChangeTracker.Clear(); a = await Attempt(attemptId, ct); } } diff --git a/tests/Helpdesk.Tests/Api/NetRatelServiceLinkContractPeer.cs b/tests/Helpdesk.Tests/Api/NetRatelServiceLinkContractPeer.cs index 564e261c..2d7a292b 100644 --- a/tests/Helpdesk.Tests/Api/NetRatelServiceLinkContractPeer.cs +++ b/tests/Helpdesk.Tests/Api/NetRatelServiceLinkContractPeer.cs @@ -46,6 +46,7 @@ internal sealed class NetRatelServiceLinkContractPeer : IAsyncDisposable private bool revoked; private volatile bool losePreparedAcknowledgementResponses; private readonly ConcurrentQueue preparedAcknowledgementOperations = new(); + private readonly ConcurrentQueue exchangeRequestFingerprints = new(); private readonly ConcurrentDictionary preparedAcknowledgements = new(StringComparer.Ordinal); public string BaseUrl { get; } public string InstanceId { get; } = Guid.NewGuid().ToString("D"); @@ -65,6 +66,7 @@ public bool LosePreparedAcknowledgementResponses set => losePreparedAcknowledgementResponses = value; } public string[] PreparedAcknowledgementOperationIds => preparedAcknowledgementOperations.ToArray(); + public string[] ExchangeRequestFingerprints => exchangeRequestFingerprints.ToArray(); public string? InvalidMetadataMember { get; set; } public ServiceDirectionalCredential InboundCredential => inbound!; public ServiceDirectionalCredential OutboundCredential => outbound!; @@ -394,6 +396,7 @@ private void Map() { if (!Proof(request, attemptId) || request.GrantHash != GrantHash) return Results.Unauthorized(); var fingerprint = ServiceLinkCanonicalJson.HashObject(request); + exchangeRequestFingerprints.Enqueue(fingerprint); if (exchangeFingerprint is not null) return fingerprint == exchangeFingerprint ? Results.Json(exchangeResponse) : Results.Conflict(); var reverse = summary.Grants.Single(x => x.DirectionId == ServiceLinkContract.ResponderToInitiator); diff --git a/tests/Helpdesk.Tests/Api/ServiceLinkKestrelPeer.cs b/tests/Helpdesk.Tests/Api/ServiceLinkKestrelPeer.cs index 98ba0ae1..8472703b 100644 --- a/tests/Helpdesk.Tests/Api/ServiceLinkKestrelPeer.cs +++ b/tests/Helpdesk.Tests/Api/ServiceLinkKestrelPeer.cs @@ -16,6 +16,7 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.DataProtection; using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -34,6 +35,7 @@ internal sealed class ServiceLinkKestrelPeer : IAsyncDisposable private readonly Action configureServices; private readonly Action mapEndpoints; private readonly Dictionary configuration; + private Socket? listenSocket; private PostgreSqlContainer? container; private WebApplication? app; public HttpClient Client { get; private set; } = null!; @@ -49,9 +51,8 @@ private ServiceLinkKestrelPeer(Action config { this.configureServices = configureServices; this.mapEndpoints = mapEndpoints; - using var reservation = new TcpListener(IPAddress.Loopback, 0); - reservation.Start(); - BaseUrl = $"http://127.0.0.1:{((IPEndPoint)reservation.LocalEndpoint).Port}"; + listenSocket = CreateListenSocket(0); + BaseUrl = $"http://127.0.0.1:{((IPEndPoint)listenSocket.LocalEndPoint!).Port}"; configuration = new() { ["StorageOptions:RootPath"] = Path.Combine(root, "storage"), @@ -79,31 +80,56 @@ public static async Task CreateAsync(bool postgres, Action configureServices, Action mapEndpoints) { var peer = new ServiceLinkKestrelPeer(configureServices, mapEndpoints); - Directory.CreateDirectory(peer.root); - peer.configuration["Database:Provider"] = postgres ? "PostgreSql" : "Sqlite"; - peer.configuration["Database:Sqlite:Path"] = Path.Combine(peer.root, "service-links.db"); - if (postgres) + try { - peer.container = new PostgreSqlBuilder("postgres:16").Build(); - await peer.container.StartAsync(); - peer.configuration["ConnectionStrings:HelpdeskDb"] = peer.container.GetConnectionString(); + Directory.CreateDirectory(peer.root); + peer.configuration["Database:Provider"] = postgres ? "PostgreSql" : "Sqlite"; + peer.configuration["Database:Sqlite:Path"] = Path.Combine(peer.root, "service-links.db"); + if (postgres) + { + peer.container = new PostgreSqlBuilder("postgres:16").Build(); + await peer.container.StartAsync(); + peer.configuration["ConnectionStrings:HelpdeskDb"] = peer.container.GetConnectionString(); + } + await peer.StartAsync(); + await using var scope = peer.Services.CreateAsyncScope(); + var identity = scope.ServiceProvider.GetRequiredService(); + await identity.Database.MigrateAsync(); + identity.Users.Add(new ApplicationUser { Id = "owner", UserName = "owner", Email = "owner@example.test", + IsEnabled = true, IsInstanceAdministrator = true }); + await identity.SaveChangesAsync(); + var db = scope.ServiceProvider.GetRequiredService(); + await db.Database.MigrateAsync(); + db.InstanceInitializations.Add(new() { InstanceId = peer.InstanceId, OperationId = Guid.NewGuid(), + SetupVersion = "service-link-contract-test", CompletedAtUtc = peer.Clock.GetUtcNow() }); + db.Organizations.Add(new() { Id = peer.OrganizationId, Name = "Synthetic link organization", IsEnabled = true }); + db.Customers.Add(new() { Id = peer.CustomerId, OrganizationId = peer.OrganizationId, + Name = "Synthetic link customer", Email = "customer@example.test", IsEnabled = true }); + await db.SaveChangesAsync(); + return peer; + } + catch + { + await peer.DisposeAsync(); + throw; + } + } + + private static Socket CreateListenSocket(int port) + { + var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + try + { + socket.Bind(new IPEndPoint(IPAddress.Loopback, port)); + // Keep the allocated authority reserved until Kestrel owns this same socket. + socket.Listen(new SocketTransportOptions().Backlog); + return socket; + } + catch + { + socket.Dispose(); + throw; } - await peer.StartAsync(); - await using var scope = peer.Services.CreateAsyncScope(); - var identity = scope.ServiceProvider.GetRequiredService(); - await identity.Database.MigrateAsync(); - identity.Users.Add(new ApplicationUser { Id = "owner", UserName = "owner", Email = "owner@example.test", - IsEnabled = true, IsInstanceAdministrator = true }); - await identity.SaveChangesAsync(); - var db = scope.ServiceProvider.GetRequiredService(); - await db.Database.MigrateAsync(); - db.InstanceInitializations.Add(new() { InstanceId = peer.InstanceId, OperationId = Guid.NewGuid(), - SetupVersion = "service-link-contract-test", CompletedAtUtc = peer.Clock.GetUtcNow() }); - db.Organizations.Add(new() { Id = peer.OrganizationId, Name = "Synthetic link organization", IsEnabled = true }); - db.Customers.Add(new() { Id = peer.CustomerId, OrganizationId = peer.OrganizationId, - Name = "Synthetic link customer", Email = "customer@example.test", IsEnabled = true }); - await db.SaveChangesAsync(); - return peer; } private async Task StartAsync() @@ -111,6 +137,17 @@ private async Task StartAsync() var builder = WebApplication.CreateBuilder(new WebApplicationOptions { EnvironmentName = "Production" }); builder.Logging.ClearProviders(); builder.WebHost.UseUrls(BaseUrl); + var socket = listenSocket ?? throw new InvalidOperationException("Peer listener is not available."); + var expectedEndpoint = new IPEndPoint(IPAddress.Loopback, new Uri(BaseUrl).Port); + var socketClaimed = 0; + builder.Services.Configure(options => options.CreateBoundListenSocket = endpoint => + { + if (!expectedEndpoint.Equals(endpoint)) + throw new InvalidOperationException("Peer listener authority changed."); + if (Interlocked.Exchange(ref socketClaimed, 1) != 0) + throw new InvalidOperationException("Peer listener was requested more than once."); + return socket; + }); builder.Configuration.AddInMemoryCollection(configuration); builder.Services.AddDataProtection().PersistKeysToFileSystem(new DirectoryInfo(Path.Combine(root, "keys"))) .SetApplicationName("RatelDeskServiceLinkContractTests"); @@ -157,7 +194,10 @@ private async Task StartAsync() public async Task RestartAsync() { Client.Dispose(); - await app!.DisposeAsync(); + try { await app!.DisposeAsync(); } + finally { listenSocket?.Dispose(); listenSocket = null; } + // Restart uses the established authority once; an unavailable port remains a failure. + listenSocket = CreateListenSocket(new Uri(BaseUrl).Port); Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); await StartAsync(); } @@ -184,10 +224,16 @@ private static object ProjectLifecycle(string path, object body) => body is Serv public async ValueTask DisposeAsync() { Client?.Dispose(); - if (app is not null) await app.DisposeAsync(); - if (container is not null) await container.DisposeAsync(); - Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); - if (Directory.Exists(root)) Directory.Delete(root, true); + try { if (app is not null) await app.DisposeAsync(); } + finally + { + // Also covers failure before Kestrel accepts the socket or before database startup completes. + listenSocket?.Dispose(); + listenSocket = null; + if (container is not null) await container.DisposeAsync(); + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); + if (Directory.Exists(root)) Directory.Delete(root, true); + } } private sealed class LifecycleAdminAuthentication(IOptionsMonitor options, diff --git a/tests/Helpdesk.Tests/Api/ServiceLinkSerializationPostgresTests.cs b/tests/Helpdesk.Tests/Api/ServiceLinkSerializationPostgresTests.cs new file mode 100644 index 00000000..9ac61bec --- /dev/null +++ b/tests/Helpdesk.Tests/Api/ServiceLinkSerializationPostgresTests.cs @@ -0,0 +1,193 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.Http.Json; +using Helpdesk.Application.Orchestration; +using Helpdesk.Infrastructure.Persistence.Connectivity; +using Helpdesk.Infrastructure.Persistence; +using Helpdesk.Infrastructure.ServiceIdentity; +using Helpdesk.Infrastructure.ServiceLink; +using Helpdesk.Shared.DTOs.Orchestration; +using Helpdesk.Shared.ServiceLink; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Npgsql; +using Xunit; + +namespace Helpdesk.Tests.Api; + +public sealed partial class ServiceLinkLifecycleTests +{ + [Theory] + [InlineData(1, false)] + [InlineData(4, false)] + [InlineData(1, true)] + public async Task Postgres_staged_exchange_abort_recovers_same_handoff_with_bounded_reload_and_no_partial_profile(int abortCount, bool backgroundWork) + { + var race = new StagedProfileSerializationRace(abortCount); + await using var peer = await NetRatelServiceLinkContractPeer.CreateAsync(); + await using var local = await LocalAsync(postgres: true, saveInterceptor: race); + // An unused, disabled profile is created through the production configuration service. + await using (var emptyProfile = local.Services.CreateAsyncScope()) + { + var providers = emptyProfile.ServiceProvider.GetRequiredService(); + var initial = await providers.UpdateOrchestratorSettingsAsync(new UpdateOrchestrationConnectivitySettingsDto + { ExpectedRevision = 0, Enabled = false }); + Assert.Equal(1, initial.Revision); + } + var start = await StartAsync(local, peer); + var callback = await peer.ApproveAsync(local.BaseUrl, start.AttemptId, BrowserState(start), SessionBinding); + using (var review = await local.AdminAsync(HttpMethod.Post, "/api/v1/admin/service-links/callback", callback)) + Assert.Equal(HttpStatusCode.OK, review.StatusCode); + using (var approval = await local.AdminAsync(HttpMethod.Post, + "/api/v1/admin/service-links/attempts/" + start.AttemptId + "/approve", + new ServiceLinkLocalApproveRequest(peer.GrantHash, SessionBinding))) + Assert.Equal(HttpStatusCode.OK, approval.StatusCode); + + ServicePrincipalRegistration originalPrincipal; + ServicePrincipalSecret originalSecret; + await using (var approved = local.Services.CreateAsyncScope()) + { + var db = approved.ServiceProvider.GetRequiredService(); + originalPrincipal = Assert.Single(await db.Set().AsNoTracking().ToListAsync()); + originalSecret = Assert.Single(await db.Set().AsNoTracking().ToListAsync()); + } + + race.Arm(); + var expectedNextWorkAt = local.Clock.GetUtcNow().ToUnixTimeSeconds() + + local.Services.GetRequiredService>().CurrentValue.WorkerIntervalSeconds; + if (backgroundWork) + { + await using var work = local.Services.CreateAsyncScope(); + var workerDb = work.ServiceProvider.GetRequiredService(); + await work.ServiceProvider.GetRequiredService().WorkAsync(CancellationToken.None); + var tracked = Assert.Single(workerDb.ChangeTracker.Entries()); + Assert.Equal(EntityState.Unchanged, tracked.State); + Assert.Equal(expectedNextWorkAt, tracked.Entity.NextWorkAtUnixSeconds); + Assert.Null(workerDb.Database.CurrentTransaction); + } + else + { + using var resumed = await ResumeAsync(local, peer.LinkId); + Assert.Equal(abortCount == 1 ? HttpStatusCode.OK : HttpStatusCode.Conflict, resumed.StatusCode); + if (abortCount == 4) + { + var problem = await resumed.Content.ReadFromJsonAsync(); + Assert.Equal("service-link-conflict", problem.GetProperty("code").GetString()); + } + } + Assert.Equal(abortCount, race.AbortedSaveSqlStates.Length); + Assert.All(race.AbortedSaveSqlStates, state => Assert.Equal(PostgresErrorCodes.SerializationFailure, state)); + Assert.Equal(abortCount == 1 ? 2 : 4, race.StagedTransactionIds.Length); + Assert.Equal(race.StagedTransactionIds.Length, race.StagedTransactionIds.Distinct().Count()); + Assert.Equal(abortCount == 1 ? 2 : 4, peer.ExchangeRequestFingerprints.Length); + Assert.Single(peer.ExchangeRequestFingerprints.Distinct(StringComparer.Ordinal)); + + if (abortCount == 4) + { + // Exhaustion reports the known abort instead of saving tracked, rolled-back state. + await using (var aborted = local.Services.CreateAsyncScope()) + { + var db = aborted.ServiceProvider.GetRequiredService(); + var attempt = await db.Set().AsNoTracking().SingleAsync(); + Assert.True(attempt.ExchangeDispatched); + Assert.Null(attempt.ProtectedOutboundCredential); + Assert.False(attempt.LocalInboundActive); + Assert.False(attempt.LocalBusinessSenderEnabled); + var profile = await db.M2MConnectivitySettings.AsNoTracking().SingleAsync(); + Assert.Null(profile.LinkId); + Assert.Equal(5, profile.Revision); + Assert.True(string.IsNullOrEmpty(profile.ProtectedClientSecret)); + Assert.False(profile.Enabled); + Assert.Equal(0, await db.Set().CountAsync()); + } + await local.RestartAsync(); + using var recovered = await ResumeAsync(local, peer.LinkId); + Assert.Equal(HttpStatusCode.OK, recovered.StatusCode); + Assert.Equal(5, peer.ExchangeRequestFingerprints.Length); + Assert.Single(peer.ExchangeRequestFingerprints.Distinct(StringComparer.Ordinal)); + } + + await using var final = local.Services.CreateAsyncScope(); + var finalDb = final.ServiceProvider.GetRequiredService(); + var prepared = await finalDb.Set().AsNoTracking().SingleAsync(); + Assert.Equal("prepared", prepared.LifecycleState); + Assert.NotNull(prepared.ProtectedOutboundCredential); + Assert.False(prepared.LocalPreparedAcknowledged); + Assert.False(prepared.LocalInboundActive); + Assert.False(prepared.LocalBusinessSenderEnabled); + if (backgroundWork) Assert.Equal(expectedNextWorkAt, prepared.NextWorkAtUnixSeconds); + var currentPrincipal = Assert.Single(await finalDb.Set().AsNoTracking().ToListAsync()); + Assert.Equal(originalPrincipal.Id, currentPrincipal.Id); + Assert.Equal(originalPrincipal.CurrentCredentialRevision, currentPrincipal.CurrentCredentialRevision); + Assert.Equal(originalPrincipal.Revision, currentPrincipal.Revision); + Assert.Equal(originalPrincipal.Version, currentPrincipal.Version); + Assert.Equal(originalPrincipal.Status, currentPrincipal.Status); + var currentSecret = Assert.Single(await finalDb.Set().AsNoTracking().ToListAsync()); + Assert.Equal(originalSecret.ServicePrincipalId, currentSecret.ServicePrincipalId); + Assert.Equal(originalSecret.SecretHash, currentSecret.SecretHash); + Assert.Equal(originalSecret.Salt, currentSecret.Salt); + Assert.Equal(originalSecret.CredentialRevision, currentSecret.CredentialRevision); + Assert.Equal(originalSecret.Status, currentSecret.Status); + var binding = Assert.Single(await finalDb.IncidentReceiverPrincipalBindings.AsNoTracking().ToListAsync()); + Assert.False(binding.IsEnabled); + var staged = await finalDb.M2MConnectivitySettings.AsNoTracking().SingleAsync(); + Assert.Equal(peer.LinkId, staged.LinkId); + Assert.Equal(2 + abortCount, staged.Revision); + Assert.Equal(staged.Revision, prepared.OutboundProfileRevision); + Assert.False(staged.Enabled); + Assert.False(staged.ManagedSenderEnabled); + var resolved = await final.ServiceProvider.GetRequiredService().GetResolvedOrchestratorSettingsAsync(); + Assert.Equal(peer.InboundCredential.ClientId, resolved.ClientId); + Assert.Equal(peer.InboundCredential.ClientSecret, resolved.ClientSecret); + Assert.False(resolved.Enabled); + Assert.Equal(0, await finalDb.Set().CountAsync()); + } + + private sealed class StagedProfileSerializationRace(int abortLimit) : SaveChangesInterceptor + { + private readonly ConcurrentQueue abortedSaveSqlStates = new(); + private readonly ConcurrentQueue stagedTransactionIds = new(); + private int armed; + private int stagedSaves; + public string[] AbortedSaveSqlStates => abortedSaveSqlStates.ToArray(); + public Guid[] StagedTransactionIds => stagedTransactionIds.ToArray(); + public void Arm() => Volatile.Write(ref armed, 1); + + public override async ValueTask> SavingChangesAsync(DbContextEventData eventData, + InterceptionResult result, CancellationToken cancellationToken = default) + { + var context = eventData.Context; + if (Volatile.Read(ref armed) == 0 || context?.Database.CurrentTransaction is null) return result; + var entry = context.ChangeTracker.Entries().SingleOrDefault(candidate => + candidate.State == EntityState.Modified && candidate.Entity.ProviderKey == "Orchestrator" && candidate.Entity.LinkId is not null); + if (entry is null) return result; + stagedTransactionIds.Enqueue(context.Database.CurrentTransaction.TransactionId); + if (Interlocked.Increment(ref stagedSaves) > abortLimit) return result; + + // The coordinator has already read this row in its real Serializable snapshot. + // A separate committed writer changes it before the attempted staged UPDATE. + // PostgreSQL, not the interceptor, raises 40001 and aborts that transaction. + await using var writer = new NpgsqlConnection(context.Database.GetConnectionString()); + await writer.OpenAsync(cancellationToken); + await using var update = writer.CreateCommand(); + update.CommandText = "UPDATE \"M2MConnectivitySettings\" SET \"Revision\" = \"Revision\" + 1 WHERE \"Id\" = @id"; + update.Parameters.AddWithValue("id", entry.Entity.Id); + Assert.Equal(1, await update.ExecuteNonQueryAsync(cancellationToken)); + return result; + } + + public override Task SaveChangesFailedAsync(DbContextErrorEventData eventData, + CancellationToken cancellationToken = default) + { + for (Exception? error = eventData.Exception; error is not null; error = error.InnerException) + if (error is PostgresException postgres) + { + abortedSaveSqlStates.Enqueue(postgres.SqlState); + break; + } + return Task.CompletedTask; + } + } +} diff --git a/tests/Helpdesk.Tests/Infrastructure/Orchestration/OrchestrationInternalClientTests.cs b/tests/Helpdesk.Tests/Infrastructure/Orchestration/OrchestrationInternalClientTests.cs index 8088393e..18538735 100644 --- a/tests/Helpdesk.Tests/Infrastructure/Orchestration/OrchestrationInternalClientTests.cs +++ b/tests/Helpdesk.Tests/Infrastructure/Orchestration/OrchestrationInternalClientTests.cs @@ -92,6 +92,34 @@ public async Task NetRatel_wire_names_are_used_and_real_acknowledgement_is_requi Assert.DoesNotContain("ClientSecret", body, StringComparison.OrdinalIgnoreCase); } + [Theory] + [InlineData("corr-1")] + [InlineData("")] + public async Task Ingest_preserves_submitted_correlation_in_body_and_optional_header(string correlation) + { + string? transmittedCorrelation = null; + var handler = new RecordingHandler(request => + { + if (request.Headers.TryGetValues("X-Correlation-Id", out var values)) + transmittedCorrelation = Assert.Single(values); + Assert.Equal("Bearer", request.Headers.Authorization?.Scheme); + Assert.Equal("token", request.Headers.Authorization?.Parameter); + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("""{"executionId":"99","status":"Accepted"}""") + }; + }); + var submitted = Request(); + submitted.CorrelationId = correlation; + + var result = await Client(handler).IngestAsync(Settings(), submitted); + + Assert.Equal("99", result.ExecutionId); + using var body = JsonDocument.Parse(Assert.Single(handler.RequestBodies)); + Assert.Equal(correlation, body.RootElement.GetProperty("CorrelationId").GetString()); + Assert.Equal(correlation.Length == 0 ? null : correlation, transmittedCorrelation); + } + [Theory] [InlineData("Accepted", OrchestrationSubmissionDisposition.Admitted, OrchestrationExecutionOutcome.Processing)] [InlineData("Processing", OrchestrationSubmissionDisposition.Admitted, OrchestrationExecutionOutcome.Processing)] diff --git a/tests/Helpdesk.Tests/Infrastructure/ServiceLink/ServiceLinkDatabaseConflictTests.cs b/tests/Helpdesk.Tests/Infrastructure/ServiceLink/ServiceLinkDatabaseConflictTests.cs new file mode 100644 index 00000000..68385db7 --- /dev/null +++ b/tests/Helpdesk.Tests/Infrastructure/ServiceLink/ServiceLinkDatabaseConflictTests.cs @@ -0,0 +1,39 @@ +using Helpdesk.Infrastructure.ServiceLink; +using Microsoft.EntityFrameworkCore; +using Npgsql; +using Xunit; + +namespace Helpdesk.Tests.Infrastructure.ServiceLink; + +public sealed class ServiceLinkDatabaseConflictTests +{ + [Theory] + [InlineData(PostgresErrorCodes.SerializationFailure)] + [InlineData(PostgresErrorCodes.DeadlockDetected)] + public void Only_known_Postgres_transaction_aborts_and_EF_wrappers_are_recoverable(string sqlState) + { + var postgres = new PostgresException("Synthetic transaction abort", "ERROR", "ERROR", sqlState); + Assert.True(ServiceLinkDatabaseConflict.IsAbortedTransaction(postgres)); + Assert.True(ServiceLinkDatabaseConflict.IsAbortedTransaction(new DbUpdateException("Synthetic save", postgres))); + Assert.True(ServiceLinkDatabaseConflict.IsAbortedTransaction(new InvalidOperationException("Synthetic EF strategy", + new DbUpdateException("Synthetic save", postgres)))); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(new HttpRequestException("Synthetic transport", postgres))); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(new Exception("Unknown application wrapper", postgres))); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(new AggregateException(postgres))); + } + + [Theory] + [InlineData(PostgresErrorCodes.UniqueViolation)] + [InlineData(PostgresErrorCodes.ConnectionFailure)] + [InlineData(PostgresErrorCodes.QueryCanceled)] + [InlineData(PostgresErrorCodes.AdminShutdown)] + public void Other_provider_errors_and_unknown_commit_failures_are_not_transaction_abort_retries(string sqlState) + { + var postgres = new PostgresException("Synthetic unrecoverable error", "ERROR", "ERROR", sqlState); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(postgres)); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(new InvalidOperationException("Synthetic EF strategy", + new DbUpdateException("Synthetic save", postgres)))); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(new InvalidOperationException("Unrelated programming failure"))); + Assert.False(ServiceLinkDatabaseConflict.IsAbortedTransaction(new DbUpdateException("Unknown commit", new IOException("Synthetic connection loss")))); + } +}