fix(gateway): hydrate secondary-profile secrets off-loop at startup too (#99519 class)
_run_secondary_profile_reconnect now pre-hydrates external secret sources in a worker thread (PR #99519); _start_one_profile_adapters entered _profile_runtime_scope on the event loop three times per profile, each running the same synchronous network-bound hydration under _SECRET_SOURCE_CACHE_LOCK. Hydrate once via asyncio.to_thread and enter every scope in that method with hydrate_secrets=False. The reconnect test is parametrized over both entry points; the startup reconnect handoff waits are deadline-based since the runner now hops to a worker thread before publishing the replacement adapter. Co-authored-by: GoBeromsu <37897508+GoBeromsu@users.noreply.github.com>
This commit is contained in:
+10
-3
@@ -16992,8 +16992,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
) -> int:
|
||||
"""Create+connect one profile's adapters under its runtime scope."""
|
||||
from gateway.config import load_gateway_config
|
||||
from hermes_cli.env_loader import hydrate_profile_secret_sources
|
||||
|
||||
with _profile_runtime_scope(profile_home):
|
||||
# Hydrate external secret sources (1Password/vault/...) off-loop ONCE,
|
||||
# then enter the scope without re-hydrating: the sync hydration is
|
||||
# network-bound and would otherwise stall every other profile's
|
||||
# heartbeat while this one boots (same class as the reconnect path).
|
||||
await asyncio.to_thread(hydrate_profile_secret_sources, profile_home)
|
||||
|
||||
with _profile_runtime_scope(profile_home, hydrate_secrets=False):
|
||||
profile_runtime_cfg = _load_gateway_runtime_config()
|
||||
from hermes_cli.plugins import discover_plugins
|
||||
|
||||
@@ -17059,7 +17066,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
):
|
||||
continue
|
||||
try:
|
||||
with _profile_runtime_scope(profile_home):
|
||||
with _profile_runtime_scope(profile_home, hydrate_secrets=False):
|
||||
adapter = self._create_adapter(platform, platform_config)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
@@ -17143,7 +17150,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
self._configure_profile_adapter(adapter, profile_name, platform)
|
||||
|
||||
try:
|
||||
with _profile_runtime_scope(profile_home):
|
||||
with _profile_runtime_scope(profile_home, hydrate_secrets=False):
|
||||
success = await self._connect_initial_adapter_with_timeout(
|
||||
adapter, platform
|
||||
)
|
||||
|
||||
@@ -260,7 +260,11 @@ def _install_secondary_reconnect_context(
|
||||
|
||||
class TestSecondaryProfileFatalRecovery:
|
||||
@pytest.mark.asyncio
|
||||
async def test_reconnect_hydrates_secrets_off_the_event_loop(self, monkeypatch):
|
||||
@pytest.mark.parametrize("entry", ["startup", "reconnect"])
|
||||
async def test_secondary_hydrates_secrets_off_the_event_loop(self, monkeypatch, entry):
|
||||
"""#99519 class: both secondary entry points (initial start + reconnect)
|
||||
hydrate external secret sources in a worker thread, exactly once, and
|
||||
enter the runtime scope with hydration disabled."""
|
||||
runner = _secondary_recovery_runner()
|
||||
replacement = _SecondaryRecoveryAdapter()
|
||||
hydration_flags = []
|
||||
@@ -287,23 +291,30 @@ class TestSecondaryProfileFatalRecovery:
|
||||
ticks_during_hydration += 1
|
||||
await asyncio.sleep(0)
|
||||
|
||||
async def connect(adapter, platform, *, is_reconnect=False):
|
||||
async def connect(adapter, platform, **_kwargs):
|
||||
assert adapter is replacement
|
||||
assert platform is Platform.DISCORD
|
||||
assert is_reconnect is True
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.env_loader.hydrate_profile_secret_sources", slow_hydrate
|
||||
)
|
||||
monkeypatch.setattr(runner, "_connect_adapter_with_timeout", connect)
|
||||
monkeypatch.setattr(runner, "_connect_initial_adapter_with_timeout", connect)
|
||||
monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: {})
|
||||
monkeypatch.setattr(runner, "_snapshot_profile_busy_modes", lambda *a, **k: None)
|
||||
monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: None)
|
||||
if entry == "startup":
|
||||
coro = runner._start_one_profile_adapters(
|
||||
"reviewer", Path("/profiles/reviewer"), {}
|
||||
)
|
||||
else:
|
||||
coro = runner._run_secondary_profile_reconnect("reviewer", Platform.DISCORD)
|
||||
ticker_task = asyncio.create_task(ticker())
|
||||
reconnect_task = asyncio.create_task(
|
||||
runner._run_secondary_profile_reconnect("reviewer", Platform.DISCORD)
|
||||
)
|
||||
work = asyncio.create_task(coro)
|
||||
try:
|
||||
assert await asyncio.to_thread(hydration_started.wait, 1.0)
|
||||
await reconnect_task
|
||||
await work
|
||||
finally:
|
||||
stop_ticker.set()
|
||||
await ticker_task
|
||||
@@ -311,7 +322,7 @@ class TestSecondaryProfileFatalRecovery:
|
||||
assert len(hydration_thread_ids) == 1
|
||||
assert hydration_thread_ids[0] != loop_thread_id
|
||||
assert ticks_during_hydration > 0
|
||||
assert hydration_flags == [False]
|
||||
assert hydration_flags and set(hydration_flags) == {False}
|
||||
assert runner._profile_adapters["reviewer"][Platform.DISCORD] is replacement
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -447,13 +458,15 @@ class TestSecondaryStartupFailureRecovery:
|
||||
# gateway is already running) to the regular reconnect task, which
|
||||
# publishes the replacement and clears its own slot.
|
||||
await asyncio.wait_for(bridge[0], timeout=0.5)
|
||||
for _ in range(20):
|
||||
if (
|
||||
runner._profile_adapters.get("reviewer", {}).get(Platform.DISCORD)
|
||||
is replacement
|
||||
):
|
||||
break
|
||||
await asyncio.sleep(0)
|
||||
# The reconnect runner hops to a worker thread for secret hydration,
|
||||
# so wait on a deadline rather than a fixed number of loop turns.
|
||||
deadline = time.monotonic() + 1.0
|
||||
while (
|
||||
runner._profile_adapters.get("reviewer", {}).get(Platform.DISCORD)
|
||||
is not replacement
|
||||
and time.monotonic() < deadline
|
||||
):
|
||||
await asyncio.sleep(0.005)
|
||||
assert (
|
||||
runner._profile_adapters["reviewer"][Platform.DISCORD] is replacement
|
||||
)
|
||||
@@ -500,13 +513,15 @@ class TestSecondaryStartupFailureRecovery:
|
||||
bridge = list(runner._background_tasks)
|
||||
assert len(bridge) == 1
|
||||
await asyncio.wait_for(bridge[0], timeout=0.5)
|
||||
for _ in range(20):
|
||||
if (
|
||||
runner._profile_adapters.get("reviewer", {}).get(Platform.DISCORD)
|
||||
is replacement
|
||||
):
|
||||
break
|
||||
await asyncio.sleep(0)
|
||||
# The reconnect runner hops to a worker thread for secret hydration,
|
||||
# so wait on a deadline rather than a fixed number of loop turns.
|
||||
deadline = time.monotonic() + 1.0
|
||||
while (
|
||||
runner._profile_adapters.get("reviewer", {}).get(Platform.DISCORD)
|
||||
is not replacement
|
||||
and time.monotonic() < deadline
|
||||
):
|
||||
await asyncio.sleep(0.005)
|
||||
assert (
|
||||
runner._profile_adapters["reviewer"][Platform.DISCORD] is replacement
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user