fix(xceadmin tests): root-cause the three flaky tests (bug-1683, bug-1485, bug-1725)
- bug-1683 (test_wire known-good capture): the client closed its TLS socket with the server's TLS 1.3 NewSessionTickets unread, so the kernel sent RST and the server lost the unread payload record; the server swallowed the OSError. The client now waits for the capture to finish, handler errors are recorded and asserted, and a 50 ms "tickets-pending" variant makes the RST race deterministic. - bug-1485 (live HA push, versionTag=0): the CLI's 0 was an error reply (-110): after a late-joiner election the primary's own-term no-op only commits ~6 s later via heartbeat catch-up, so a push in that window misses the majority barrier. The fixture now waits until the latest own-term no-op has committed (wait_for_primary_write_ready); versionTag asserts print the CLI's stdout/stderr. The controller latency itself is bug-1913. - bug-1725 (activation line missing): "Config upgraded" and the activation line are logged after the reply; both tests now poll for all three commit lines (shared assert_primary_committed). xceadmin suite 373 passed; live push tests looped (95 runs incl. heavy CPU load, 0 failures from these mechanisms). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
91782e4089
commit
362098b12a
GPG key ID:
80533536B0EB651C
5 changed files with 329 additions and 51 deletions
|
|
@ -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)})"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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}")
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue