fix: stop replaced RoomServer sync loops on identity re-registration

A hot reload from PUT /api/update_identity re-runs TextHelper.register_identity,
which replaced self.room_servers[hash] and started a new sync loop without
stopping the old one, so the room pushed posts twice (the old loop with its
stale ACL). Take the RoomServer under the same hash, and any registered under
another hash by the same or previous name (a key change), out of room_servers,
and stop them before starting the replacement, on the running loop or via
run_coroutine_threadsafe from a request thread. update_identity passes the
original name through _register_identity_everywhere as previous_name.
This commit is contained in:
agessaman committed 2026-09-26 14:23:40 -07:00
1 parent 13eb8b2ea8
commit 5de9f0358a
4 files changed
+198 -25

No files matched your search

+66 -24
View File
@@ -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.
+7 -1
View File
@@ -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:
+1
View File
@@ -6704,6 +6704,7 @@ class APIEndpoints:
identity=room_identity,
config=identity,
identity_type="room_server",
previous_name=name_s,
)
if registration_success:
logger.info(
+124
View File
@@ -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()