diff --git a/src/xceadmin/tests/_live_e2e_common.py b/src/xceadmin/tests/_live_e2e_common.py index c6408393..fdd0c170 100644 --- a/src/xceadmin/tests/_live_e2e_common.py +++ b/src/xceadmin/tests/_live_e2e_common.py @@ -124,3 +124,115 @@ def wait_for_port(host: str, port: int, timeout_s: float) -> None: last_err = exc time.sleep(0.2) raise TimeoutError(f"controller never opened {host}:{port}: {last_err}") + + +# ── Log polling ────────────────────────────────────────────────────────────── + +LOG_POLL_INTERVAL_S = 0.2 + + +def poll_log(log_file: Path, done, timeout_s: float) -> tuple[str, bool]: + """Re-read ``log_file`` every :data:`LOG_POLL_INTERVAL_S` until + ``done(log_text)`` is true or ``timeout_s`` passes. Returns the last text + read and whether ``done`` was satisfied. A log file that does not exist + yet reads as empty. + """ + deadline = time.monotonic() + timeout_s + while True: + log_text = log_file.read_text() if log_file.is_file() else "" + if done(log_text): + return log_text, True + if time.monotonic() >= deadline: + return log_text, False + time.sleep(LOG_POLL_INTERVAL_S) + + +# ── bug-1725: the primary's commit evidence is partly logged AFTER the reply ─ + +ACTIVATION_POLL_TIMEOUT_S = 5.0 + + +def assert_primary_committed( + log_file: Path, version_tag_hex: str, timeout_s: float = ACTIVATION_POLL_TIMEOUT_S +) -> str: + """Assert the controller log at ``log_file`` shows the config push with + tag ``version_tag_hex`` committed, upgraded and activated; return the log + text. + + Two of these lines can be written after the client already has its + success reply (bug-1725), so a single read straight after the CLI exits + can miss them: + + - ``Config upgraded to tag=...`` is logged just after ``xce_reply()`` in + ``do_upgrade_config_with_bytes`` (config_repl.c). + - ``config activated from committed marker tag=...`` comes from + ``config_activate_committed()``, which runs on the WAL apply pump + (persist.c, bug-1469). Apply is a separate step from commit, and the + reply only waits for commit. + + So this polls until every expected line is there, up to ``timeout_s``, + and then asserts each one separately so a failure names the missing line. + """ + expected = [ + "Config change committed", + f"Config upgraded to tag={version_tag_hex}", + f"config activated from committed marker tag={version_tag_hex}", + ] + log_text, _ = poll_log(log_file, lambda text: all(line in text for line in expected), timeout_s) + for line in expected: + assert line in log_text, ( + f"{line!r} not in {log_file} within {timeout_s}s of the push " + f"replying (it may be logged after the reply, so this waits -- " + f"see assert_primary_committed). log_text={log_text!r}" + ) + return log_text + + +# ── bug-1485: an elected primary is not write-ready until its own-term no-op ─ +# commits + +# ha.c ha_arm_own_term_gate: logged once per promotion, after the leader +# appends its own-term WAL_NOOP and arms the commit engine's writable_lsn gate. +OWN_TERM_NOOP_RE = re.compile(r"own-term noop appended lsn=(\d+)") +# commit.c commit_advance_locked: logged on every real committed_lsn advance +# (its own comment calls it the operator/test signal for the watermark). +COMMIT_ADVANCE_RE = re.compile(r"commit: committed_lsn \d+ -> (\d+)") + + +def primary_write_ready(log_text: str) -> bool: + """True once the primary's committed_lsn has reached the LSN of its + latest own-term no-op. + + Raft does not let a new leader commit anything until an entry from its + own term is committed on a majority. Before that, a client write such as + ``config push`` can only time out. bug-1485 is that case: when the + standbys join as late joiners (outbound connect after promotion, not via + their vote connections), the no-op was appended while no standby was + registered, so it was not replicated. It only reaches them from the + heartbeat-driven catch-up, a heartbeat cycle or two after the election. + A push in that window fails the 500 ms majority barrier with -ETIMEDOUT + (-110), which the CLI prints as ``versionTag=0 ackedCount=0``. + """ + noops = list(OWN_TERM_NOOP_RE.finditer(log_text)) + if not noops: + return False + latest = noops[-1] + noop_lsn = int(latest.group(1)) + # Only commits logged after the latest no-op count: an earlier term's + # commit progress says nothing about this term's no-op. + commits = [int(m.group(1)) for m in COMMIT_ADVANCE_RE.finditer(log_text, latest.end())] + return any(lsn >= noop_lsn for lsn in commits) + + +def wait_for_primary_write_ready(log_file: Path, timeout_s: float) -> None: + """Poll the primary's log until :func:`primary_write_ready` holds. Raises + ``TimeoutError`` if it does not within ``timeout_s``. + """ + log_text, ready = poll_log(log_file, primary_write_ready, timeout_s) + if not ready: + raise TimeoutError( + f"primary never committed its own-term no-op within {timeout_s}s, " + f"so it cannot commit a client write yet ({log_file}; " + f"noops={OWN_TERM_NOOP_RE.findall(log_text)} " + f"commits={COMMIT_ADVANCE_RE.findall(log_text)})" + ) diff --git a/src/xceadmin/tests/test_e2e_live_ha_push.py b/src/xceadmin/tests/test_e2e_live_ha_push.py index 136029ba..d7d70619 100644 --- a/src/xceadmin/tests/test_e2e_live_ha_push.py +++ b/src/xceadmin/tests/test_e2e_live_ha_push.py @@ -192,6 +192,23 @@ data point against it. Total this session: 40/40 passed, 0/120 stderr logs non-empty. Left as an open, low-priority, currently-unreproducible flake; the stderr-capture infrastructure below is now in place for the next person who hits it live to actually get a diagnostic instead of DEVNULL silence. + +bug-1485 root cause (2026-10-03): the ``versionTag=0`` failure, and the +"status -110 for a short window after promotion" described above, is the +primary rejecting the push with -ETIMEDOUT, not a warmup problem. It +happens when the election completes without promoting the vote connections +to standby channels, so both standbys register as late joiners (outbound +connect after ``raft_cb_become_leader``). The own-term WAL no-op was +appended while no standby was registered (``repl_broadcast: no standbys +registered``), so it was never sent. The heartbeat-driven catch-up takes +one action per standby per 3 s cycle, config push first and WAL stream on a +later cycle, so the no-op can stay uncommitted for several seconds. A push +in that window appends the config marker, the standbys NACK it (they are +missing the no-op), and the 500 ms majority barrier times out. Under CPU +load this hit 7 of 28 runs, and every failure was a late-joiner election +with no ``commit: committed_lsn 0 -> 1`` before the push. The fixture now +waits for the primary to commit its own-term no-op +(``wait_for_primary_write_ready``) before yielding. """ from __future__ import annotations @@ -211,9 +228,11 @@ from _live_e2e_common import DEPLOY_HA_SH, REPO_CONFIGS_DIR from _live_e2e_common import READY_TIMEOUT_S from _live_e2e_common import VERSION_TAG_RE from _live_e2e_common import XCEADMIN_CANDIDATES +from _live_e2e_common import assert_primary_committed from _live_e2e_common import find_first_executable from _live_e2e_common import free_tcp_ports from _live_e2e_common import wait_for_port +from _live_e2e_common import wait_for_primary_write_ready _N_NODES = 3 _STAGGER_S = 0.5 @@ -378,6 +397,14 @@ def live_ha_cluster(tmp_path: Path): primary_node = _wait_for_primary(nodes, _ELECTION_TIMEOUT_S) standby_nodes = [n for n in nodes if n.name != primary_node.name] _wait_for_standbys(standby_nodes, _ELECTION_TIMEOUT_S) + # bug-1485: CTRL_PRIMARY/CTRL_STANDBY only say the roles are settled. + # The primary cannot commit a client write until its own-term no-op + # is committed on a majority, and when the standbys joined as late + # joiners that no-op only reaches them via heartbeat-driven catch-up, + # a heartbeat cycle or two later. A push before then fails the + # majority barrier with -ETIMEDOUT (versionTag=0 ackedCount=0). Wait + # for the commit instead of hoping the settle sleep below covers it. + wait_for_primary_write_ready(primary_node.log_file, _ELECTION_TIMEOUT_S) # Settle window + fresh port reachability recheck (module # docstring's "Election-wait" section has the full live-repro @@ -430,38 +457,6 @@ def _assert_no_decode_crash(push_result: subprocess.CompletedProcess) -> None: assert "Traceback" not in push_result.stderr, push_result.stderr -# config_activate_committed() (config_repl.c) runs off the WAL apply-pump -# dispatch (persist.c, the on_commit_advanced-adjacent seam bug-1469 added), -# asynchronously relative to the majority-ack-gated client reply -# (commitIndex/lastApplied are distinct steps -- controller/CLAUDE.md's TLA+ -# mapping) -- a client can legitimately see its push's "success" reply -# before the primary's OWN local apply of that same marker has logged -# "config activated...". Poll instead of a single-shot read to close this -# real race rather than assume an ordering the design never guarantees. -_ACTIVATION_POLL_TIMEOUT_S = 5.0 -_ACTIVATION_POLL_INTERVAL_S = 0.2 - - -def _assert_primary_committed(log_file: Path, version_tag_hex: str) -> str: - marker_line = f"config activated from committed marker tag={version_tag_hex}" - deadline = time.monotonic() + _ACTIVATION_POLL_TIMEOUT_S - log_text = log_file.read_text() - while marker_line not in log_text and time.monotonic() < deadline: - time.sleep(_ACTIVATION_POLL_INTERVAL_S) - log_text = log_file.read_text() - - assert f"tag={version_tag_hex}" in log_text, log_text - assert "Config change committed" in log_text - assert f"Config upgraded to tag={version_tag_hex}" in log_text - assert marker_line in log_text, ( - f"{marker_line!r} never appeared within {_ACTIVATION_POLL_TIMEOUT_S}s " - f"of the push replying success -- see this function's own comment " - f"for why activation is asynchronous relative to the reply. " - f"log_text={log_text!r}" - ) - return log_text - - def test_live_config_push_against_primary_reaches_success_path( live_ha_cluster: HaCluster, tmp_path: Path ) -> None: @@ -494,7 +489,10 @@ def test_live_config_push_against_primary_reaches_success_path( match = VERSION_TAG_RE.search(push_result.stdout) assert match, f"no versionTag/ackedCount found in stdout: {push_result.stdout!r}" version_tag, acked_count = int(match.group(1)), int(match.group(2)) - assert version_tag > 0 + assert version_tag > 0, ( + f"expected a non-zero versionTag, got {version_tag} " + f"(stdout={push_result.stdout!r} stderr={push_result.stderr!r})" + ) # The real, live 3-node success path task 6.2 could not reach: a # genuine positive ackedCount (real standby ACKs exist now) and exit 0 @@ -506,7 +504,7 @@ def test_live_config_push_against_primary_reaches_success_path( assert push_result.returncode == 0, push_result.stderr version_tag_hex = f"{version_tag:016x}" - primary_log_text = _assert_primary_committed(live_ha_cluster.primary.log_file, version_tag_hex) + primary_log_text = assert_primary_committed(live_ha_cluster.primary.log_file, version_tag_hex) # Majority-ACK barrier evidence on the primary's own log: at least one # standby genuinely acked this exact tag (push_config_to_standbys). @@ -592,7 +590,10 @@ def test_live_config_push_against_standby_redirects_and_succeeds( match = VERSION_TAG_RE.search(push_result.stdout) assert match, f"no versionTag/ackedCount found in stdout: {push_result.stdout!r}" version_tag, acked_count = int(match.group(1)), int(match.group(2)) - assert version_tag > 0 + assert version_tag > 0, ( + f"expected a non-zero versionTag, got {version_tag} " + f"(stdout={push_result.stdout!r} stderr={push_result.stderr!r})" + ) # Same success bar as the direct-to-primary test: a genuine positive # ackedCount from real standby ACKs, exit 0 -- the redirect hop changes @@ -605,4 +606,4 @@ def test_live_config_push_against_standby_redirects_and_succeeds( assert push_result.returncode == 0, push_result.stderr version_tag_hex = f"{version_tag:016x}" - _assert_primary_committed(live_ha_cluster.primary.log_file, version_tag_hex) + assert_primary_committed(live_ha_cluster.primary.log_file, version_tag_hex) diff --git a/src/xceadmin/tests/test_e2e_live_push.py b/src/xceadmin/tests/test_e2e_live_push.py index efbe0b9c..8c6c31a3 100644 --- a/src/xceadmin/tests/test_e2e_live_push.py +++ b/src/xceadmin/tests/test_e2e_live_push.py @@ -60,6 +60,7 @@ from _live_e2e_common import DEPLOY_HA_SH, REPO_CONFIGS_DIR from _live_e2e_common import READY_TIMEOUT_S as _READY_TIMEOUT_S from _live_e2e_common import VERSION_TAG_RE as _VERSION_TAG_RE from _live_e2e_common import XCEADMIN_CANDIDATES as _XCEADMIN_CANDIDATES +from _live_e2e_common import assert_primary_committed from _live_e2e_common import find_first_executable as _find_first_executable from _live_e2e_common import free_tcp_port as _free_tcp_port from _live_e2e_common import wait_for_port as _wait_for_port @@ -186,7 +187,10 @@ def test_live_config_push_against_real_single_node_controller(live_controller, t match = _VERSION_TAG_RE.search(push_result.stdout) assert match, f"no versionTag/ackedCount found in stdout: {push_result.stdout!r}" version_tag, acked_count = int(match.group(1)), int(match.group(2)) - assert version_tag > 0 + assert version_tag > 0, ( + f"expected a non-zero versionTag, got {version_tag} " + f"(stdout={push_result.stdout!r} stderr={push_result.stderr!r})" + ) # REAL, live-verified single-node behavior (module docstring): a # genuinely peer-less controller has no standbys to ACK, so @@ -197,9 +201,7 @@ def test_live_config_push_against_real_single_node_controller(live_controller, t assert push_result.returncode == 1 assert "push rejected" in push_result.stderr - log_text = log_file.read_text() - version_tag_hex = f"{version_tag:016x}" - assert f"tag={version_tag_hex}" in log_text, log_text - assert "Config change committed" in log_text - assert f"Config upgraded to tag={version_tag_hex}" in log_text - assert f"config activated from committed marker tag={version_tag_hex}" in log_text + # bug-1725: "Config upgraded" (logged after xce_reply) and "config + # activated" (logged by the asynchronous WAL apply pump) can land after + # the CLI has its reply, so poll rather than read the log once. + assert_primary_committed(log_file, f"{version_tag:016x}") diff --git a/src/xceadmin/tests/test_live_e2e_common.py b/src/xceadmin/tests/test_live_e2e_common.py index 20c1a9dd..ab0942c6 100644 --- a/src/xceadmin/tests/test_live_e2e_common.py +++ b/src/xceadmin/tests/test_live_e2e_common.py @@ -74,3 +74,142 @@ def test_controller_and_xceadmin_candidates_respect_their_own_env_vars(monkeypat monkeypatch.delenv("XCE_CONTROLLER_BIN", raising=False) monkeypatch.delenv("XCEADMIN_BIN", raising=False) importlib.reload(mod) # restore default candidates for any test after this one + + +# ── bug-1725: post-reply log assertions must poll, not read once ───────────── +# +# The controller replies to OP_upgradeConfig BEFORE two of the log lines the +# live tests check: "Config upgraded to tag=..." is logged right after +# xce_reply() in do_upgrade_config_with_bytes, and "config activated from +# committed marker tag=..." comes from config_activate_committed() on the +# asynchronous WAL apply pump. A single read right after the CLI exits can +# therefore miss either one. These tests stand in for the slow controller +# with a thread that appends the lines after a delay. + +_TAG_HEX = "8ae6096797997dd3" +_COMMITTED_LINES = ( + "Config change committed (marker_lsn=2, tag=8ae6096797997dd3)\n", + "Config upgraded to tag=8ae6096797997dd3 gen=2 (acked=2 standbys)\n", + "config activated from committed marker tag=8ae6096797997dd3 (hash-verified out-of-band bytes)\n", +) + + +def _append_later(path, lines, delay_s): + import threading + import time + + def _writer(): + time.sleep(delay_s) + with open(path, "a") as f: + f.writelines(lines) + + t = threading.Thread(target=_writer) + t.start() + return t + + +def test_assert_primary_committed_waits_for_lines_logged_after_the_reply(tmp_path): + from _live_e2e_common import assert_primary_committed + + log_file = tmp_path / "ctrl-1.log" + # Only the pre-reply line is there when the CLI returns; the other two + # land 0.5s later. + log_file.write_text(_COMMITTED_LINES[0]) + writer = _append_later(log_file, _COMMITTED_LINES[1:], 0.5) + try: + log_text = assert_primary_committed(log_file, _TAG_HEX, timeout_s=5.0) + finally: + writer.join() + + assert f"config activated from committed marker tag={_TAG_HEX}" in log_text + assert f"Config upgraded to tag={_TAG_HEX}" in log_text + + +def test_assert_primary_committed_fails_when_activation_never_logged(tmp_path): + import pytest + + from _live_e2e_common import assert_primary_committed + + log_file = tmp_path / "ctrl-1.log" + log_file.write_text("".join(_COMMITTED_LINES[:2])) + + with pytest.raises(AssertionError, match="config activated from committed marker"): + assert_primary_committed(log_file, _TAG_HEX, timeout_s=0.5) + + +def test_assert_primary_committed_fails_when_upgrade_line_never_logged(tmp_path): + import pytest + + from _live_e2e_common import assert_primary_committed + + log_file = tmp_path / "ctrl-1.log" + log_file.write_text(_COMMITTED_LINES[0] + _COMMITTED_LINES[2]) + + with pytest.raises(AssertionError, match="Config upgraded to tag="): + assert_primary_committed(log_file, _TAG_HEX, timeout_s=0.5) + + +def test_primary_write_ready_ignores_commits_logged_before_the_latest_noop(): + from _live_e2e_common import primary_write_ready + + # A commit to lsn 5 from an earlier term, then a new term's no-op at + # lsn 3 that has not committed yet: not write-ready. + earlier = "[ INFO] [ commit_advance_locked] commit: committed_lsn 4 -> 5 (n_peers=3)\n" + noop_3 = _NOOP_LINE.replace("lsn=1", "lsn=3") + assert primary_write_ready(earlier + noop_3) is False + commit_3 = "[ INFO] [ commit_advance_locked] commit: committed_lsn 2 -> 3 (n_peers=3)\n" + assert primary_write_ready(earlier + noop_3 + commit_3) is True + + +# ── bug-1485: a freshly elected primary is not write-ready until its own-term +# no-op commits ────────────────────────────────────────────────────────────── +# +# Real log lines (ha.c ha_arm_own_term_gate, commit.c commit_advance_locked). + +_NOOP_LINE = "[ INFO] [ ha_arm_own_term_gate] HA: own-term noop appended lsn=1, writable_lsn gate armed\n" +_COMMIT_0_1 = "[ INFO] [ commit_advance_locked] commit: committed_lsn 0 -> 1 (n_peers=3)\n" + + +def test_primary_write_ready_false_until_own_term_noop_commits(): + from _live_e2e_common import primary_write_ready + + assert not primary_write_ready("") + # Elected, no-op appended but not yet replicated to a majority -- the + # late-joiner election shape that made every push time out (-110). + assert not primary_write_ready(_NOOP_LINE) + assert primary_write_ready(_NOOP_LINE + _COMMIT_0_1) + + +def test_primary_write_ready_uses_the_latest_noop(): + from _live_e2e_common import primary_write_ready + + second_noop = _NOOP_LINE.replace("lsn=1", "lsn=3") + # A later re-election appends a new no-op at lsn 3: commit up to 1 is no + # longer enough. + assert not primary_write_ready(_NOOP_LINE + _COMMIT_0_1 + second_noop) + commit_1_3 = _COMMIT_0_1.replace("0 -> 1", "1 -> 3") + assert primary_write_ready(_NOOP_LINE + _COMMIT_0_1 + second_noop + commit_1_3) + + +def test_wait_for_primary_write_ready_waits_for_delayed_commit(tmp_path): + from _live_e2e_common import wait_for_primary_write_ready + + log_file = tmp_path / "ctrl-1.log" + log_file.write_text(_NOOP_LINE) + writer = _append_later(log_file, [_COMMIT_0_1], 0.5) + try: + wait_for_primary_write_ready(log_file, timeout_s=5.0) + finally: + writer.join() + + +def test_wait_for_primary_write_ready_times_out_when_noop_never_commits(tmp_path): + import pytest + + from _live_e2e_common import wait_for_primary_write_ready + + log_file = tmp_path / "ctrl-1.log" + log_file.write_text(_NOOP_LINE) + + with pytest.raises(TimeoutError, match="own-term no-op"): + wait_for_primary_write_ready(log_file, timeout_s=0.5) diff --git a/src/xceadmin/tests/test_wire.py b/src/xceadmin/tests/test_wire.py index 9505ee37..1050858a 100644 --- a/src/xceadmin/tests/test_wire.py +++ b/src/xceadmin/tests/test_wire.py @@ -15,6 +15,7 @@ import socket import ssl import struct import threading +import time from pathlib import Path import pytest @@ -43,6 +44,10 @@ class _TlsEchoServer: self._listener.bind(("127.0.0.1", 0)) self._listener.listen(1) self.port = self._listener.getsockname()[1] + #: OSError the handler raised, if any. ``_run`` still swallows it + #: (a peer hang-up is normal for most handlers) but keeps it so a + #: test can tell "handler failed" from "handler saw wrong bytes". + self.handler_error: OSError | None = None self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() @@ -60,8 +65,8 @@ class _TlsEchoServer: return try: self._handler(tls_conn) - except OSError: - pass + except OSError as exc: + self.handler_error = exc finally: tls_conn.close() @@ -168,32 +173,51 @@ def test_connect_handshake_timeout_raises_connection_timeout_error_and_closes_cl # --------------------------------------------------------------------------- -def test_send_frame_writes_exact_known_good_byte_sequence(tls_certs): +@pytest.mark.parametrize("settle_s", [0.0, 0.05], ids=["immediate", "tickets-pending"]) +def test_send_frame_writes_exact_known_good_byte_sequence(tls_certs, settle_s): """Requirement: framing matches the controller's own byte-for-byte, verified against a known-good byte sequence (not just a round-trip). + + ``settle_s`` (bug-1683 regression): pausing after connect() lets the + server's post-handshake TLS 1.3 NewSessionTicket records land, unread, + in the client's receive buffer. Closing a socket with unread inbound + data makes the kernel send RST instead of FIN, and Linux then discards + whatever the server had not yet read (the payload record), so the + capture must finish before the client closes. """ captured = bytearray() + capture_done = threading.Event() def _capture_handler(conn: ssl.SSLSocket) -> None: # Read exactly what send_frame wrote, raw, without going through # recv_frame -- this is what proves the header shape independently # of this module's own decoder. - while len(captured) < 8 + 4: - chunk = conn.recv(4096) - if not chunk: - break - captured.extend(chunk) + try: + while len(captured) < 8 + 4: + chunk = conn.recv(4096) + if not chunk: + break + captured.extend(chunk) + finally: + capture_done.set() srv = _TlsEchoServer(_capture_handler, tls_certs.cert_file, tls_certs.key_file) try: sock = wire.connect("127.0.0.1", srv.port, timeout_s=5.0) try: + if settle_s: + time.sleep(settle_s) wire.send_frame(sock, b"xce!") + # Keep the client open until the server has read all 12 bytes: + # an early close() can RST the connection and drop the + # still-unread payload record (bug-1683). + assert capture_done.wait(timeout=5.0), "server capture never finished" finally: sock.close() finally: srv.close() + assert srv.handler_error is None, f"capture handler failed: {srv.handler_error!r}" # 8-byte native-order uint64_t length header (4), then the 4 payload # bytes -- byte-for-byte, matching struct.pack("=Q", 4) + b"xce!". expected_header = struct.pack("=Q", 4)