Stop the node_fold tests racing the real :4243 — the flake that cost three shard re-runs (#1166)

ciris_engine/logic/runtime/node_fold.py

@@ -27,6 +27,21 @@
 logger = logging.getLogger(__name__)
 
 
+#: The port the folded node's substrate read-API listens on.
+#:
+#: A NAMED constant because the three functional uses below were bare literals,
+#: and a test that wants to exercise the ownership logic therefore had to bind
+#: the REAL 4243 — which made `test_refuses_a_live_node_this_process_does_not_own`
+#: race anything else on the box that touched the port. It failed on three
+#: unrelated PRs (#1162, #1164, #1169), each costing a shard re-run
+#: (CIRISAgent#1166). Tests now point this at an ephemeral port instead.
+#:
+#: NOT a configuration knob: the node's real port is fixed by the substrate and
+#: by CIRISClient's NODE_ONLY_ENDPOINT. Changing it here changes only what this
+#: module probes, which is why nothing reads it from the environment.
+NODE_FOLD_PORT = 4243
+
+
 def _this_process_owns_port(port: int) -> Optional[bool]:
     """Does THIS process hold the listening socket on `port`?
 
@@ -131,7 +146,6 @@ def _resolve_home() -> str:
     return str(get_ciris_home())
 
 
-
 def _repair_if_bricked_then_raise(node_error: object) -> None:
     """Last stop for a boot that cannot succeed on this home.
 
@@ -168,6 +182,7 @@ def _repair_if_bricked_then_raise(node_error: object) -> None:
         ) from None
     raise err
 
+
 def _resolve_key_id() -> Optional[str]:
     """Federation keystore alias for the node — the SAME alias the Engine uses.
 
@@ -347,7 +362,9 @@ def stop_node_fold(timeout_secs: float = 30.0) -> Optional[bool]:
         return None
     shutdown_node = getattr(ciris_server, "shutdown_node", None)
     if shutdown_node is None:
-        logger.warning("Node fold: ciris_server.shutdown_node unavailable; :4243 may stay bound after exit (CIRISAgent#1102)")
+        logger.warning(
+            "Node fold: ciris_server.shutdown_node unavailable; :4243 may stay bound after exit (CIRISAgent#1102)"
+        )
         return None
     try:
         freed = bool(shutdown_node(timeout_secs=timeout_secs))
@@ -395,7 +412,7 @@ def start_node_fold(brain_port: int, *, home: Optional[str] = None, key_id: Opti
     import socket as _socket
 
     try:
-        with _socket.create_connection(("127.0.0.1", 4243), timeout=1):
+        with _socket.create_connection(("127.0.0.1", NODE_FOLD_PORT), timeout=1):
             live_node = True
     except OSError:
         live_node = False
@@ -421,7 +438,7 @@ def start_node_fold(brain_port: int, *, home: Optional[str] = None, key_id: Opti
             # sufficient condition; the node must also be ALIVE.
             for probe_path in ("/v1/self/identity", "/v1/health"):
                 try:
-                    with _urlreq.urlopen(f"http://127.0.0.1:4243{probe_path}", timeout=2) as resp:
+                    with _urlreq.urlopen(f"http://127.0.0.1:{NODE_FOLD_PORT}{probe_path}", timeout=2) as resp:
                         identity_text += resp.read(65536).decode("utf-8", errors="replace")
                         http_alive = True
                 except _urlerr.HTTPError:
@@ -457,10 +474,10 @@ def start_node_fold(brain_port: int, *, home: Optional[str] = None, key_id: Opti
             #
             # Ownership alone was the wrong sufficient condition — my first fix. It
             # answers 'is it ours', and a zombie of ours is still unusable.
-            owns = _this_process_owns_port(4243)
+            owns = _this_process_owns_port(NODE_FOLD_PORT)
             if not http_alive:
                 raise RuntimeError(
-                    "node fold: :4243 is bound but does not speak HTTP "
+                    f"node fold: :{NODE_FOLD_PORT} is bound but does not speak HTTP "
                     f"(owned_by_us={owns}). Reusing a dead listener leaves every "
                     "/v1/auth call answering 502 for the life of the process — observed "
                     "in CI as 0 node-side successes with 39 proxy failures.\n"
@@ -472,7 +489,7 @@ def start_node_fold(brain_port: int, *, home: Optional[str] = None, key_id: Opti
                 )
             if owns is False:
                 raise RuntimeError(
-                    "node fold: :4243 is serving, its identity could not be read, and the "
+                    f"node fold: :{NODE_FOLD_PORT} is serving, its identity could not be read, and the "
                     "listening socket is NOT held by this process — so it is not ours. "
                     "Reusing it would ship traces to a foreign node (verify_unknown_key, "
                     "QA all_1 RCA). Run one node-folded stack per host."
@@ -629,7 +646,7 @@ def _compose_phase() -> Optional[str]:
         if _node_error is not None:
             _repair_if_bricked_then_raise(_node_error)
         try:
-            with socket.create_connection(("127.0.0.1", 4243), timeout=1):
+            with socket.create_connection(("127.0.0.1", NODE_FOLD_PORT), timeout=1):
                 node_up = True
                 break
         except OSError:
@@ -652,10 +669,11 @@ def _compose_phase() -> Optional[str]:
     # here rather than at process start: a home that is about to brick must not
     # be marked as already fixed.
     try:
+        from pathlib import Path as _P
+
         from ciris_engine.constants import CIRIS_VERSION
         from ciris_engine.logic.setup.bricked_install import record_install_version
         from ciris_engine.logic.utils.path_resolution import get_ciris_home
-        from pathlib import Path as _P
 
         record_install_version(_P(get_ciris_home()), CIRIS_VERSION)
     except Exception:  # noqa: BLE001

tests/ciris_engine/logic/runtime/test_node_fold_reuse_requires_proof.py

@@ -35,8 +35,30 @@
 from ciris_engine.logic.runtime import node_fold
 
 
+@pytest.fixture(autouse=True)
+def node_port(monkeypatch) -> int:
+    """Point `node_fold` at an EPHEMERAL port instead of the real 4243.
+
+    These tests assert on the ownership logic, not on a port number, and binding
+    the real 4243 made them race every other thing on the box that touches it:
+    the sibling `own_listener` in another xdist worker, a leftover node from an
+    earlier test, a TIME_WAIT socket with SO_REUSEADDR. The foreign-listener test
+    failed on three unrelated PRs (#1162 shard 6, #1164 shard 2, #1169 shard 7),
+    each costing a ~18-minute shard re-run, and the flaky-retry plugin re-ran into
+    the same shared environment so retries rarely helped (CIRISAgent#1166).
+
+    The kernel hands out a port nothing else holds, so the race is gone by
+    construction rather than by hoping the box is quiet.
+    """
+    with closing(socket.socket()) as probe:
+        probe.bind(("127.0.0.1", 0))
+        port = probe.getsockname()[1]
+    monkeypatch.setattr(node_fold, "NODE_FOLD_PORT", port)
+    return port
+
+
 @pytest.fixture
-def own_listener():
+def own_listener(node_port: int):
     """A listener on 4243 held by THIS process — the in-process-restart shape.
 
     Socket ownership is the discriminator, so the fixtures differ only in WHO
@@ -48,17 +70,17 @@ def own_listener():
     s = socket.socket()
     s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     try:
-        s.bind(("127.0.0.1", 4243))
+        s.bind(("127.0.0.1", node_port))
     except OSError:  # pragma: no cover - a real node is running on this box
         s.close()
-        pytest.skip("port 4243 already in use")
+        pytest.skip(f"port {node_port} already in use")
     s.listen(8)
     with closing(s):
         yield s
 
 
 @pytest.fixture
-def foreign_listener():
+def foreign_listener(node_port: int):
     """A listener on 4243 held by a DIFFERENT process — the CI shape.
 
     A child process, not a thread: the point is that the socket belongs to
@@ -76,17 +98,17 @@ def foreign_listener():
             "class H(BaseHTTPRequestHandler):\n"
             "    def do_GET(self): self.send_error(404)\n"
             "    def log_message(self,*a): pass\n"
-            "HTTPServer(('127.0.0.1',4243),H).serve_forever()",
+            f"HTTPServer(('127.0.0.1',{node_port}),H).serve_forever()",
         ]
     )
     for _ in range(50):
         with socket.socket() as probe:
-            if probe.connect_ex(("127.0.0.1", 4243)) == 0:
+            if probe.connect_ex(("127.0.0.1", node_port)) == 0:
                 break
         time.sleep(0.1)
     else:  # pragma: no cover
         child.kill()
-        pytest.skip("could not stand up a foreign listener on 4243")
+        pytest.skip(f"could not stand up a foreign listener on {node_port}")
     try:
         yield child
     finally:
@@ -132,9 +154,7 @@ def test_liveness_and_ownership_are_both_consulted() -> None:
 
     source = inspect.getsource(node_fold.start_node_fold)
     guard = source[source.index("if not identity_text:") :]
-    assert "_this_process_owns_port" in guard[:3000], (
-        "the cannot-read-identity arm no longer consults socket ownership"
-    )
+    assert "_this_process_owns_port" in guard[:3000], "the cannot-read-identity arm no longer consults socket ownership"
     assert "http_alive" in guard[:3000], (
         "the cannot-read-identity arm no longer distinguishes a LIVE node with "
         "drifted endpoints from a dead socket — that collapse is what left the "
@@ -169,7 +189,7 @@ def test_refuses_a_zombie_listener_even_though_it_is_ours(own_listener, tmp_path
     )
 
 
-def test_reuses_our_live_node_whose_identity_endpoints_drifted(tmp_path):
+def test_reuses_our_live_node_whose_identity_endpoints_drifted(tmp_path, node_port: int):
     """Alive but unreadable is the older-wheel case this branch exists for.
 
     An HTTPError PROVES a server answered; connection-refused proves one did not.
@@ -186,13 +206,54 @@ def log_message(self, *args):  # noqa: A003 - silence the test server
             return
 
     try:
-        srv = HTTPServer(("127.0.0.1", 4243), _Drifted)
+        srv = HTTPServer(("127.0.0.1", node_port), _Drifted)
     except OSError:  # pragma: no cover
-        pytest.skip("port 4243 already in use")
+        pytest.skip(f"port {node_port} already in use")
     threading.Thread(target=srv.serve_forever, daemon=True).start()
     try:
         # Must not raise: our socket, alive, endpoints simply absent.
         node_fold.start_node_fold(8080, home=str(tmp_path))
     finally:
         srv.shutdown()
         srv.server_close()
+
+
+# ---------------------------------------------------------------------------
+# stop_node_fold: the #1102 arm that cannot free the port
+# ---------------------------------------------------------------------------
+
+
+def test_stop_node_fold_says_so_when_the_wheel_cannot_free_the_port(monkeypatch, caplog):
+    """A wheel older than `shutdown_node` leaves the port bound after exit.
+
+    `stop_node_fold` returns None for BOTH "this process never started a node"
+    and "the wheel cannot stop it" -- so the warning is the only thing that
+    distinguishes a clean no-op from a node that will still be holding the port
+    when the next boot tries to bind it (CIRISAgent#1102, the EADDRINUSE the
+    five-platform gate's post-reset port check exists to catch). Untested until
+    now, which is how it came to be the uncovered half of this module's
+    port-constant change.
+    """
+    import logging
+    import sys
+    import types
+
+    monkeypatch.setattr(node_fold, "_node_thread", object(), raising=False)
+    # A wheel that predates shutdown_node: importable, but without the symbol.
+    monkeypatch.setitem(sys.modules, "ciris_server", types.ModuleType("ciris_server"))
+
+    with caplog.at_level(logging.WARNING, logger=node_fold.logger.name):
+        assert node_fold.stop_node_fold() is None
+
+    assert any("shutdown_node unavailable" in r.message for r in caplog.records), caplog.text
+    assert any("CIRISAgent#1102" in r.message for r in caplog.records), caplog.text
+
+
+def test_stop_node_fold_is_a_clean_no_op_when_we_started_nothing(monkeypatch, caplog):
+    """The other None: nothing to stop, and nothing to warn about."""
+    import logging
+
+    monkeypatch.setattr(node_fold, "_node_thread", None, raising=False)
+    with caplog.at_level(logging.WARNING, logger=node_fold.logger.name):
+        assert node_fold.stop_node_fold() is None
+    assert not [r for r in caplog.records if "shutdown_node unavailable" in r.message]