Testing: a store's port is held machine-wide; its container is acted on by id, never by name; another live process's container is refused (cross-process store race)

src/MicroPlumberd.Testing.StoreHolder/MicroPlumberd.Testing.StoreHolder.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <!-- A child process for StorePortRaceTests: two of these, started at one barrier instant, each hold a store and assert it
+       stays THEIRS. Not packed. -->
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\MicroPlumberd.Testing\MicroPlumberd.Testing.csproj" />
+  </ItemGroup>
+</Project>

src/MicroPlumberd.Testing.StoreHolder/Program.cs

@@ -0,0 +1,55 @@
+// Holds one EventStoreServer for <hold> seconds, starting it at <startAtUtcTicks> (a barrier shared with a sibling process),
+// and checks every 500 ms that the store is still THIS process's: its own container (by the id seen right after start) is
+// running, and its own marker event is still readable (a restart of an in-memory store wipes it).
+// Prints one line: "PORT=<p> CONTAINER=<id> RESULT=SURVIVED" or "... RESULT=LOST:<why>"; exit 0 only when it survived.
+using Docker.DotNet;
+using Docker.DotNet.Models;
+using KurrentDB.Client;
+using MicroPlumberd.Testing;
+
+var startAt = new DateTime(long.Parse(args[0]), DateTimeKind.Utc);
+var hold = TimeSpan.FromSeconds(int.Parse(args[1]));
+var delay = startAt - DateTime.UtcNow;
+if (delay > TimeSpan.Zero) await Task.Delay(delay);
+
+using var docker = new DockerClientConfiguration().CreateClient();
+string? id = null;
+string result;
+var server = new EventStoreServer();
+try
+{
+    await server.StartInDocker();
+    id = (await docker.Containers.ListContainersAsync(new ContainersListParameters { All = true }))
+        .FirstOrDefault(c => c.Names.Contains("/" + server.ContainerName))?.ID;
+    var client = new KurrentDBClient(server.GetEventStoreSettings());
+    var mine = $"holder-{Environment.ProcessId}";
+    await client.AppendToStreamAsync(mine, StreamState.Any,
+        [new EventData(Uuid.NewUuid(), "Marker", "{}"u8.ToArray())]);
+
+    result = "SURVIVED";
+    var until = DateTime.UtcNow + hold;
+    while (DateTime.UtcNow < until)
+    {
+        await Task.Delay(500);
+        var running = id != null && (await docker.Containers.ListContainersAsync(new ContainersListParameters { All = true }))
+            .Any(c => c.ID == id && c.State == "running");
+        if (!running) { result = "LOST:my-container-is-gone-or-stopped"; break; }
+        try
+        {
+            var read = client.ReadStreamAsync(Direction.Forwards, mine, StreamPosition.Start);
+            if (await read.ReadState == ReadState.StreamNotFound) { result = "LOST:my-marker-is-gone(store-restarted)"; break; }
+        }
+        catch (Exception e) { result = $"LOST:store-unreachable({e.GetType().Name})"; break; }
+    }
+}
+catch (Exception e)
+{
+    result = $"LOST:start-failed({e.GetType().Name}: {e.Message.Split('\n')[0]})";
+}
+finally
+{
+    await server.DisposeAsync();
+}
+
+Console.WriteLine($"PORT={server.HttpPort} CONTAINER={id?[..12] ?? "none"} RESULT={result}");
+return result == "SURVIVED" ? 0 : 1;

src/MicroPlumberd.Testing/EventStoreServer.cs

@@ -37,8 +37,15 @@ public class EventStoreServer :  IDisposable, IAsyncDisposable
     public static TimeSpan ReadyBudget(TimeSpan perStore, int startingAtOnce) => perStore * Math.Max(1, startingAtOnce);
 
     /// <summary>The readiness wait itself — a seam so a test can see the budget each start actually waited with.</summary>
-    internal static Func<KurrentDBClientSettings, TimeSpan, Task> WaitReady { get; set; } =
-        (settings, budget) => settings.WaitUntilReady(budget);
+    /// <summary>
+    /// The removal Cleanup makes — a seam so a test can make it fail (as a concurrent removal's 409 did). It is given the
+    /// server, so a test touches only its own store while the suite runs classes in parallel.
+    /// </summary>
+    internal static Func<EventStoreServer, DockerClient, string, Task> RemoveContainer { get; set; } =
+        (_, docker, id) => docker.Containers.RemoveContainerAsync(id, new ContainerRemoveParameters { Force = true, RemoveVolumes = true });
+
+    internal static Func<EventStoreServer, KurrentDBClientSettings, TimeSpan, Task> WaitReady { get; set; } =
+        (_, settings, budget) => settings.WaitUntilReady(budget);
     private static readonly ConcurrentDictionary<EventStoreServer, byte> Undisposed = new();
     private static readonly string CurrentOwner = OwnerOf(Environment.ProcessId, Process.GetCurrentProcess().StartTime);
 
@@ -56,14 +63,20 @@ internal EventStoreServer(string? containerName = null)
     {
         if (containerName != null)
             _containerName = containerName;
-        httpPort = _searcher.FindNextAvailablePort();
+        (httpPort, _portLock) = PortLock.Acquire(_searcher);
         const string eventStoreHostName = "127.0.0.1";
         HttpUrl = new Uri($"esdb://admin:changeit@{eventStoreHostName}:{httpPort}?tls=false&tlsVerifyCert=false");
         client = new DockerClientConfiguration()
             .CreateClient();
     }
 
     private string? _containerName;
+
+    // card microplumberd-testing-store-port-race: this store's port, held machine-wide for its lifetime (see PortLock), and
+    // the id of the container IT created. Removal and restart go by that id — never by name: the name is derived from the
+    // port, so another process's store can carry it.
+    private PortLock? _portLock;
+    private string? _containerId;
     public string ContainerName => _containerName ?? $"eventstore-mem-{httpPort}";
     public int HttpPort => httpPort;
 
@@ -86,10 +99,10 @@ internal EventStoreServer(string? containerName = null)
 
     public async Task<bool> Stop()
     {
-        var container = await GetEventStoreContainer();
-        if (container != null)
+        // This server's own container only (by id) — a name can be another process's store.
+        if (_containerId != null)
         {
-            var data = await client.Containers.InspectContainerAsync(container.ID);
+            var data = await client.Containers.InspectContainerAsync(_containerId);
             if (data.State.Running)
             {
                 await client.Containers.StopContainerAsync(data.ID, new ContainerStopParameters());
@@ -102,10 +115,9 @@ public async Task<bool> Stop()
 
     public async Task Restart(TimeSpan delay)
     {
-        var container = await GetEventStoreContainer();
-        if (container != null)
+        if (_containerId != null)
         {
-            var data = await client.Containers.InspectContainerAsync(container.ID);
+            var data = await client.Containers.InspectContainerAsync(_containerId);
             if (data.State.Running)
             {
                 await client.Containers.StopContainerAsync(data.ID, new ContainerStopParameters());
@@ -139,10 +151,20 @@ private async Task<EventStoreServer> StartAndWait(bool wait, bool inMemory, Time
         await ReapOrphans(client);
 
         var container = await GetEventStoreContainer();
+        if (container != null && container.ID != _containerId)
+        {
+            // A container with this name that THIS server did not create: another live process's store (refused, named —
+            // restarting it would wipe it, removing it would pull it from under that process), or a dead owner's (reaped).
+            if (!container.Labels.TryGetValue(OwnerLabel, out var owner) || !IsDeadOwnerOnThisMachine(owner))
+                throw new InvalidOperationException(
+                    $"Container '{ContainerName}' ({container.ID[..12]}) belongs to another process ({owner ?? "no owner label"}); " +
+                    "it was neither restarted nor removed. Two processes chose the same store port.");
+            await client.Containers.RemoveContainerAsync(container.ID, new ContainerRemoveParameters { Force = true, RemoveVolumes = true });
+            container = null;
+        }
         if (!inMemory && container != null)
         {
-            var data = await client.Containers.InspectContainerAsync(container.ID);
-            await client.Containers.RemoveContainerAsync(data.ID, new ContainerRemoveParameters() { RemoveVolumes = true});
+            await client.Containers.RemoveContainerAsync(container.ID, new ContainerRemoveParameters() { RemoveVolumes = true});
             container = null;
         }
 
@@ -178,6 +200,7 @@ private async Task<EventStoreServer> StartAndWait(bool wait, bool inMemory, Time
                         { "2113", default},
                     }
                 });
+                _containerId = response.ID;
                 await client.Containers.StartContainerAsync(response.ID, new ContainerStartParameters());
             }
             else
@@ -191,12 +214,16 @@ private async Task<EventStoreServer> StartAndWait(bool wait, bool inMemory, Time
 
             // Health THEN a gRPC round trip (WaitUntilReady), on a budget that grows with the stores starting
             // alongside this one — read now, when they are all competing for the host.
-            await WaitReady(this.GetEventStoreSettings(), ReadyBudget(readyTimeout, Volatile.Read(ref _starting)));
+            await WaitReady(this, this.GetEventStoreSettings(), ReadyBudget(readyTimeout, Volatile.Read(ref _starting)));
         }
-        catch
+        catch (Exception started)
         {
-            await Cleanup();
-            throw;
+            // The START failure is the one to see (dev-clean, 2026-10-18): a cleanup that throws too (Docker 409 "removal
+            // already in progress") replaced it, and the first failure was lost. Cleanup's failure rides along, never instead.
+            try { await Cleanup(); }
+            catch (Exception cleanup) { started.Data["MicroPlumberd.Testing.CleanupFailure"] = cleanup.ToString(); }
+            System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(started).Throw();
+            throw; // unreachable
         }
 
         return this;
@@ -285,19 +312,21 @@ private static void RemoveUndisposed()
 
     async ValueTask Cleanup()
     {
-        var container = await GetEventStoreContainer();
-        if (container != null)
+        // ONLY the container this server created, by its id: a name lookup here removed another process's live store.
+        if (_containerId != null)
         {
             try
             {
-                await client.Containers.RemoveContainerAsync(container.ID, new ContainerRemoveParameters() { Force = true, RemoveVolumes = true });
+                await RemoveContainer(this, client, _containerId);
             }
             catch (DockerContainerNotFoundException)
             {
             }
+            _containerId = null;
         }
 
         Undisposed.TryRemove(this, out _);
+        Interlocked.Exchange(ref _portLock, null)?.Dispose();
     }
 
     public void Dispose()

src/MicroPlumberd.Testing/PortLock.cs

@@ -0,0 +1,48 @@
+using System.Net;
+using System.Net.Sockets;
+
+namespace MicroPlumberd.Testing;
+
+/// <summary>
+/// A store port held MACHINE-WIDE for the store's lifetime (card microplumberd-testing-store-port-race). Every process's
+/// PortSearcher starts at the same port, and a free-port probe only sees a port as taken once its container has STARTED —
+/// seconds later — so two processes chose the same port. A lock file per port, opened exclusively (on Unix .NET takes an
+/// advisory flock for FileShare.None), is held until the store is disposed and released by the OS if the process dies.
+/// </summary>
+internal sealed class PortLock : IDisposable
+{
+    /// <summary>One directory for every process on the machine (not the per-user/per-process TMPDIR).</summary>
+    internal static string Directory { get; set; } = OperatingSystem.IsWindows()
+        ? Path.Combine(Environment.GetFolderPath(Environment.SpecialFolder.CommonApplicationData), "microplumberd-testing-ports")
+        : "/tmp/microplumberd-testing-ports";
+
+    private readonly FileStream _held;
+    private PortLock(FileStream held) => _held = held;
+
+    /// <summary>The next port that no other process holds AND that is free to bind, with its lock held.</summary>
+    internal static (int Port, PortLock Lock) Acquire(PortSearcher searcher)
+    {
+        System.IO.Directory.CreateDirectory(Directory);
+        while (true)
+        {
+            var port = searcher.FindNextAvailablePort();
+            var held = TryHold(port);
+            if (held != null) return (port, held);
+        }
+    }
+
+    internal static PortLock? TryHold(int port)
+    {
+        try
+        {
+            var stream = new FileStream(Path.Combine(Directory, $"{port}.lock"), FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
+            return new PortLock(stream);
+        }
+        catch (IOException)
+        {
+            return null; // another process holds this port
+        }
+    }
+
+    public void Dispose() => _held.Dispose();
+}

src/MicroPlumberd.Tests/Integration/EventStoreServerForeignContainerTests.cs

@@ -0,0 +1,71 @@
+using System.Diagnostics;
+using Docker.DotNet;
+using Docker.DotNet.Models;
+using FluentAssertions;
+using MicroPlumberd.Testing;
+using MicroPlumberd.Tests.Utils;
+
+namespace MicroPlumberd.Tests.Integration;
+
+/// <summary>
+/// card microplumberd-testing-store-port-race: a container that carries this server's name but belongs to ANOTHER LIVE
+/// process is never restarted (that wiped its in-memory store) and never removed (that pulled it from under that process) —
+/// not at start, not at dispose. The port lock makes such a name clash rare; these rows hold the second line of defence
+/// without a race. The planted container's owner is PID 1 with its real start time: alive for as long as the machine is.
+/// </summary>
+[TestCategory("Integration")]
+public class EventStoreServerForeignContainerTests : IAsyncLifetime
+{
+    private readonly DockerClient _docker = new DockerClientConfiguration().CreateClient();
+    private readonly string _name = $"mp-foreign-{Guid.NewGuid():N}"[..24];
+    private string _planted = "";
+
+    public async Task InitializeAsync()
+    {
+        var liveForeignOwner = EventStoreServer.OwnerOf(1, Process.GetProcessById(1).StartTime);
+        _planted = (await _docker.Containers.CreateContainerAsync(new CreateContainerParameters
+        {
+            Image = "docker.kurrent.io/kurrent-latest/kurrentdb:latest",
+            Name = _name,
+            Labels = new Dictionary<string, string> { [EventStoreServer.Label] = "eventstore", [EventStoreServer.OwnerLabel] = liveForeignOwner },
+        })).ID;
+    }
+
+    public async Task DisposeAsync()
+    {
+        try { await _docker.Containers.RemoveContainerAsync(_planted, new ContainerRemoveParameters { Force = true }); }
+        catch (DockerContainerNotFoundException) { }
+        _docker.Dispose();
+    }
+
+    private async Task<ContainerInspectResponse?> Planted()
+    {
+        try { return await _docker.Containers.InspectContainerAsync(_planted); }
+        catch (DockerContainerNotFoundException) { return null; }
+    }
+
+    [Fact]
+    public async Task AStart_OverAnotherLiveProcesssContainerOfTheSameName_IsRefused_AndLeavesItAlone()
+    {
+        await using var server = new EventStoreServer(_name);
+
+        var start = await Record.ExceptionAsync(() => server.StartInDocker());
+
+        start.Should().BeOfType<InvalidOperationException>().Which.Message.Should()
+            .Contain(_name).And.Contain("another process").And.Contain("neither restarted nor removed");
+        var planted = await Planted();
+        planted.Should().NotBeNull("it was not removed");
+        planted!.State.Running.Should().BeFalse("nor (re)started");
+    }
+
+    [Fact]
+    public async Task ADispose_NeverRemovesAContainerThisServerDidNotCreate()
+    {
+        var server = new EventStoreServer(_name);
+        await Record.ExceptionAsync(() => server.StartInDocker());   // refused: it created nothing
+
+        await server.DisposeAsync();
+
+        (await Planted()).Should().NotBeNull("dispose removes only the container THIS server created, by its id");
+    }
+}

src/MicroPlumberd.Tests/Integration/EventStoreServerReadinessTests.cs

@@ -17,11 +17,16 @@ public async Task StoresStartedTogether_WaitWithABudgetThatCountsEachOther()
     {
         var budgets = new ConcurrentBag<TimeSpan>();
         var real = EventStoreServer.WaitReady;
-        EventStoreServer.WaitReady = async (settings, budget) => { budgets.Add(budget); await real(settings, budget); };
+        await using var a = EventStoreServer.Create();
+        await using var b = EventStoreServer.Create();
+        // Only these two stores' budgets: other classes start stores through the same static seam in parallel.
+        EventStoreServer.WaitReady = async (server, settings, budget) =>
+        {
+            if (ReferenceEquals(server, a) || ReferenceEquals(server, b)) budgets.Add(budget);
+            await real(server, settings, budget);
+        };
         try
         {
-            await using var a = EventStoreServer.Create();
-            await using var b = EventStoreServer.Create();
             await Task.WhenAll(a.StartInDocker(), b.StartInDocker());
         }
         finally