root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resolve race condition in project-store-resolver breaking real-time dashboard updates

Browse Source 
Concurrent SSE and API requests for the same projectId both missed the
cache (storeCache.set ran after await store.watch()), creating independent
TaskStore instances with separate EventEmitters. SSE listeners attached
to one instance while mutations fired on the other, so no events reached
the browser.

Fix: add a pendingCreations promise map that deduplicates concurrent
calls, ensuring all callers share the same in-flight promise and thus
the same store instance. Also clear pendingCreations in evictProjectStore
and evictAllProjectStores. Adds a concurrent-call regression test.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-03 07:08:47 -07:00
parent df73a82917
commit bed9fcb378
2 changed files with 58 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
packages/dashboard/src/project-store-resolver.ts
View File

@@ -23,6 +23,13 @@ import type { TaskStore } from "@fusion/core";
*/
const storeCache = new Map<string, TaskStore>();
/**
* In-flight creation promises, keyed by projectId.
* Prevents concurrent requests from creating duplicate store instances
* before the first creation completes and is added to storeCache.
*/
const pendingCreations = new Map<string, Promise<TaskStore>>();
/**
* Track which stores have been fully initialized for real-time operation
* (watcher started). This prevents duplicate watch() calls on repeated
@@ -38,6 +45,11 @@ const initializedProjects = new Set<string>();
* engine agents) are detected and emitted as events.
* - Subsequent calls: returns the cached instance immediately.
*
* Concurrent calls for the same projectId are deduplicated via a pending
* promise map, preventing the race condition where the SSE endpoint and
* an API mutation request both miss the cache and create separate store
* instances with independent EventEmitters.
*
* @param projectId - The central-registry project ID
* @returns A shared TaskStore instance for this project
*/
@@ -47,19 +59,33 @@ export async function getOrCreateProjectStore(projectId: string): Promise<TaskSt
return cached;
}
const { TaskStore: TaskStoreClass } = await import("@fusion/core");
const store = await TaskStoreClass.getOrCreateForProject(projectId);
// Start watching for external changes (CLI, engine agents, etc.)
// so SSE listeners receive live events even when mutations happen
// outside this process.
if (!initializedProjects.has(projectId)) {
initializedProjects.add(projectId);
await store.watch();
// Deduplicate concurrent creation requests so SSE and API routes always
// share the same store instance even when both call this before the first
// creation completes.
const pending = pendingCreations.get(projectId);
if (pending) {
return pending;
}
storeCache.set(projectId, store);
return store;
const creation = (async () => {
const { TaskStore: TaskStoreClass } = await import("@fusion/core");
const store = await TaskStoreClass.getOrCreateForProject(projectId);
// Start watching for external changes (CLI, engine agents, etc.)
// so SSE listeners receive live events even when mutations happen
// outside this process.
if (!initializedProjects.has(projectId)) {
initializedProjects.add(projectId);
await store.watch();
}
storeCache.set(projectId, store);
pendingCreations.delete(projectId);
return store;
})();
pendingCreations.set(projectId, creation);
return creation;
}
/**
@@ -67,6 +93,7 @@ export async function getOrCreateProjectStore(projectId: string): Promise<TaskSt
* Useful for cleanup on project removal or server shutdown.
*/
export function evictProjectStore(projectId: string): void {
pendingCreations.delete(projectId);
const store = storeCache.get(projectId);
if (store) {
store.stopWatching();
@@ -80,6 +107,7 @@ export function evictProjectStore(projectId: string): void {
* Evict all cached stores. Used during server shutdown.
*/
export function evictAllProjectStores(): void {
pendingCreations.clear();
for (const projectId of storeCache.keys()) {
evictProjectStore(projectId);
}