mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(channels): serialize lifecycle mutations under per-channel lock (#6304)
* fix(channels): serialize lifecycle mutations under per-channel lock * fix(channels): invalidate stale config snapshots across lifecycle mutations under per-channel lock
This commit is contained in:
Shxiao
GitHub
parent
c0895d295b
commit
b85e4ae74b
@ -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()
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user