apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-12 14:52:11 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix: prevent duplicate dispatcher creation race condition in invoke_global_service

Browse source 
Concurrent coroutines could all pass the `if key in self.dispatchers` check
before any of them wrote the result back, because `await dispatcher.start()`
yields to the event loop. This caused multiple Pulsar consumers to be created
on the same shared subscription, distributing responses round-robin and
dropping ~2/3 of them — manifesting as a permanent spinner in the Workbench UI.

Apply a double-checked asyncio.Lock in both `invoke_global_service` and
`invoke_flow_service` so only one dispatcher is ever created per service key.
This commit is contained in:
Sreeram Venkatasubramanian 2026-03-26 14:22:38 +05:30
parent 3ccff800c7
commit 928bda41ae
1 changed files with 46 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

94
trustgraph-flow/trustgraph/gateway/dispatch/manager.py
View file

@ -116,6 +116,7 @@ class DispatcherManager:
self.flows = {} self.flows = {}
self.dispatchers = {} self.dispatchers = {}
self.dispatcher_lock = asyncio.Lock()
async def start_flow(self, id, flow): async def start_flow(self, id, flow):
logger.info(f"Starting flow {id}") logger.info(f"Starting flow {id}")
@ -163,30 +164,28 @@ class DispatcherManager:
key = (None, kind) key = (None, kind)
if key in self.dispatchers: if key not in self.dispatchers:
return await self.dispatchers[key].process(data, responder) async with self.dispatcher_lock:
if key not in self.dispatchers:
request_queue = None
response_queue = None
if kind in self.queue_overrides:
request_queue = self.queue_overrides[kind].get("request")
response_queue = self.queue_overrides[kind].get("response")
# Get queue overrides if specified for this service dispatcher = global_dispatchers[kind](
request_queue = None backend = self.backend,
response_queue = None timeout = 120,
if kind in self.queue_overrides: consumer = f"{self.prefix}-{kind}-request",
request_queue = self.queue_overrides[kind].get("request") subscriber = f"{self.prefix}-{kind}-request",
response_queue = self.queue_overrides[kind].get("response") request_queue = request_queue,
response_queue = response_queue,
)
dispatcher = global_dispatchers[kind]( await dispatcher.start()
backend = self.backend, self.dispatchers[key] = dispatcher
timeout = 120,
consumer = f"{self.prefix}-{kind}-request",
subscriber = f"{self.prefix}-{kind}-request",
request_queue = request_queue,
response_queue = response_queue,
)
await dispatcher.start() return await self.dispatchers[key].process(data, responder)
self.dispatchers[key] = dispatcher
return await dispatcher.process(data, responder)
def dispatch_flow_import(self): def dispatch_flow_import(self):
return self.process_flow_import return self.process_flow_import
@ -297,36 +296,35 @@ class DispatcherManager:
key = (flow, kind) key = (flow, kind)
if key in self.dispatchers: if key not in self.dispatchers:
return await self.dispatchers[key].process(data, responder) async with self.dispatcher_lock:
if key not in self.dispatchers:
intf_defs = self.flows[flow]["interfaces"]
intf_defs = self.flows[flow]["interfaces"] if kind not in intf_defs:
raise RuntimeError("This kind not supported by flow")
if kind not in intf_defs: qconfig = intf_defs[kind]
raise RuntimeError("This kind not supported by flow")
qconfig = intf_defs[kind] if kind in request_response_dispatchers:
dispatcher = request_response_dispatchers[kind](
backend = self.backend,
request_queue = qconfig["request"],
response_queue = qconfig["response"],
timeout = 120,
consumer = f"{self.prefix}-{flow}-{kind}-request",
subscriber = f"{self.prefix}-{flow}-{kind}-request",
)
elif kind in sender_dispatchers:
dispatcher = sender_dispatchers[kind](
backend = self.backend,
queue = qconfig,
)
else:
raise RuntimeError("Invalid kind")
if kind in request_response_dispatchers: await dispatcher.start()
dispatcher = request_response_dispatchers[kind]( self.dispatchers[key] = dispatcher
backend = self.backend,
request_queue = qconfig["request"],
response_queue = qconfig["response"],
timeout = 120,
consumer = f"{self.prefix}-{flow}-{kind}-request",
subscriber = f"{self.prefix}-{flow}-{kind}-request",
)
elif kind in sender_dispatchers:
dispatcher = sender_dispatchers[kind](
backend = self.backend,
queue = qconfig,
)
else:
raise RuntimeError("Invalid kind")
await dispatcher.start()
self.dispatchers[key] = dispatcher return await self.dispatchers[key].process(data, responder)
return await dispatcher.process(data, responder)