diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index 3ad0aa2b95..11f15d2111 100644 --- a/backend/app/channels/service.py +++ b/backend/app/channels/service.py @@ -161,6 +161,7 @@ class ChannelService: self._config = config self._running = False self._readiness_locks: dict[str, asyncio.Lock] = {} + self._config_epochs: dict[str, int] = {} @classmethod def from_app_config( @@ -239,13 +240,19 @@ class ChannelService: logger.warning("ChannelService is not running; cannot ensure channel readiness") return False - if config is not None: - self._config[name] = dict(config) + epoch = self._config_epochs.get(name, 0) + async with self._channel_lock(name): + if not self._running: + return False - # Serialize per channel: readiness is polled from request handlers, so - # concurrent calls must not stop/start the same channel worker twice. - lock = self._readiness_locks.setdefault(name, asyncio.Lock()) - async with lock: + if config is not None: + if self._config_epochs.get(name, 0) == epoch: + self._config[name] = dict(config) + else: + logger.info( + "Ignoring stale config snapshot for channel %s after concurrent reconfiguration", + name, + ) channel_config = self._config.get(name) if not channel_config or not isinstance(channel_config, dict): logger.warning("No config for requested channel") @@ -270,6 +277,8 @@ class ChannelService: max_attempts = max(1, attempts) for attempt in range(max_attempts): + if not self._running: + return False if attempt > 0: logger.info("Retrying channel startup after readiness check") if await self._start_channel(name, channel_config): @@ -343,8 +352,25 @@ class ChannelService: logger.exception("Failed to reload config for channel %s, using cached version", name) return self._config.get(name) + def _channel_lock(self, name: str) -> asyncio.Lock: + """Return the per-channel lifecycle lock ensuring serialized mutations.""" + return self._readiness_locks.setdefault(name, asyncio.Lock()) + + def _bump_config_epoch(self, name: str) -> int: + """Bump and return the configuration epoch for a channel. + + Invalidates queued readiness checks carrying an earlier config snapshot. + """ + self._config_epochs[name] = self._config_epochs.get(name, 0) + 1 + return self._config_epochs[name] + async def restart_channel(self, name: str, *, reload_config: bool = True) -> bool: """Restart a specific channel. Returns True if successful.""" + async with self._channel_lock(name): + return await self._restart_channel_locked(name, reload_config=reload_config) + + async def _restart_channel_locked(self, name: str, *, reload_config: bool = True) -> bool: + """Restart a specific channel. The caller must already hold ``_channel_lock(name)``.""" if name in self._channels: channel = self._channels[name] # Same ownership rule as readiness retries: retain an instance @@ -359,6 +385,7 @@ class ChannelService: # Reading config.yaml and the runtime store is disk IO; keep it # off the event loop. config = await asyncio.to_thread(self._load_channel_config, name) + self._bump_config_epoch(name) else: config = self._config.get(name) if not config or not isinstance(config, dict): @@ -373,29 +400,39 @@ class ChannelService: async def configure_channel(self, name: str, config: dict[str, Any]) -> bool: """Apply runtime config for a channel and restart it if the service is running.""" - self._config[name] = dict(config) - if not self._running: - return True - # The caller just supplied the authoritative config (e.g. credentials - # entered in the browser that are never written to config.yaml) — a - # file reload here would clobber it with the stale on-disk entry. - return await self.restart_channel(name, reload_config=False) + async with self._channel_lock(name): + self._bump_config_epoch(name) + try: + self._config[name] = dict(config) + if not self._running: + return True + # The caller just supplied the authoritative config (e.g. credentials + # entered in the browser that are never written to config.yaml) — a + # file reload here would clobber it with the stale on-disk entry. + return await self._restart_channel_locked(name, reload_config=False) + finally: + self._bump_config_epoch(name) async def remove_channel(self, name: str) -> bool: """Remove runtime config for a channel and stop it if currently running.""" - self._config.pop(name, None) - channel = self._channels.get(name) - if channel is None: - return True - # Stop-then-drop with the shared ownership rule: a channel whose - # stop() fails stays tracked (and returns False) instead of being - # popped first and leaking its subscribed listener on failure. - await self._stop_and_discard_channel(name, channel) - if self._channels.get(name) is channel: - logger.warning("Removal incomplete: %s channel failed to stop and remains tracked", name) - return False - logger.info("Channel stopped and removed") - return True + async with self._channel_lock(name): + self._bump_config_epoch(name) + try: + self._config.pop(name, None) + channel = self._channels.get(name) + if channel is None: + return True + # Stop-then-drop with the shared ownership rule: a channel whose + # stop() fails stays tracked (and returns False) instead of being + # popped first and leaking its subscribed listener on failure. + await self._stop_and_discard_channel(name, channel) + if self._channels.get(name) is channel: + logger.warning("Removal incomplete: %s channel failed to stop and remains tracked", name) + return False + logger.info("Channel stopped and removed") + return True + finally: + self._bump_config_epoch(name) async def _stop_and_discard_channel(self, name: str, channel: Channel) -> None: """Stop a channel and drop it only once its ``stop()`` has completed. @@ -421,8 +458,10 @@ class ChannelService: resources nobody can clean up anymore. Callers check for retention (``self._channels.get(name) is channel``) and defer starting or removing a replacement for that round, so startup cannot silently - overwrite a still-listening retained instance; ``ensure_channel_ready`` - additionally serializes on the per-channel readiness lock. + overwrite a still-listening retained instance; all per-channel + lifecycle mutations (``ensure_channel_ready``, ``restart_channel``, + ``configure_channel``, ``remove_channel``) additionally serialize on + the per-channel lock. """ try: await channel.stop() diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index f6748e1c09..0daf7bb540 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -8760,6 +8760,186 @@ class TestChannelService: assert stopped # old channel was stopped assert not started # _start_channel was NOT called + def test_concurrent_restart_channel_serializes(self): + """Concurrent restart_channel calls serialize per channel rather than racing stop/start.""" + from app.channels.service import ChannelService + + service = ChannelService(channels_config={"telegram": {"enabled": True, "token": "test"}}) + service._running = True + + class FakeChannel: + def __init__(self): + self.is_running = True + self.stop_count = 0 + + async def stop(self): + self.stop_count += 1 + self.is_running = False + + current_instance = FakeChannel() + service._channels["telegram"] = current_instance + + starts = 0 + active_starts = 0 + max_concurrent_starts = 0 + + async def mock_start(name, config): + nonlocal starts, active_starts, max_concurrent_starts + active_starts += 1 + max_concurrent_starts = max(max_concurrent_starts, active_starts) + await asyncio.sleep(0.02) + service._channels[name] = FakeChannel() + active_starts -= 1 + starts += 1 + return True + + service._start_channel = mock_start + + async def go(): + results = await asyncio.gather( + service.restart_channel("telegram", reload_config=False), + service.restart_channel("telegram", reload_config=False), + ) + assert all(results) + + _run(go()) + + assert max_concurrent_starts == 1 + assert starts == 2 + + def test_concurrent_ensure_channel_ready_and_restart(self): + """ensure_channel_ready and restart_channel serialize without collision.""" + from app.channels.service import ChannelService + + service = ChannelService(channels_config={"telegram": {"enabled": True, "token": "test"}}) + service._running = True + + class FakeChannel: + def __init__(self): + self.is_running = True + + async def stop(self): + # Suspend in stop to yield control to concurrent coroutines + await asyncio.sleep(0.02) + self.is_running = False + + service._channels["telegram"] = FakeChannel() + + starts = 0 + active_starts = 0 + max_concurrent_starts = 0 + + async def mock_start(name, config): + nonlocal starts, active_starts, max_concurrent_starts + active_starts += 1 + max_concurrent_starts = max(max_concurrent_starts, active_starts) + await asyncio.sleep(0.03) + service._channels[name] = FakeChannel() + active_starts -= 1 + starts += 1 + return True + + service._start_channel = mock_start + + async def go(): + # restart_channel suspends in stop(), so ensure_channel_ready must wait on + # _channel_lock rather than observing is_running=False and racing _start_channel. + results = await asyncio.gather( + service.restart_channel("telegram", reload_config=False), + service.ensure_channel_ready("telegram"), + ) + assert all(results) + + _run(go()) + + assert max_concurrent_starts == 1 + assert starts == 1 + + def test_concurrent_remove_and_ensure_channel_ready(self): + """remove_channel and ensure_channel_ready serialize cleanly without zombie respawns.""" + from app.channels.service import ChannelService + + service = ChannelService(channels_config={"telegram": {"enabled": True, "token": "test"}}) + service._running = True + + class FakeChannel: + def __init__(self): + self.is_running = True + + async def stop(self): + await asyncio.sleep(0.02) + self.is_running = False + + service._channels["telegram"] = FakeChannel() + + starts = 0 + + async def mock_start(name, config): + nonlocal starts + starts += 1 + await asyncio.sleep(0.02) + service._channels[name] = FakeChannel() + return True + + service._start_channel = mock_start + + async def go(): + # Pass config to ensure_channel_ready (the router pattern) while remove_channel is in flight. + results = await asyncio.gather( + service.remove_channel("telegram"), + service.ensure_channel_ready("telegram", config={"enabled": True, "token": "test"}), + ) + return results + + results = _run(go()) + # remove_channel succeeds; ensure_channel_ready serialized after it returns False + # because the channel was removed and the stale pre-wait config snapshot is discarded. + assert results == [True, False] + assert starts == 0 + assert "telegram" not in service._channels + assert "telegram" not in service._config + + def test_concurrent_configure_and_ensure_channel_ready_keeps_authoritative_config(self): + """ensure_channel_ready does not overwrite authoritative credentials from configure_channel.""" + from app.channels.service import ChannelService + + service = ChannelService(channels_config={"telegram": {"enabled": True, "token": "old"}}) + service._running = True + + class FakeChannel: + def __init__(self): + self.is_running = True + + async def stop(self): + await asyncio.sleep(0.01) + self.is_running = False + + service._channels["telegram"] = FakeChannel() + + started_configs = [] + + async def mock_start(name, config): + started_configs.append(dict(config)) + await asyncio.sleep(0.02) + service._channels[name] = FakeChannel() + return True + + service._start_channel = mock_start + + async def go(): + # configure_channel applies new credentials. A concurrent readiness check + # carrying a stale snapshot must not revert self._config to "stale". + results = await asyncio.gather( + service.configure_channel("telegram", {"enabled": True, "token": "new"}), + service.ensure_channel_ready("telegram", config={"enabled": True, "token": "stale"}), + ) + return results + + results = _run(go()) + assert results == [True, True] + assert service._config["telegram"]["token"] == "new" + assert all(c["token"] == "new" for c in started_configs) + # --------------------------------------------------------------------------- # Slack send retry tests