test(connscale): dynamic contiguous ports, and the flaky marker dropped (BACKLOG #1014)

docs/BACKLOG.md

@@ -4713,7 +4713,7 @@ Retiring the tree costs the engine nothing operationally: **`tests/test_ech_egre
 
 ## 1014. connscale smoke test's fixed 24-port block is not parallel-safe across worktrees; the flaky marker hides the collision
 
-> 🔢 **Filed 2026-08-04 — not started.** Value **5/10** · Difficulty **3/10** · _fill-in_. `test_connscale_smoke_end_to_end` hard-codes `base_port = 41000` and requires 24 **contiguous** inbound ports, so two checkouts running the suite at once contend for the same block. A `@pytest.mark.flaky` marker retries past the collision, so a determinate resource conflict wears a noise label.
+> ✅ **SHIPPED 2026-08-06 — dynamic contiguous inbound-port allocation replaces the fixed 24-port block; the flaky marker is dropped.** Value **5/10** · Difficulty **3/10** · _fill-in_. `test_connscale_smoke_end_to_end` now reserves a random contiguous inbound-port block at runtime (`_free_contiguous_ports`), asserts contiguity at acquisition, and fails loudly if no free block is found, so a genuine cross-worktree collision surfaces as a red rather than a masked retry.
 
 **Cluster:** Testing / CI reliability. **Priority:** P3. **Verdict:** build (small). **Severity:** low — it costs retries and misdiagnosis, not correctness.
 

tests/test_connscale_smoke.py

@@ -18,6 +18,7 @@
 
 from __future__ import annotations
 
+import random
 import socket
 import sys
 
@@ -28,6 +29,22 @@
 
 pytestmark = pytest.mark.timeout(120)  # the per-test 60s default is too tight for two engine spawns
 
+# The connection-count sweep; its max sets the contiguous inbound-port width (BACKLOG #1014).
+_SMOKE_COUNTS = (12, 24)
+
+# Contiguous inbound-port window for the random anchor (BACKLOG #1014). Both bounds are
+# chosen to keep the block clear of ports OTHER tests bind, so a concurrent worktree's
+# connscale block cannot land on a sibling's fixed listener:
+#   - LOWER bound sits ABOVE the sibling fixed-port MLLP band. Sibling tests bind fixed
+#     inbound ports in the 11xxx-19xxx range (e.g. 15099, 19601); anchoring at 20000+ keeps
+#     the connscale block entirely above them. [20000,30000) is empty of fixed test binds.
+#   - UPPER bound stays BELOW the OS ephemeral floors (Linux 32768+, Windows/macOS 49152+)
+#     so a kernel-assigned ephemeral port -- the sink/API ports from _free_port(), or any
+#     unrelated connection -- can never land inside the block after it is probed.
+# The upper bound is exclusive.
+_INBOUND_PORT_LO = 20000
+_INBOUND_PORT_HI = 30000
+
 
 def _free_port() -> int:
     s = socket.socket()
@@ -39,12 +56,52 @@ def _free_port() -> int:
         s.close()
 
 
+def _free_contiguous_ports(n: int, *, tries: int = 200) -> list[int]:
+    """Reserve ``n`` contiguous free inbound ports anchored at a RANDOM base.
+
+    The random anchor is the concurrency fix (BACKLOG #1014): it de-correlates worktrees so
+    two suites rarely pick overlapping blocks. The old fixed ``base_port = 41000`` guaranteed
+    a collision whenever two checkouts ran the suite at once. Probe/bind-and-release only holds
+    the block momentarily, so it cannot truly reserve it against a concurrent engine -- the
+    random anchor over a wide window is the real defense, and a genuine future collision now
+    surfaces as a RED rather than a masked retry.
+    """
+    if _INBOUND_PORT_HI - n <= _INBOUND_PORT_LO:
+        raise RuntimeError(
+            f"cannot reserve {n} contiguous ports in [{_INBOUND_PORT_LO},{_INBOUND_PORT_HI})"
+        )
+    for _ in range(tries):
+        base = random.randint(_INBOUND_PORT_LO, _INBOUND_PORT_HI - n)
+        socks: list[socket.socket] = []
+        try:
+            for i in range(n):
+                s = socket.socket()
+                # No SO_REUSEADDR on purpose: honest free-detection. A live listener must make
+                # bind FAIL here, unlike SO_REUSEADDR's Windows steal semantics. The block is
+                # released before the engine binds, so REUSEADDR would only add false-frees.
+                try:
+                    s.bind(("127.0.0.1", base + i))
+                except OSError:
+                    s.close()
+                    break
+                socks.append(s)
+            if len(socks) == n:
+                return list(range(base, base + n))
+        finally:
+            for sock in socks:
+                sock.close()
+    raise RuntimeError(
+        f"could not reserve {n} contiguous free ports in "
+        f"[{_INBOUND_PORT_LO},{_INBOUND_PORT_HI}) after {tries} tries"
+    )
+
+
 def _smoke_profile(base_port: int) -> object:
     # Small N (12 → 24) + short holds so the smoke fits the pytest budget; both sweep modes by default.
     return load_connscale_profile_text(f"""
 [connscale]
 name = "smoke-it"
-counts = [12, 24]
+counts = {list(_SMOKE_COUNTS)}
 sweep_mode = "both"
 aggregate_rate = 24.0
 per_conn_rate = 1.0
@@ -66,13 +123,18 @@ def _smoke_profile(base_port: int) -> object:
 """)
 
 
-@pytest.mark.flaky(
-    reruns=2, reruns_delay=3
-)  # CI runners are noisy (mf-ci-test-flakes): re-run clears
 async def test_connscale_smoke_end_to_end() -> None:
-    # Reserve a base inbound-port block that won't collide with the sink/API ports. The 24-conn max
-    # sweep needs 24 contiguous inbound ports; pick a high base well clear of the ephemeral churn.
-    base_port = 41000
+    # Dynamically reserve a contiguous inbound-port block (BACKLOG #1014). The sweep's max
+    # connection count needs that many contiguous inbound ports, and the engine binds
+    # base_port + i for each. A RANDOM anchor de-correlates concurrent worktrees so they no
+    # longer contend for one fixed block; contiguity is asserted at the acquisition site, and
+    # the allocator fails loudly if no free block can be found (never a silent fixed fallback).
+    # The sink/API ports stay ephemeral (above the inbound window) and won't hit the block.
+    inbound_ports = _free_contiguous_ports(max(_SMOKE_COUNTS))
+    assert inbound_ports == list(range(inbound_ports[0], inbound_ports[0] + max(_SMOKE_COUNTS))), (
+        inbound_ports
+    )
+    base_port = inbound_ports[0]
     sink_port = _free_port()
     api_port = _free_port()
     profile = _smoke_profile(base_port)
@@ -148,3 +210,29 @@ def test_fd_sampler_reads_self() -> None:
     assert live is None or live > 0  # None only if the OS tool is unavailable on this runner
     dead = FdSampler(2**31 - 1).sample()  # an implausible PID
     assert dead is None
+
+
+def test_free_contiguous_ports_are_contiguous_and_in_window() -> None:
+    # The allocator returns exactly n ascending, contiguous ports inside the window. It
+    # deliberately does NOT re-bind to "prove free" -- that is TOCTOU-racy and would reintroduce
+    # the exact flake class BACKLOG #1014 removes.
+    ports = _free_contiguous_ports(8)
+    assert len(ports) == 8
+    assert ports == list(range(ports[0], ports[0] + 8))
+    assert ports[0] >= _INBOUND_PORT_LO
+    assert ports[-1] < _INBOUND_PORT_HI
+
+
+def test_free_contiguous_ports_fails_loud_when_unsatisfiable() -> None:
+    # tries=0 hits the post-loop exhaustion branch deterministically (without occupying the
+    # whole window) and must raise -- never fall back silently to a fixed port (BACKLOG #1014).
+    with pytest.raises(RuntimeError, match="could not reserve"):
+        _free_contiguous_ports(8, tries=0)
+
+
+def test_free_contiguous_ports_fails_loud_when_window_too_narrow() -> None:
+    # The width guard fires BEFORE any probing when the requested block cannot fit the window
+    # at all: asking for one more port than the window holds can never be satisfied, so it
+    # raises up front rather than looping (BACKLOG #1014 -- fail loud, never a silent fallback).
+    with pytest.raises(RuntimeError, match="cannot reserve"):
+        _free_contiguous_ports(_INBOUND_PORT_HI - _INBOUND_PORT_LO + 1)