From 26f9d528c65daf8cc164ea96c2cab2d077a6fbb1 Mon Sep 17 00:00:00 2001 From: Nick Nassiri Date: Fri, 3 Jul 2026 23:44:21 -0700 Subject: [PATCH] fix(cluster): synchronize shared-listener start-once to stop worker-listen race Root-cause a flaky 60s CI timeout in the cluster port-sharing tests (RoundRobin_DistributesConnectionsAcrossWorkers, Fork_WorkersShareNetPort, Fork_WorkersShareHttpPort), Compiled mode. It was not test flakiness but a real data race in SharedListenerRegistry. SharedTcpListener.AddWorker / SharedHttpListener.AddWorker started the underlying listener with an unsynchronized check-then-act: if (!_started) { _started = true; _listener.Start(); StartAccepting(); } Under SCHED_RR both cluster workers call server.listen() on the same port from their own interpreter threads, and ConcurrentDictionary.GetOrAdd hands both the SAME SharedTcpListener. When the two AddWorker calls overlap, both pass the `!_started` check and both call _listener.Start(); the second Bind on the same socket throws. That exception propagates out of ListenAsClusterWorker before it fires the `listening` callback / process.send('ready'), so the primary's readyCount never reaches 2, the connections are never made, and the module hangs to the 60s timeout. Multi-core + CI load widens the window, so it surfaced intermittently and as a timeout (not an assertion failure). Wrap the start-once transition in a lock (double-checked, publishing `_started` only after Start() succeeds so a genuinely failed first Start can be retried) in both shared listeners, and guard Stop() with the same lock. Reproduced locally on the pre-fix build: Fork_WorkersShareNetPort(Compiled) hung 61s on the first loop iteration. Post-fix: 40/40 iterations of the three port-sharing tests pass with no hang; full ClusterModule/Net/Http suite 130/0. --- Runtime/Types/SharedListenerRegistry.cs | 47 +++++++++++++++++++------ 1 file changed, 36 insertions(+), 11 deletions(-) diff --git a/Runtime/Types/SharedListenerRegistry.cs b/Runtime/Types/SharedListenerRegistry.cs index 1ee354b4..4ac173ce 100644 --- a/Runtime/Types/SharedListenerRegistry.cs +++ b/Runtime/Types/SharedListenerRegistry.cs @@ -126,6 +126,7 @@ internal class SharedTcpListener { private readonly TcpListener _listener; private readonly ConcurrentDictionary _workers = new(); + private readonly object _startGate = new(); private int _roundRobinIndex; private CancellationTokenSource? _cts; private bool _started; @@ -140,12 +141,22 @@ public SharedTcpListener(int port, IPAddress address) public void AddWorker(double workerId, Action onConnection, Interp workerInterpreter) { _workers[workerId] = new WorkerRegistration(workerId, onConnection, workerInterpreter); - if (!_started) + + // Start the underlying listener exactly once. Two cluster workers calling server.listen() on + // the same port register concurrently on their own interpreter threads and are handed the SAME + // SharedTcpListener by GetOrAdd, so this start-once transition MUST be synchronized. Without the + // lock both workers can pass the `!_started` check, both call _listener.Start(), and the second + // Bind on the same socket throws — the throwing worker's ListenAsClusterWorker then never fires + // its `listening` callback / 'ready' message, deadlocking a round-robin test until its timeout. + // Publish `_started` only after Start succeeds so a genuinely failed first Start can be retried. + if (Volatile.Read(ref _started)) return; + lock (_startGate) { - _started = true; + if (_started) return; _cts = new CancellationTokenSource(); _listener.Start(); StartAccepting(); + Volatile.Write(ref _started, true); } } @@ -156,9 +167,12 @@ public void RemoveWorker(double workerId) public void Stop() { - _cts?.Cancel(); - try { _listener.Stop(); } catch { } - _started = false; + lock (_startGate) + { + _cts?.Cancel(); + try { _listener.Stop(); } catch { } + Volatile.Write(ref _started, false); + } } private void StartAccepting() @@ -207,6 +221,7 @@ internal class SharedHttpListener { private HttpListener? _listener; private readonly ConcurrentDictionary _workers = new(); + private readonly object _startGate = new(); private int _roundRobinIndex; private CancellationTokenSource? _cts; private bool _started; @@ -222,9 +237,15 @@ public SharedHttpListener(int port) public void AddWorker(double workerId, Action onRequest, Interp workerInterpreter) { _workers[workerId] = new HttpWorkerRegistration(workerId, onRequest, workerInterpreter); - if (!_started) + + // Start the underlying HttpListener exactly once — see SharedTcpListener.AddWorker for the race: + // two workers sharing a port register concurrently on the same SharedHttpListener, so the + // start-once transition must be synchronized or the second Start() throws and its worker never + // reports `listening`/`ready`, hanging the test. Publish `_started` only after Start succeeds. + if (Volatile.Read(ref _started)) return; + lock (_startGate) { - _started = true; + if (_started) return; _cts = new CancellationTokenSource(); _listener = new HttpListener(); _listener.Prefixes.Add($"http://+:{_port}/"); @@ -239,6 +260,7 @@ public void AddWorker(double workerId, Action onRequest, In _listener.Start(); } StartAccepting(); + Volatile.Write(ref _started, true); } } @@ -249,10 +271,13 @@ public void RemoveWorker(double workerId) public void Stop() { - _cts?.Cancel(); - try { _listener?.Stop(); } catch { } - try { _listener?.Close(); } catch { } - _started = false; + lock (_startGate) + { + _cts?.Cancel(); + try { _listener?.Stop(); } catch { } + try { _listener?.Close(); } catch { } + Volatile.Write(ref _started, false); + } } private void StartAccepting()