Recover service-link profiles and preserve orchestration correlation

Directory.Build.props

@@ -1,7 +1,7 @@
 <Project>
   <PropertyGroup>
     <VersionPrefix Condition="'$(VersionPrefix)' == ''">0.1.1</VersionPrefix>
-    <VersionSuffix Condition="'$(VersionSuffix)' == ''">beta.11</VersionSuffix>
+    <VersionSuffix Condition="'$(VersionSuffix)' == ''">beta.12</VersionSuffix>
 
     <Version Condition="'$(Version)' == '' and '$(VersionSuffix)' == ''">$(VersionPrefix)</Version>
     <Version Condition="'$(Version)' == '' and '$(VersionSuffix)' != ''">$(VersionPrefix)-$(VersionSuffix)</Version>

src/Helpdesk.API/Endpoints/ServiceLink/ServiceLinkEndpoints.cs

@@ -124,7 +124,8 @@ private static async Task<IResult> Respond<T>(HttpContext http, Func<Task<T>> ac
         { return Results.Problem(statusCode: error.StatusCode, title: error.Message, extensions: new Dictionary<string, object?> { ["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<string, object?> { ["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<string, object?> { ["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<string, object?> { ["code"] = "protected-state-unavailable" }); }

src/Helpdesk.Infrastructure/Orchestration/OrchestrationInternalClient.cs

@@ -88,6 +88,8 @@ public async Task<OrchestrationIngestResult> 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);

src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkDatabaseConflict.cs

@@ -0,0 +1,24 @@
+using Microsoft.EntityFrameworkCore;
+using Npgsql;
+
+namespace Helpdesk.Infrastructure.ServiceLink;
+
+/// <summary>Database errors that prove PostgreSQL aborted this transaction.</summary>
+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;
+    }
+}

src/Helpdesk.Infrastructure/ServiceLink/ServiceLinkWorker.cs

@@ -78,8 +78,13 @@ private async Task<ServiceLinkAttempt> 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);
             }
         }

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<string> preparedAcknowledgementOperations = new();
+    private readonly ConcurrentQueue<string> exchangeRequestFingerprints = new();
     private readonly ConcurrentDictionary<string, string> 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);

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<IServiceCollection, IConfiguration> configureServices;
     private readonly Action<WebApplication> mapEndpoints;
     private readonly Dictionary<string, string?> configuration;
+    private Socket? listenSocket;
     private PostgreSqlContainer? container;
     private WebApplication? app;
     public HttpClient Client { get; private set; } = null!;
@@ -49,9 +51,8 @@ private ServiceLinkKestrelPeer(Action<IServiceCollection, IConfiguration> 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,38 +80,74 @@ public static async Task<ServiceLinkKestrelPeer> CreateAsync(bool postgres,
         Action<IServiceCollection, IConfiguration> configureServices, Action<WebApplication> 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<RatelDeskIdentityDbContext>();
+            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<HelpdeskDbContext>();
+            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<RatelDeskIdentityDbContext>();
-        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<HelpdeskDbContext>();
-        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()
     {
         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<SocketTransportOptions>(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<AuthenticationSchemeOptions> options,