apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-01 09:29:38 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix: prevent dispatcher race condition via promise-based lazy init

Browse source 
Store the initialization Promise in the requestors map synchronously
before yielding, so concurrent callers for the same key await the same
instance — prevents orphaned RequestResponse objects and duplicate NATS
subscriptions. Mirrors upstream fix 8f18ba02.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
elpresidank 2026-04-07 12:11:11 -05:00
parent 2a2e8e76a3
commit 580ee319a3
1 changed files with 18 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

32
ts/packages/flow/src/gateway/dispatch/manager.ts
View file

@ -67,7 +67,7 @@ function topicName(name: string): string {
export class DispatcherManager {
private readonly pubsub: PubSubBackend;
private requestors = new Map<string, RequestResponse<unknown, unknown>>();
private requestors = new Map<string, Promise<RequestResponse<unknown, unknown>>>();
constructor(config: GatewayConfig) {
this.pubsub = new NatsBackend(config.natsUrl ?? "nats://localhost:4222");
@ -78,7 +78,8 @@ export class DispatcherManager {
}
async stop(): Promise<void> {
for (const rr of this.requestors.values()) {
for (const pending of this.requestors.values()) {
const rr = await pending;
await rr.stop();
}
await this.pubsub.close();
@ -86,23 +87,26 @@ export class DispatcherManager {
// ---------- Internal helpers ----------
private async getRequestor(
private getRequestor(
requestTopic: string,
responseTopic: string,
key: string,
): Promise<RequestResponse<unknown, unknown>> {
let rr = this.requestors.get(key);
if (!rr) {
rr = new RequestResponse({
pubsub: this.pubsub,
requestTopic,
responseTopic,
subscription: `gateway-${key}`,
});
await rr.start();
this.requestors.set(key, rr);
let pending = this.requestors.get(key);
if (!pending) {
pending = (async () => {
const rr = new RequestResponse({
pubsub: this.pubsub,
requestTopic,
responseTopic,
subscription: `gateway-${key}`,
});
await rr.start();
return rr;
})();
this.requestors.set(key, pending);
}
return rr;
return pending;
}
private resolveGlobalTopics(