Fix native Aspire launch and maintenance work ownership

Use the public native AddProject launch, retain exact process/source evidence, and make maintenance work transfer and joined shutdown explicit. Validated primary AppHost plus runtime dependencies with Release build -warnaserror: 0 warnings, 0 errors. Full solution and live storage-mode qualification remain open.

Author: KSemenenko

build/KeyLoad.AppHost/ClusterReplication/Commands/ProcessKillCommand.cs

@@ -39,7 +39,9 @@ internal static void Apply(IDistributedApplicationBuilder builder, IResourceBuil
 
     private static InteractionInput Input(string name) => new()
     {
-        Name = name, InputType = InputType.Text, Required = true,
+        Name = name,
+        InputType = InputType.Text,
+        Required = true,
         MaxLength = ProcessFaultCommandProtocol.MaximumIdentityCharacters
     };
 

build/KeyLoad.AppHost/ClusterReplication/Observations/CompiledSourceEvidence.cs

@@ -16,6 +16,11 @@ internal static class CompiledSourceEvidence
     private const string Central = Prefix + "Central";
     private const string Producer = Prefix + "Producer";
     private const string SourceCount = Prefix + "SourceCount";
+    private const string AttributeSeparator = "=";
+    private const char InputSeparator = '\n';
+    private const string InvalidAttribute = "The compiled source attribute is invalid.";
+    private const string IncompleteIdentity = "The observed managed assembly has no complete compilation source identity.";
+    private const int UnobservedSourceCount = 0;
     private const int CustomAttributeProlog = 1;
     private const int MinimumSources = 1;
     private const int MaximumSources = 5_000;
@@ -26,7 +31,7 @@ internal static (int Count, string Sha256) Read(string assembly)
         using var portable = new PEReader(stream, PEStreamOptions.PrefetchMetadata);
         var reader = portable.GetMetadataReader();
         var inputs = new List<string>();
-        var count = 0;
+        var count = UnobservedSourceCount;
         foreach (var handle in reader.GetAssemblyDefinition().GetCustomAttributes())
         {
             var attribute = reader.GetCustomAttribute(handle);
@@ -37,24 +42,25 @@ internal static (int Count, string Sha256) Read(string assembly)
             var value = reader.GetBlobReader(attribute.Value);
             if (value.ReadUInt16() != CustomAttributeProlog)
             {
-                throw new InvalidDataException("The compiled source attribute is invalid.");
+                throw new InvalidDataException(InvalidAttribute);
             }
             var key = value.ReadSerializedString();
             var text = value.ReadSerializedString();
             if (key is Source or Central or Producer && text is not null)
             {
-                inputs.Add(key + "=" + text);
+                inputs.Add(key + AttributeSeparator + text);
             }
             if (key == SourceCount)
             {
                 count = int.Parse(text!, NumberStyles.None, CultureInfo.InvariantCulture);
             }
         }
-        if (count is < MinimumSources or > MaximumSources || inputs.Count(value => value.StartsWith(Source + "=", StringComparison.Ordinal)) != count)
+        if (count is < MinimumSources or > MaximumSources
+            || inputs.Count(value => value.StartsWith(Source + AttributeSeparator, StringComparison.Ordinal)) != count)
         {
-            throw new InvalidDataException("The observed managed assembly has no complete compilation source identity.");
+            throw new InvalidDataException(IncompleteIdentity);
         }
-        var canonical = string.Join('\n', inputs.Order(StringComparer.Ordinal));
+        var canonical = string.Join(InputSeparator, inputs.Order(StringComparer.Ordinal));
         return (count, Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(canonical))));
     }
 

build/KeyLoad.AppHost/ClusterReplication/Observations/ProjectProcessEvidence.cs

@@ -12,6 +12,13 @@ internal static class ProjectProcessEvidence
     private const string WorkDirectoryProperty = "executable.workdir";
     private const string AssemblySuffix = ".dll";
     private const string SymbolsSuffix = ".pdb";
+    private const string ExecutableUnavailable = "The native executable is unavailable.";
+    private const string EntryAssemblyUnavailable = "The managed entry assembly is unavailable.";
+    private const string ProcessChanged = "The native process changed during evidence capture.";
+    private const string EvidenceMismatch = "The native process or its compiled files differ from the observed identity.";
+    private const string AmbiguousEntryAssembly = "The native project launch does not identify exactly one managed entry assembly.";
+    private const int SingleEntryAssembly = 1;
+    private const int NoExplicitEntryAssembly = 0;
 
     internal static NativeProcessEvidence Read(ResourceNotificationService notifications, string resourceName)
     {
@@ -21,14 +28,14 @@ internal static NativeProcessEvidence Read(ResourceNotificationService notificat
             throw new InvalidOperationException(ProcessFaultCommandProtocol.Refused);
         }
         using var process = Process.GetProcessById(identity.ProcessId);
-        var executable = process.MainModule?.FileName ?? throw new InvalidOperationException("The native executable is unavailable.");
+        var executable = process.MainModule?.FileName ?? throw new InvalidOperationException(ExecutableUnavailable);
         var directory = (string)current.Snapshot.Properties.Single(property => property.Name == WorkDirectoryProperty).Value!;
         var arguments = (IEnumerable<string>)current.Snapshot.Properties.Single(property => property.Name == ArgumentsProperty).Value!;
         var managed = ResolveAssembly(project.GetProjectMetadata().ProjectPath, executable, directory, arguments);
         var binary = CaptureBinary(executable, managed);
         if (ProjectProcessIdentity.Read(notifications, resourceName) != identity)
         {
-            throw new InvalidOperationException("The native process changed during evidence capture.");
+            throw new InvalidOperationException(ProcessChanged);
         }
         return new(resourceName, identity.ProcessId, new DateTime(identity.ProcessStartTimeUtcTicks, DateTimeKind.Utc),
             new DateTime(identity.AspireStartTimeUtcTicks, DateTimeKind.Utc), binary);
@@ -37,8 +44,8 @@ internal static NativeProcessEvidence Read(ResourceNotificationService notificat
     internal static NativeProcessEvidence ReadCurrent(string name)
     {
         using var process = Process.GetCurrentProcess();
-        var executable = process.MainModule?.FileName ?? throw new InvalidOperationException("The native executable is unavailable.");
-        var assembly = Assembly.GetEntryAssembly()?.Location ?? throw new InvalidOperationException("The managed entry assembly is unavailable.");
+        var executable = process.MainModule?.FileName ?? throw new InvalidOperationException(ExecutableUnavailable);
+        var assembly = Assembly.GetEntryAssembly()?.Location ?? throw new InvalidOperationException(EntryAssemblyUnavailable);
         return new(name, process.Id, process.StartTime.ToUniversalTime(), null, CaptureBinary(executable, assembly));
     }
 
@@ -51,7 +58,7 @@ internal static void Verify(NativeProcessEvidence expected)
             || executable != expected.Binary.ExecutablePath
             || CaptureBinary(executable, expected.Binary.AssemblyPath) != expected.Binary)
         {
-            throw new InvalidDataException("The native process or its compiled files differ from the observed identity.");
+            throw new InvalidDataException(EvidenceMismatch);
         }
     }
 
@@ -61,15 +68,15 @@ private static string ResolveAssembly(string project, string executable, string
         var expected = name + AssemblySuffix;
         var explicitPaths = arguments.Where(argument => Path.GetFileName(argument) == expected)
             .Select(argument => Path.GetFullPath(argument, directory)).Distinct(StringComparer.Ordinal).ToArray();
-        if (explicitPaths.Length == 1)
+        if (explicitPaths.Length == SingleEntryAssembly)
         {
             return explicitPaths.Single();
         }
-        if (Path.GetFileNameWithoutExtension(executable) == name && explicitPaths.Length == 0)
+        if (Path.GetFileNameWithoutExtension(executable) == name && explicitPaths.Length == NoExplicitEntryAssembly)
         {
             return Path.ChangeExtension(executable, AssemblySuffix);
         }
-        throw new InvalidOperationException("The native project launch does not identify exactly one managed entry assembly.");
+        throw new InvalidOperationException(AmbiguousEntryAssembly);
     }
 
     private static NativeBinaryEvidence CaptureBinary(string executable, string assembly)

build/KeyLoad.AppHost/ClusterReplication/Resources/ClusterResources.cs

@@ -31,6 +31,7 @@ internal static class ClusterResources
     private const string ServerUrlsEnvironment = "ASPNETCORE_URLS";
     private const string LoopbackAddress = "127.0.0.1";
     private const int FirstNodeIndex = 0;
+    private const int LastNodeOffset = 1;
 
     /// <summary>Adds real native project silos; each process retains its own canonical storage directory.</summary>
     internal static IResourceBuilder<ProjectResource>[] Add(IDistributedApplicationBuilder builder,
@@ -42,7 +43,7 @@ internal static IResourceBuilder<ProjectResource>[] Add(IDistributedApplicationB
         var nodeNames = ClusterTopologyNodeNames.Read(benchmarkNodeCount, functionalNodeCount, ephemeral,
             options.Cluster.Value.SiloCount);
         var replicas = benchmarkNodeCount ?? options.Cluster.Value.ReplicaCount;
-        if (!ephemeral && (long)options.Cluster.Value.FirstPublicPort + nodeNames.Length - 1 > IPEndPoint.MaxPort)
+        if (!ephemeral && (long)options.Cluster.Value.FirstPublicPort + nodeNames.Length - LastNodeOffset > IPEndPoint.MaxPort)
         {
             throw new OptionsValidationException(Options.DefaultName, typeof(ClusterDeploymentOptions),
                 [ClusterDeploymentOptions.ValidationMessage]);
@@ -84,7 +85,6 @@ private static IResourceBuilder<ProjectResource> AddNode(IDistributedApplication
         ClusterProfileStore.PrepareDirectory(directory);
         var port = ephemeral ? (int?)null : AppHostOptionsRegistration.Get(builder).Cluster.Value.FirstPublicPort + index;
         var resource = builder.AddProject<Projects.KeyLoad_Server>(resourceName, launchProfileName: null)
-            .WithAnnotation(new ForceProcessExecutionAnnotation())
             .WithHttpEndpoint(port: port, name: HttpEndpoint, isProxied: false)
             .WithEndpoint(name: SiloEndpoint, scheme: TcpScheme, isExternal: false, isProxied: false);
         resource.WithEnvironment(ClusterEnvironment, ClusterPrefix + profile.Incarnation.ToString(ClusterGuidFormat))

build/KeyLoad.AppHost/Hosting/AppHostControlOptionsRegistration.cs

@@ -17,6 +17,7 @@ internal static class AppHostControlOptionsRegistration
     private const string EphemeralSetting = "KeyLoad:Ephemeral";
     private const string BenchmarksEnabledSetting = "Benchmarks:Enabled";
     private const string BenchmarkTargetSetting = "Benchmarks:Target";
+    private const string KeyLoadTarget = "KeyLoad";
     private const string BenchmarkSessionSetting = "Benchmarks:Session";
     private const string BenchmarkDataRootSetting = "Benchmarks:DataRoot";
     private const string DataRootSetting = "KeyLoad:DataRoot";
@@ -71,7 +72,7 @@ internal static IOptions<AppHostControlOptions> Bind(IConfiguration configuratio
     private static bool ReadProcessFaults(IConfiguration configuration, AppHostControlOptions options)
     {
         var enabled = configuration.GetValue<bool>(ProcessFaultCommandProtocol.EnabledSetting);
-        var selectedLoad = options.BenchmarksEnabled && configuration[BenchmarkTargetSetting] == "KeyLoad"
+        var selectedLoad = options.BenchmarksEnabled && configuration[BenchmarkTargetSetting] == KeyLoadTarget
             && configuration.GetValue<bool>(BenchmarkSessionSetting)
             && configuration[BenchmarkDataRootSetting] is { } root && Path.IsPathFullyQualified(root);
         if (enabled && (!options.Ephemeral || !selectedLoad && (options.BenchmarksEnabled || options.TargetSelected

src/KeyLoad.Server/Search/Execution/FullText/Incremental/NativeTextIncrementalBegin.cs

@@ -8,14 +8,15 @@ namespace KeyLoad.Server.Features.Search;
 
 internal static class NativeTextIncrementalBegin
 {
+    private const int OriginalFailureOnlyCount = 1;
     internal static TextMaintenanceCapabilityResult Run(DatabaseEngine database, string root, PrincipalRecord principal,
         TextMaintenanceCapabilityRequest request, ServerRuntimeOptions configured, TimeProvider clock,
-        CancellationToken shutdown, NativeTextIncrementalSessions sessions, Action<NativeTextFaultStage>? faultObserver,
+        CancellationTokenSource shutdown, NativeTextIncrementalSessions sessions, Action<NativeTextFaultStage>? faultObserver,
         NativeTextResourceOwnership? resources, Action<string, NativeTextResourceReservation> retainGeneration,
         DateTimeOffset? expiry, IResourceLifetimeWorkBinding? binding, CancellationToken token)
     {
         var work = new NativeMaintenanceSessionWork(configured.Core.DatabaseLimits, clock,
-            expiry ?? clock.GetUtcNow().AddSeconds(configured.Core.DatabaseLimits.Value.QueryDeadlineSeconds), shutdown);
+            expiry ?? clock.GetUtcNow().AddSeconds(configured.Core.DatabaseLimits.Value.QueryDeadlineSeconds), shutdown.Token);
         var captureStarted = false;
         var captured = false;
         try
@@ -95,7 +96,7 @@ void CaptureOwnership(NativeTextIncrementalSession actual)
             {
                 ServerFailureObserver.Observe(reservation.CompleteAfterJoinedCleanup, failures);
             }
-            if (captured is null && failures.Count == 1)
+            if (captured is null && failures.Count == OriginalFailureOnlyCount)
             {
                 ServerFailureObserver.Observe(work.DisposeAfterStage, failures);
             }

src/KeyLoad.Server/Search/Execution/FullText/Incremental/NativeTextIncrementalMaintenanceService.cs

@@ -101,7 +101,7 @@ private TextMaintenanceCapabilityResult ExecuteExclusive(PrincipalRecord princip
         if (request.Kind == TextMaintenanceCapabilityKind.Begin)
         {
             return NativeTextIncrementalBegin.Run(database, root, principal, request, configured, clock,
-                shutdown.Token, sessions, faultObserver, resources, generations.RetainGeneration, originalExpiry, binding, token);
+                shutdown, sessions, faultObserver, resources, generations.RetainGeneration, originalExpiry, binding, token);
         }
         var retained = sessions.Require(request.SessionId);
         retained.Work.RequireExpiry(originalExpiry);
@@ -161,7 +161,7 @@ private async Task DisposeCoreAsync()
         var failures = new List<Exception>();
         await ServerFailureObserver.ObserveAsync(shutdown.CancelAsync, failures).ConfigureAwait(false);
         await ServerFailureObserver.ObserveAsync(() => NativeTextIncrementalOwnerShutdown.RunAsync(
-            sessions, selectedAdmission, worker, selectedGate, selectedSlots), failures).ConfigureAwait(false);
+            sessions, selectedAdmission, worker.DisposeAsync().AsTask(), selectedGate, selectedSlots), failures).ConfigureAwait(false);
         ServerFailureObserver.ThrowIfAny(failures);
         shutdown.Dispose();
     }

src/KeyLoad.Server/Search/Execution/FullText/Online/NativeTextOnlineMaintenanceService.cs

@@ -68,13 +68,15 @@ private async Task<OnlineTextCapabilityResult> ExecuteOwnedAsync(PrincipalRecord
             throw NativeTextErrors.Ownership();
         }
         var session = request.Kind == OnlineTextCapabilityKind.ResolveOriginal ? null : sessions.Require(request.SessionId);
-        var work = session?.Work ?? new NativeMaintenanceSessionWork(options.Core.DatabaseLimits, clock, expiry, shutdown.Token);
-        var retained = session is not null;
+        NativeMaintenanceSessionWork? created = null;
         try
         {
-            if (!retained)
+            var work = session?.Work;
+            if (work is null)
             {
-                sessions.BeginWork(request.SessionId, work);
+                created = new NativeMaintenanceSessionWork(options.Core.DatabaseLimits, clock, expiry, shutdown.Token);
+                work = created;
+                sessions.BeginWork(request.SessionId, created);
             }
             using var stage = work.EnterStage(binding, token);
             work.RequireExpiry(expiry);
@@ -87,17 +89,17 @@ private async Task<OnlineTextCapabilityResult> ExecuteOwnedAsync(PrincipalRecord
                 }
                 session = await sessions.AddOwnedAsync(new(request.SessionId, request.Request, principal.Id, expiry, work))
                     .ConfigureAwait(false);
-                retained = true;
+                created = null;
             }
             return await ExecuteStageAsync(session ?? throw NativeTextErrors.Ownership(), principal, request, expiry)
                 .ConfigureAwait(false);
         }
         finally
         {
-            if (!retained)
+            if (created is not null)
             {
-                work.DisposeAfterStage();
-                sessions.RemoveWorkWithoutSession(request.SessionId, work);
+                created.Dispose();
+                sessions.RemoveWorkWithoutSession(request.SessionId, created);
             }
         }
     }

src/KeyLoad.Server/Search/Lifecycle/Ann/NativeAnnMaintenanceAdmissions.cs

@@ -1,5 +1,4 @@
 using KeyLoad.Clustering;
-using KeyLoad.Core;
 
 namespace KeyLoad.Server.Features.Search;
 

src/KeyLoad.Server/Search/Lifecycle/FullText/Incremental/NativeTextIncrementalOwnerShutdown.cs

@@ -3,14 +3,14 @@ namespace KeyLoad.Server.Features.Search;
 internal static class NativeTextIncrementalOwnerShutdown
 {
     internal static async Task RunAsync(NativeTextIncrementalSessions sessions, NativeTextSelectedReadAdmission selectedAdmission,
-        NativeTextIncrementalWorker worker, Lock selectedGate, Dictionary<string, NativeTextSelectedIndexSlot> selectedSlots)
+        Task originalWorker, Lock selectedGate, Dictionary<string, NativeTextSelectedIndexSlot> selectedSlots)
     {
         sessions.CloseAdmission();
         var originalReaders = selectedAdmission.CloseAndJoinReaders();
         var failures = new List<Exception>();
         try
         {
-            await worker.DisposeAsync().ConfigureAwait(false);
+            await originalWorker.ConfigureAwait(false);
         }
         catch (Exception error) when (ManagedCode.Communication.CQRS.CqrsRuntimeFailures.FindFatal(error) is null)
         {