RunServer's Runqueue should have a single Run Queue instead of one per client

src/XTMF2.Client/Program.cs

@@ -122,7 +122,12 @@ static void Main(string[] args)
                                 Console.WriteLine("Unable to create a connection to the host!");
                                 return;
                             }
-                            RunClient(serverStream, dllsToLoad);
+                            using (var remoteRunRegistry = new RemoteRunRegistry())
+                            {
+                                RunClient(serverStream, dllsToLoad, config: null,
+                                    usePrivateWorkspace: false, remoteEstimationRegistry: null,
+                                    remoteRunRegistry: remoteRunRegistry, allowDeployment: false);
+                            }
                         }
                         finally
                         {
@@ -250,10 +255,9 @@ private static void RunTcpClientUnsecured(TcpClient client, List<string> extraDl
             {
                 try
                 {
-                    RunClient(stream, extraDlls, usePrivateWorkspace: true,
-                        allowDeployment: false,
+                    RunClient(stream, extraDlls, config: null, usePrivateWorkspace: true,
                         remoteEstimationRegistry: remoteEstimationRegistry,
-                        remoteRunRegistry: remoteRunRegistry);
+                        remoteRunRegistry: remoteRunRegistry, allowDeployment: false);
                 }
                 catch (Exception ex)
                 {
@@ -282,10 +286,9 @@ private static void RunTcpClient(TcpClient client, X509Certificate2 certificate,
                 {
                     try
                     {
-                        RunClient(stream, extraDlls, usePrivateWorkspace: true,
-                            allowDeployment: true,
+                        RunClient(stream, extraDlls, config: null, usePrivateWorkspace: true,
                             remoteEstimationRegistry: remoteEstimationRegistry,
-                            remoteRunRegistry: remoteRunRegistry);
+                            remoteRunRegistry: remoteRunRegistry, allowDeployment: true);
                     }
                     catch (Exception ex)
                     {
@@ -298,9 +301,9 @@ private static void RunTcpClient(TcpClient client, X509Certificate2 certificate,
             }
         }
 
-        private static void RunClient(Stream serverStream, List<string> extraDlls, SystemConfiguration? config = null,
-            bool usePrivateWorkspace = false, RemoteSharedEstimationRegistry? remoteEstimationRegistry = null,
-            RemoteRunRegistry? remoteRunRegistry = null, bool allowDeployment = false)
+        private static void RunClient(Stream serverStream, List<string> extraDlls, SystemConfiguration? config,
+            bool usePrivateWorkspace, RemoteSharedEstimationRegistry? remoteEstimationRegistry,
+            RemoteRunRegistry remoteRunRegistry, bool allowDeployment)
         {
             var runtimeStopwatch = Stopwatch.StartNew();
             LogStartup("Initializing the XTMF runtime for a client connection.");
@@ -325,16 +328,16 @@ private static void RunClient(Stream serverStream, List<string> extraDlls, Syste
             clientBus.SetDeploymentGate(
                 () =>
                 {
-                    remoteRunRegistry?.BeginDrain();
+                    remoteRunRegistry.BeginDrain();
                     registry.BeginDrain();
                 },
                 () =>
                 {
-                    remoteRunRegistry?.EndDrain();
+                    remoteRunRegistry.EndDrain();
                     registry.EndDrain();
                 },
                 () =>
-                    (remoteRunRegistry?.IsIdle ?? true) && registry.IsIdle,
+                    remoteRunRegistry.IsIdle && registry.IsIdle,
                 RestartWithOriginalArguments);
             clientBus.SetSharedActivityProviders(registry.GetActiveActivities,
                 sharedEstimationWorker.GetActiveActivities);
@@ -344,7 +347,7 @@ private static void RunClient(Stream serverStream, List<string> extraDlls, Syste
             }
             finally
             {
-                remoteRunRegistry?.Detach(clientBus);
+                remoteRunRegistry.Detach(clientBus);
             }
         }
 

src/XTMF2/Bus/RemoteRunRegistry.cs

@@ -59,14 +59,27 @@ public sealed class RemoteRunRegistry : IDisposable
     private readonly string _storageDirectory;
     private bool _disposed;
 
+    /// <summary>
+    /// Gets whether the RunServer scheduler has no active or queued work.
+    /// </summary>
     public bool IsIdle => _scheduler.IsIdle;
 
+    /// <summary>
+    /// Stops the registry from accepting new work while existing work drains.
+    /// </summary>
     public void BeginDrain()
         => _scheduler.BeginDrain();
 
+    /// <summary>
+    /// Allows the registry to accept work after a drain has been cancelled.
+    /// </summary>
     public void EndDrain()
         => _scheduler.EndDrain();
 
+    /// <summary>
+    /// Creates a registry that owns remote run state and the process-wide RunServer scheduler.
+    /// </summary>
+    /// <param name="storageDirectory">The directory for persisted run manifests and artifacts; when <see langword="null"/>, the default per-user XTMF2 directory is used.</param>
     public RemoteRunRegistry(string? storageDirectory = null)
     {
         _storageDirectory = storageDirectory ?? Path.Combine(
@@ -76,6 +89,20 @@ public RemoteRunRegistry(string? storageDirectory = null)
         LoadPersistedJobs();
     }
 
+    /// <summary>
+    /// Registers a remote run and queues it on this registry's scheduler.
+    /// </summary>
+    /// <param name="observer">The connected RunServer bus that receives run output and completion notifications.</param>
+    /// <param name="context">The prepared run context to execute.</param>
+    /// <param name="runName">The display name of the run.</param>
+    /// <param name="runMode">The execution mode of the run.</param>
+    /// <param name="workingDirectory">The run's private working directory.</param>
+    /// <param name="startToExecute">The model-system start point to execute.</param>
+    /// <param name="modelSystem">The serialized model system, used to identify the submitted run.</param>
+    /// <param name="projectId">The identifier of the owning project, if known.</param>
+    /// <param name="modelSystemId">The identifier of the model system, if known.</param>
+    /// <param name="ownerUserId">The identifier of the owning user, if known.</param>
+    /// <returns><see langword="true"/> if the run was registered and queued; otherwise, <see langword="false"/> if the registry is draining or the run ID is already registered.</returns>
     public bool Submit(RunServerBus observer, RunContext context, string runName,
         RunMode runMode, string workingDirectory, string startToExecute, byte[] modelSystem,
         Guid? projectId = null, Guid? modelSystemId = null, Guid? ownerUserId = null)
@@ -107,6 +134,10 @@ public bool Submit(RunServerBus observer, RunContext context, string runName,
     internal Scheduler.ReservationLease ReserveWorkerSlot(RunContext context, int expectedSlotCount)
         => _scheduler.ReserveGroup(context.ID, context, expectedSlotCount);
 
+    /// <summary>
+    /// Attaches a connected bus to retained remote runs so it can receive their subsequent updates.
+    /// </summary>
+    /// <param name="observer">The bus to attach as a run observer.</param>
     public void Attach(RunServerBus observer)
     {
         lock (_sync)
@@ -116,6 +147,10 @@ public void Attach(RunServerBus observer)
         }
     }
 
+    /// <summary>
+    /// Detaches a bus from retained remote runs without cancelling the runs.
+    /// </summary>
+    /// <param name="observer">The bus to detach from run notifications.</param>
     public void Detach(RunServerBus observer)
     {
         lock (_sync)
@@ -125,12 +160,20 @@ public void Detach(RunServerBus observer)
         }
     }
 
+    /// <summary>
+    /// Gets snapshots of all retained remote runs.
+    /// </summary>
+    /// <returns>A snapshot of each run currently retained by the registry.</returns>
     public IReadOnlyList<RemoteRunSnapshot> GetSnapshots()
     {
         lock (_sync)
             return _jobs.Values.Select(job => job.GetSnapshot()).ToArray();
     }
 
+    /// <summary>
+    /// Gets the active and queued run and worker-reservation activities in the shared scheduler.
+    /// </summary>
+    /// <returns>The scheduler's current activity inventory.</returns>
     public IReadOnlyList<RunServerActivity> GetActiveActivities()
     {
         lock (_sync)
@@ -154,6 +197,12 @@ public IReadOnlyList<RunServerActivity> GetActiveActivities()
         }
     }
 
+    /// <summary>
+    /// Reads a run's retained artifact archive, rebuilding it from the private run directory when necessary.
+    /// </summary>
+    /// <param name="runId">The identifier of the run whose artifacts are requested.</param>
+    /// <param name="archive">Receives the archive bytes when available; otherwise, <see langword="null"/>.</param>
+    /// <returns><see langword="true"/> if the artifact archive was read successfully; otherwise, <see langword="false"/>.</returns>
     public bool TryReadArtifacts(string runId, out byte[]? archive)
     {
         RemoteRunJob? job;
@@ -197,6 +246,11 @@ public bool TryReadArtifacts(string runId, out byte[]? archive)
         }
     }
 
+    /// <summary>
+    /// Acknowledges receipt of a completed run's artifacts and removes the retained archive.
+    /// </summary>
+    /// <param name="runId">The identifier of the run whose artifact receipt is acknowledged.</param>
+    /// <returns><see langword="true"/> if the run exists and is no longer running; otherwise, <see langword="false"/>.</returns>
     public bool AcknowledgeReceived(string runId)
     {
         RemoteRunJob? job;
@@ -237,6 +291,12 @@ private static bool IsPrivateRunDirectory(string runId, string workingDirectory)
         }
     }
 
+    /// <summary>
+    /// Deletes a completed or interrupted run and its files from the registry's private workspace.
+    /// </summary>
+    /// <param name="runId">The identifier of the run to delete.</param>
+    /// <param name="error">Receives an explanation if deletion fails; otherwise, <see langword="null"/>.</param>
+    /// <returns><see langword="true"/> if the run and its files were deleted; otherwise, <see langword="false"/>.</returns>
     public bool DeleteRun(string runId, out string? error)
     {
         error = null;
@@ -288,6 +348,10 @@ public bool DeleteRun(string runId, out string? error)
         }
     }
 
+    /// <summary>
+    /// Requests cancellation of a queued or running remote run.
+    /// </summary>
+    /// <param name="runId">The identifier of the run to cancel.</param>
     public void Cancel(string runId)
     {
         lock (_sync)
@@ -297,9 +361,17 @@ public void Cancel(string runId)
         }
     }
 
+    /// <summary>
+    /// Kills a queued or running remote run, removing queued work or terminating the active run as applicable.
+    /// </summary>
+    /// <param name="runId">The identifier of the run to kill.</param>
+    /// <returns><see langword="true"/> if a matching run was found and handled; otherwise, <see langword="false"/>.</returns>
     public bool Kill(string runId)
         => _scheduler.Kill(runId);
 
+    /// <summary>
+    /// Stops the scheduler and releases resources owned by this registry.
+    /// </summary>
     public void Dispose()
     {
         lock (_sync)

src/XTMF2/Bus/RunServerBus.cs

@@ -108,6 +108,8 @@ public RunServerBus(Stream serverStream, bool streamOwner, XTMFRuntime runtime,
             bool allowDeployment = false)
         {
             Runtime = runtime;
+            // Standalone debug/test buses own their scheduler. Production RunClient sessions
+            // receive a process-owned registry so independent host connections share one queue.
             _remoteRunRegistry = remoteRunRegistry;
             _runScheduler = remoteRunRegistry is null ? new Scheduler(this, runLocal) : null;
             _clientHost = serverStream;

tests/XTMF2.UnitTests/Bus/TestRemoteRunRegistry.cs

@@ -1,17 +1,91 @@
 using System;
 using System.IO;
+using System.Linq;
 using System.Security.Cryptography;
 using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization;
+using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using XTMF2.Bus;
+using XTMF2.Bus.Optimization;
 
 namespace XTMF2.UnitTests.Bus;
 
 [TestClass]
 public class TestRemoteRunRegistry
 {
+    [TestMethod]
+    public async Task SharedWorkerReservation_BlocksOrdinaryRunsFromIndependentObservers()
+    {
+        var directory = Path.Combine(Path.GetTempPath(), "XTMF2-RemoteRunRegistry-" + Guid.NewGuid().ToString("N"));
+        Directory.CreateDirectory(directory);
+        try
+        {
+            using var registry = new RemoteRunRegistry(directory);
+            using var firstObserver = CreateObserver();
+            using var secondObserver = CreateObserver();
+            Assert.IsTrue(RunContext.CreateRunContext(null!, "worker-run", [], Path.GetTempPath(),
+                "Start", RunMode.Normal, out var workerContext));
+            Assert.IsTrue(RunContext.CreateRunContext(null!, "first-ordinary-run", [], Path.GetTempPath(),
+                "Start", RunMode.Normal, out var firstRun));
+            Assert.IsTrue(RunContext.CreateRunContext(null!, "second-ordinary-run", [], Path.GetTempPath(),
+                "Start", RunMode.Normal, out var secondRun));
+
+            using var workerLease = registry.ReserveWorkerSlot(workerContext, 1);
+            await workerLease.Started.WaitAsync(TimeSpan.FromSeconds(3));
+            Assert.IsTrue(registry.Submit(firstObserver, firstRun, "first", RunMode.Normal,
+                firstRun.WorkingDirectory, firstRun.StartToExecute, []));
+            Assert.IsTrue(registry.Submit(secondObserver, secondRun, "second", RunMode.Normal,
+                secondRun.WorkingDirectory, secondRun.StartToExecute, []));
+
+            var activities = registry.GetActiveActivities();
+            Assert.HasCount(3, activities);
+            Assert.AreEqual("worker-run", activities[0].RunId);
+            Assert.AreEqual(RunServerActivityState.Running, activities[0].State);
+            CollectionAssert.AreEqual(new[] { "first-ordinary-run", "second-ordinary-run" },
+                activities.Skip(1).Select(activity => activity.RunId).ToArray());
+            Assert.IsTrue(activities.Skip(1).All(activity => activity.State == RunServerActivityState.Queued));
+
+            registry.Dispose();
+            workerLease.Dispose();
+        }
+        finally
+        {
+            Directory.Delete(directory, recursive: true);
+        }
+    }
+
+    [TestMethod]
+    public async Task SeparateRegistries_HaveIndependentExecutionSlots()
+    {
+        var firstDirectory = Path.Combine(Path.GetTempPath(), "XTMF2-RemoteRunRegistry-" + Guid.NewGuid().ToString("N"));
+        var secondDirectory = Path.Combine(Path.GetTempPath(), "XTMF2-RemoteRunRegistry-" + Guid.NewGuid().ToString("N"));
+        Directory.CreateDirectory(firstDirectory);
+        Directory.CreateDirectory(secondDirectory);
+        try
+        {
+            using var firstRegistry = new RemoteRunRegistry(firstDirectory);
+            using var secondRegistry = new RemoteRunRegistry(secondDirectory);
+            Assert.IsTrue(RunContext.CreateRunContext(null!, "first-process-run", [], Path.GetTempPath(),
+                "Start", RunMode.Normal, out var firstContext));
+            Assert.IsTrue(RunContext.CreateRunContext(null!, "second-process-run", [], Path.GetTempPath(),
+                "Start", RunMode.Normal, out var secondContext));
+
+            using var firstLease = firstRegistry.ReserveWorkerSlot(firstContext, 1);
+            using var secondLease = secondRegistry.ReserveWorkerSlot(secondContext, 1);
+            await Task.WhenAll(firstLease.Started, secondLease.Started).WaitAsync(TimeSpan.FromSeconds(3));
+
+            Assert.AreEqual(RunServerActivityState.Running, firstRegistry.GetActiveActivities().Single().State);
+            Assert.AreEqual(RunServerActivityState.Running, secondRegistry.GetActiveActivities().Single().State);
+        }
+        finally
+        {
+            Directory.Delete(firstDirectory, recursive: true);
+            Directory.Delete(secondDirectory, recursive: true);
+        }
+    }
+
     [TestMethod]
     public void RegistryRestart_MarksPersistedActiveRunInterrupted()
     {
@@ -47,6 +121,10 @@ public void RegistryRestart_MarksPersistedActiveRunInterrupted()
         }
     }
 
+    private static RunServerBus CreateObserver()
+        => new(new MemoryStream(), true, runtime: null!, remoteRunRegistry: null,
+            allowDeployment: false);
+
     [TestMethod]
     public void DeleteRun_RemovesWorkspaceArchiveAndManifest()
     {