fix(gateway): track deferred-platform load ownership per flow, not thread
A plugin's register() runs on a plugin-load deadline worker. When it re-walks the platform registry (i18n/YAML layering -> plugin_entries()), _resolve sees its own parent's in-flight loader, but ownership was keyed on threading.get_ident() -- the parent's thread, not the worker -- so the nested walk event.wait()ed on itself until the deadline abandoned the load (#130575). Record in-flight load keys in a per-registry ContextVar instead. Deadline workers start from copy_context(), so a nested walk on the parent's worker takes the existing "recursively requested while loading" return, while unrelated concurrent loaders (fresh contexts) still wait on the event.
This commit is contained in:
@@ -7,6 +7,7 @@ registry first, then the legacy built-in path. Plugin side: ``platform_registry
|
||||
.register(PlatformEntry(...))``; gateway side: ``create_adapter("irc", platform_config)``.
|
||||
"""
|
||||
|
||||
import contextvars
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
@@ -126,7 +127,11 @@ class PlatformRegistry:
|
||||
self._scoped_deferred: dict[str, dict[str, _Loader]] = {}
|
||||
self._inflight: dict[_LoadKey, threading.Event] = {}
|
||||
self._inflight_loaders: dict[_LoadKey, _Loader] = {}
|
||||
self._inflight_owners: dict[_LoadKey, int] = {}
|
||||
# Load keys whose loader is running in the current flow; copied into plugin deadline
|
||||
# workers, so a nested walk from register() sees its own parent's load as recursive.
|
||||
self._loading: contextvars.ContextVar[frozenset[_LoadKey]] = contextvars.ContextVar(
|
||||
f"platform_registry_loading_{id(self)}", default=frozenset()
|
||||
)
|
||||
self._cancelled_inflight: set[_LoadKey] = set()
|
||||
# A failed loader is no longer discoverable, but its identity remains
|
||||
# until ownership teardown can CAS-restore the displaced predecessor.
|
||||
@@ -224,11 +229,10 @@ class PlatformRegistry:
|
||||
event = threading.Event()
|
||||
self._inflight[load_key] = event
|
||||
self._inflight_loaders[load_key] = loader
|
||||
self._inflight_owners[load_key] = threading.get_ident()
|
||||
is_loader = True
|
||||
if event is None:
|
||||
return
|
||||
if not is_loader and self._inflight_owners.get(load_key) == threading.get_ident():
|
||||
if not is_loader and load_key in self._loading.get():
|
||||
logger.warning("Deferred platform '%s' recursively requested while loading", name)
|
||||
return
|
||||
if not is_loader:
|
||||
@@ -237,11 +241,13 @@ class PlatformRegistry:
|
||||
# we waited for; resolve that predecessor instead of a one-shot false negative.
|
||||
self._resolve(name, active_scope)
|
||||
return
|
||||
token = self._loading.set(self._loading.get() | {load_key})
|
||||
try:
|
||||
loader()
|
||||
except Exception as e:
|
||||
logger.warning("Deferred load of platform '%s' failed: %s", name, e, exc_info=True)
|
||||
finally:
|
||||
self._loading.reset(token)
|
||||
with self._lock:
|
||||
was_cancelled = load_key in self._cancelled_inflight
|
||||
entries, deferred = self._scope_maps(load_key[0])
|
||||
@@ -249,7 +255,6 @@ class PlatformRegistry:
|
||||
self._consumed_loaders[load_key] = loader
|
||||
self._inflight.pop(load_key, None)
|
||||
self._inflight_loaders.pop(load_key, None)
|
||||
self._inflight_owners.pop(load_key, None)
|
||||
self._cancelled_inflight.discard(load_key)
|
||||
event.set()
|
||||
if was_cancelled:
|
||||
|
||||
@@ -677,3 +677,26 @@ class TestMigratedPlatformWiring:
|
||||
"probe and the active installer — status displays would "
|
||||
"pip-install as a side effect"
|
||||
)
|
||||
|
||||
|
||||
def test_registry_walk_from_plugin_load_worker_does_not_wait_on_parent(monkeypatch):
|
||||
"""register() re-walking the registry on its deadline worker must not block on its own load."""
|
||||
from hermes_cli import plugins_loader
|
||||
|
||||
monkeypatch.setattr(plugins_loader, "_resolve_plugin_load_timeout", lambda: 0.5)
|
||||
reg = PlatformRegistry()
|
||||
abandoned = []
|
||||
|
||||
class Ctx:
|
||||
def _abandon_load(self):
|
||||
abandoned.append(True)
|
||||
|
||||
def register():
|
||||
reg.plugin_entries()
|
||||
reg.register(PlatformEntry(name="reentrant", label="Re-entrant", adapter_factory=lambda cfg: None,
|
||||
check_fn=lambda: True, source="plugin"))
|
||||
|
||||
reg.register_deferred("reentrant", lambda: plugins_loader.run_with_load_deadline("reentrant", Ctx(), register))
|
||||
|
||||
assert [e.name for e in reg.plugin_entries()] == ["reentrant"]
|
||||
assert abandoned == []
|
||||
|
||||
Reference in New Issue
Block a user