diff --git a/src/exo/shared/election.py b/src/exo/shared/election.py index 958a83d2fa..9c2e3217bd 100644 --- a/src/exo/shared/election.py +++ b/src/exo/shared/election.py @@ -81,8 +81,20 @@ def __init__( self._candidates: list[ElectionMessage] = [] self._campaign_cancel_scope: CancelScope | None = None self._campaign_done: Event | None = None + self._campaigns_started = 0 + self._campaigns_finished = 0 self._tg = TaskGroup() + @property + def _campaign_active(self) -> bool: + return self._campaigns_finished < self._campaigns_started + + def _start_campaign( + self, candidates: list[ElectionMessage], campaign_timeout: float + ) -> None: + self._campaigns_started += 1 + self._tg.start_soon(self._campaign, candidates, campaign_timeout) + async def run(self): logger.info("Starting Election") try: @@ -95,6 +107,7 @@ async def run(self): candidates: list[ElectionMessage] = [] logger.debug("Starting initial campaign") self._candidates = candidates + self._campaigns_started += 1 await self._campaign(candidates, campaign_timeout=0.0) logger.debug("Initial campaign finished") finally: @@ -143,18 +156,28 @@ async def _election_receiver(self) -> None: self._candidates = candidates logger.debug(f"New candidates: {self._candidates}") logger.debug("Starting new campaign") - self._tg.start_soon( - self._campaign, candidates, DEFAULT_ELECTION_TIMEOUT - ) + self._start_campaign(candidates, DEFAULT_ELECTION_TIMEOUT) logger.debug("Campaign started") continue - # Dismiss old messages + # Reply to stale messages with our status so that a node + # campaigning below the cluster clock (e.g. a restarted + # master starting over from clock 0) deterministically + # learns the current clock and rejoins the election. if message.clock < self.clock: - logger.debug(f"Dropping old message: {message}") + logger.debug(f"Replying to stale message: {message}") + await self._em_sender.send(self._election_status()) continue logger.debug(f"Election added candidate {message}") # Now we are processing this rounds messages - including the message that triggered this round. self._candidates.append(message) + # An equal-clock candidacy with no campaign running would + # otherwise sit in a candidate list that nothing evaluates; + # start a campaign so the round deterministically resolves. + if not self._campaign_active: + logger.debug( + "Equal-clock candidacy with no active campaign; starting one" + ) + self._start_campaign(self._candidates, DEFAULT_ELECTION_TIMEOUT) async def _connection_receiver(self) -> None: with self._cm_receiver as connection_messages: @@ -173,9 +196,7 @@ async def _connection_receiver(self) -> None: candidates: list[ElectionMessage] = [] self._candidates = candidates logger.debug("Starting new campaign") - self._tg.start_soon( - self._campaign, candidates, DEFAULT_ELECTION_TIMEOUT - ) + self._start_campaign(candidates, DEFAULT_ELECTION_TIMEOUT) logger.debug("Campaign started") logger.debug("Connection message added") @@ -245,6 +266,7 @@ async def _campaign( logger.debug(f"Election {clock} cancelled") finally: logger.debug(f"Election {clock} finally") + self._campaigns_finished += 1 if self._campaign_cancel_scope is scope: self._campaign_cancel_scope = None logger.debug("Setting done event") diff --git a/src/exo/shared/tests/test_election.py b/src/exo/shared/tests/test_election.py index d1612e85e4..dd48812311 100644 --- a/src/exo/shared/tests/test_election.py +++ b/src/exo/shared/tests/test_election.py @@ -1,5 +1,5 @@ import pytest -from anyio import create_task_group, fail_after, move_on_after +from anyio import create_task_group, fail_after, move_on_after, sleep from exo.routing.connection_message import ConnectionMessage from exo.shared.election import Election, ElectionMessage, ElectionResult @@ -189,14 +189,19 @@ async def test_ignores_older_messages() -> None: if first.clock == 2: break - # Older message (clock=1) must be ignored (no second broadcast) + # An older message (clock=1) must not start a new round, but the + # sender must be told the current clock so it can rejoin (#2197) await em_in_tx.send(em(clock=1, seniority=999, node_id="B")) - got_second = False + reply = await em_out_rx.receive() + assert reply.clock == 2, "Reply must carry the current clock" + assert election.clock == 2, "An older message must not change the clock" + + got_more = False with move_on_after(0.05): _ = await em_out_rx.receive() - got_second = True - assert not got_second, "Should not receive a broadcast for an older round" + got_more = True + assert not got_more, "An older round must not trigger a new campaign" em_in_tx.close() cm_tx.close() @@ -402,3 +407,182 @@ async def test_tie_breaker_prefers_node_with_more_commands_seen() -> None: em_in_tx.close() cm_tx.close() co_tx.close() + + +@pytest.mark.anyio +async def test_equal_clock_candidacy_without_active_campaign_starts_a_round() -> None: + """ + Regression test for the "dead round" (issue #2197, part 2). + + A candidacy arriving with a clock equal to ours while no campaign is + running used to be appended to a candidate list that nothing would ever + evaluate. It must instead trigger a round that resolves deterministically. + """ + em_out_tx, _em_out_rx = channel[ElectionMessage]() + em_in_tx, em_in_rx = channel[ElectionMessage]() + er_tx, er_rx = channel[ElectionResult]() + cm_tx, cm_rx = channel[ConnectionMessage]() + co_tx, co_rx = channel[ForwarderCommand]() + + election = Election( + node_id=NodeId("B"), + election_message_receiver=em_in_rx, + election_message_sender=em_out_tx, + election_result_sender=er_tx, + connection_message_receiver=cm_rx, + command_receiver=co_rx, + is_candidate=True, + ) + + async with create_task_group() as tg: + with fail_after(2): + tg.start_soon(election.run) + + # The initial campaign resolves instantly with ourselves as master + first = await er_rx.receive() + assert first.session_id.master_node_id == NodeId("B") + + # Equal-clock candidacy (clock 0) from a clearly better candidate, + # while no campaign is active anymore + await em_in_tx.send(em(clock=0, seniority=50, node_id="A")) + + result = await er_rx.receive() + assert result.session_id.master_node_id == NodeId("A") + + em_in_tx.close() + cm_tx.close() + co_tx.close() + + +@pytest.mark.anyio +async def test_stale_message_gets_current_status_reply() -> None: + """ + Regression test for the wedged restarted master (issue #2197, part 1). + + A message below our clock (e.g. a restarted master campaigning from + clock 0) used to be dropped silently, leaving the sender unaware of the + cluster clock. We must reply with our current status instead. + """ + em_out_tx, em_out_rx = channel[ElectionMessage]() + em_in_tx, em_in_rx = channel[ElectionMessage]() + er_tx, er_rx = channel[ElectionResult]() + cm_tx, cm_rx = channel[ConnectionMessage]() + co_tx, co_rx = channel[ForwarderCommand]() + + election = Election( + node_id=NodeId("B"), + election_message_receiver=em_in_rx, + election_message_sender=em_out_tx, + election_result_sender=er_tx, + connection_message_receiver=cm_rx, + command_receiver=co_rx, + is_candidate=True, + ) + + async with create_task_group() as tg: + with fail_after(2): + tg.start_soon(election.run) + + # Establish a round at clock 5 and wait for it to resolve + await em_in_tx.send(em(clock=5, seniority=0, node_id="A")) + while True: + result = await er_rx.receive() + if result.won_clock == 5: + break + + # Drain campaign broadcasts already emitted + em_out_rx.collect() + + # A restarted node campaigns from clock 1, below the cluster clock + await em_in_tx.send(em(clock=1, seniority=0, node_id="R")) + + reply = await em_out_rx.receive() + assert reply.clock == 5 + assert reply.proposed_session.master_node_id == NodeId("B") + + em_in_tx.close() + cm_tx.close() + co_tx.close() + + +@pytest.mark.anyio +async def test_restarted_master_rejoins_cluster_deterministically() -> None: + """ + End-to-end regression for issue #2197: a node that restarts (losing its + election clock) while the cluster is at a higher clock must converge on + the same session as the surviving cluster, without relying on a + connection event. + """ + a_em_out_tx, a_em_out_rx = channel[ElectionMessage]() + a_em_in_tx, a_em_in_rx = channel[ElectionMessage]() + a_er_tx, a_er_rx = channel[ElectionResult]() + _a_cm_tx, a_cm_rx = channel[ConnectionMessage]() + _a_co_tx, a_co_rx = channel[ForwarderCommand]() + + r_em_out_tx, r_em_out_rx = channel[ElectionMessage]() + r_em_in_tx, r_em_in_rx = channel[ElectionMessage]() + r_er_tx, r_er_rx = channel[ElectionResult]() + _r_cm_tx, r_cm_rx = channel[ConnectionMessage]() + _r_co_tx, r_co_rx = channel[ForwarderCommand]() + + survivor = Election( + node_id=NodeId("A"), + election_message_receiver=a_em_in_rx, + election_message_sender=a_em_out_tx, + election_result_sender=a_er_tx, + connection_message_receiver=a_cm_rx, + command_receiver=a_co_rx, + is_candidate=True, + ) + restarted = Election( + node_id=NodeId("R"), + election_message_receiver=r_em_in_rx, + election_message_sender=r_em_out_tx, + election_result_sender=r_er_tx, + connection_message_receiver=r_cm_rx, + command_receiver=r_co_rx, + is_candidate=True, + ) + + async with create_task_group() as tg: + with fail_after(5): + tg.start_soon(survivor.run) + + # Advance the surviving cluster to clock 5 + await a_em_in_tx.send(em(clock=5, seniority=0, node_id="X")) + while True: + result = await a_er_rx.receive() + if result.won_clock == 5: + break + a_em_out_rx.collect() + + # Wire both nodes together, then boot the "restarted" node, + # which starts over from clock 0 + async def forward_a_to_r() -> None: + with a_em_out_rx as messages: + async for message in messages: + await r_em_in_tx.send(message) + + async def forward_r_to_a() -> None: + with r_em_out_rx as messages: + async for message in messages: + await a_em_in_tx.send(message) + + tg.start_soon(forward_a_to_r) + tg.start_soon(forward_r_to_a) + tg.start_soon(restarted.run) + + # The restarted node must deterministically reach the cluster + # clock (who wins the round depends on seniority tie-breaking, + # which is not what this regression is about) + while True: + result = await r_er_rx.receive() + if result.won_clock == 5: + break + assert restarted.clock == 5 + + # ... and both nodes must converge on a single session + while restarted.current_session != survivor.current_session: + await sleep(0.05) + + tg.cancel_scope.cancel()