diff --git a/repeater/handler_helpers/text.py b/repeater/handler_helpers/text.py index fb48000..d974dcf 100644 --- a/repeater/handler_helpers/text.py +++ b/repeater/handler_helpers/text.py @@ -201,7 +201,12 @@ class TextHelper: task.add_done_callback(_on_done) def register_identity( - self, name: str, identity, identity_type: str = "room_server", radio_config=None + self, + name: str, + identity, + identity_type: str = "room_server", + radio_config=None, + previous_name: str | None = None, ): hash_byte = identity.get_public_key()[0] @@ -265,7 +270,14 @@ class TextHelper: "Initialized CLI handler for repeater commands with identity and storage" ) - # Create RoomServer instance for room_server identities + # A re-registration (hot reload after a rename or key change) replaces + # a live RoomServer; take the old ones out so their sync loops can be + # stopped before the replacement starts, or a room pushes posts twice. + stale_rooms = [] + if identity_type == "room_server": + stale_rooms = self._take_stale_room_servers(hash_byte, {name, previous_name}) + + room_server = None if identity_type == "room_server" and self.sqlite_handler: try: from .room_server import MAX_UNSYNCED_POSTS @@ -296,37 +308,67 @@ class TextHelper: self.room_servers[hash_byte] = room_server - # Start sync loop — may be called from a non-async HTTP handler thread - try: - loop = asyncio.get_running_loop() - start_task = loop.create_task(room_server.start()) - self._track_task(start_task) - except RuntimeError: - # No running event loop in this thread - if self._loop and self._loop.is_running(): - future = asyncio.run_coroutine_threadsafe(room_server.start(), self._loop) - future.add_done_callback( - lambda f: ( - logger.error( - f"Room server '{name}' failed: {f.exception()}", - exc_info=f.exception(), - ) - if not f.cancelled() and f.exception() - else None - ) - ) - else: - logger.error(f"Cannot start room server '{name}': no event loop available") - logger.info( f"Registered room server '{name}': hash=0x{hash_byte:02X}, " f"max_posts={max_posts}" ) except Exception as e: + room_server = None logger.error(f"Failed to create room server '{name}': {e}", exc_info=True) + if stale_rooms or room_server: + self._schedule_room_swap(name, stale_rooms, room_server) + logger.info(f"Registered {identity_type} '{name}' text handler: hash=0x{hash_byte:02X}") + def _take_stale_room_servers(self, hash_byte: int, names: set) -> list: + """Remove and return the RoomServers a re-registration replaces. + + That is the one under ``hash_byte`` plus any registered under another + hash by the same (or previous) name, which is what a key change leaves. + """ + stale = [] + for room_hash, room in list(self.room_servers.items()): + if room_hash == hash_byte or room.room_name in names: + stale.append(self.room_servers.pop(room_hash)) + return stale + + async def _swap_room_servers(self, stale_rooms: list, room_server) -> None: + for old in stale_rooms: + try: + await old.stop() + except Exception as e: + logger.error(f"Error stopping replaced room server '{old.room_name}': {e}") + if room_server is not None: + await room_server.start() + + def _schedule_room_swap(self, name: str, stale_rooms: list, room_server) -> None: + """Stop ``stale_rooms`` then start ``room_server`` on the event loop. + + May be called from a non-async HTTP handler thread. + """ + coro = self._swap_room_servers(stale_rooms, room_server) + try: + loop = asyncio.get_running_loop() + self._track_task(loop.create_task(coro)) + except RuntimeError: + # No running event loop in this thread + if self._loop and self._loop.is_running(): + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + future.add_done_callback( + lambda f: ( + logger.error( + f"Room server '{name}' failed: {f.exception()}", + exc_info=f.exception(), + ) + if not f.cancelled() and f.exception() + else None + ) + ) + else: + coro.close() + logger.error(f"Cannot start room server '{name}': no event loop available") + def _client_by_pubkey(self, identity_hash: int, pubkey: bytes): """The ACL client for an already-authenticated public key. diff --git a/repeater/main.py b/repeater/main.py index 8454ac2..dddd54c 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -1415,7 +1415,12 @@ class RepeaterDaemon: logger.debug("Push trace data to companion: %s", e) def _register_identity_everywhere( - self, name: str, identity, config: dict, identity_type: str + self, + name: str, + identity, + config: dict, + identity_type: str, + previous_name: str | None = None, ) -> bool: """ Register an identity with the manager and all helpers in one place. @@ -1441,6 +1446,7 @@ class RepeaterDaemon: identity=identity, identity_type=identity_type, radio_config=self.config.get("radio", {}), + previous_name=previous_name, ) if self.protocol_request_helper: diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 0a9b971..0a4efe9 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -6704,6 +6704,7 @@ class APIEndpoints: identity=room_identity, config=identity, identity_type="room_server", + previous_name=name_s, ) if registration_success: logger.info( diff --git a/tests/test_text_helper_room_reregister.py b/tests/test_text_helper_room_reregister.py new file mode 100644 index 0000000..be2ac93 --- /dev/null +++ b/tests/test_text_helper_room_reregister.py @@ -0,0 +1,124 @@ +"""Re-registering a room identity (hot reload) must not leave a second sync loop.""" + +import asyncio +import threading +from unittest.mock import MagicMock + +import pytest + +from repeater.handler_helpers.text import TextHelper + + +class _FakeIdentity: + def __init__(self, pubkey: bytes): + self._pubkey = pubkey + + def get_public_key(self): + return self._pubkey + + +class _FakeACL: + def get_all_clients(self): + return [] + + +def _make_helper(acl_dict): + helper = TextHelper( + identity_manager=MagicMock(), + acl_dict=acl_dict, + sqlite_handler=MagicMock(), + ) + helper._loop = asyncio.get_running_loop() + return helper + + +def _running_sync_tasks(): + return [ + t + for t in asyncio.all_tasks() + if not t.done() and t.get_coro().__qualname__ == "RoomServer._sync_loop" + ] + + +async def _settle(helper): + for _ in range(5): + if helper._pending_tasks: + await asyncio.gather(*list(helper._pending_tasks), return_exceptions=True) + await asyncio.sleep(0) + + +@pytest.mark.asyncio +async def test_reregister_same_hash_leaves_one_sync_loop(): + old_acl, new_acl = _FakeACL(), _FakeACL() + acl_dict = {0x41: old_acl} + helper = _make_helper(acl_dict) + identity = _FakeIdentity(b"A" * 32) + + helper.register_identity("room", identity, "room_server") + await _settle(helper) + old_room = helper.room_servers[0x41] + assert len(_running_sync_tasks()) == 1 + + acl_dict[0x41] = new_acl + helper.register_identity("room-renamed", identity, "room_server", previous_name="room") + await _settle(helper) + + new_room = helper.room_servers[0x41] + assert new_room is not old_room + assert new_room.acl is new_acl + assert old_room._running is False + assert old_room._sync_task.done() + assert _running_sync_tasks() == [new_room._sync_task] + + await helper.cleanup() + + +@pytest.mark.asyncio +async def test_reregister_new_key_stops_room_under_old_hash(): + acl_dict = {0x41: _FakeACL(), 0x42: _FakeACL()} + helper = _make_helper(acl_dict) + + helper.register_identity("room", _FakeIdentity(b"A" * 32), "room_server") + await _settle(helper) + old_room = helper.room_servers[0x41] + + helper.register_identity( + "room-renamed", _FakeIdentity(b"B" * 32), "room_server", previous_name="room" + ) + await _settle(helper) + + assert set(helper.room_servers) == {0x42} + assert old_room._running is False + assert _running_sync_tasks() == [helper.room_servers[0x42]._sync_task] + + await helper.cleanup() + + +@pytest.mark.asyncio +async def test_reregister_from_request_thread_stops_old_before_starting_new(): + acl_dict = {0x41: _FakeACL()} + helper = _make_helper(acl_dict) + identity = _FakeIdentity(b"A" * 32) + + helper.register_identity("room", identity, "room_server") + await _settle(helper) + old_room = helper.room_servers[0x41] + + # CherryPy request threads have no running loop: the swap is handed to + # helper._loop with run_coroutine_threadsafe. + thread = threading.Thread( + target=helper.register_identity, args=("room", identity, "room_server") + ) + thread.start() + await asyncio.to_thread(thread.join) + for _ in range(20): + if helper.room_servers[0x41]._sync_task is not None: + break + await asyncio.sleep(0.01) + + new_room = helper.room_servers[0x41] + assert new_room is not old_room + assert old_room._running is False + assert _running_sync_tasks() == [new_room._sync_task] + + await helper.cleanup()