feat(engine): add background project engine reconciliation + global concurrency setting
- ProjectEngineManager.startReconciliation() polls for newly registered projects every 30s and starts their engines without requiring UI access - Expose global concurrency limit in dashboard settings - Fix SettingsModal test cleanup Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
5
.changeset/expose-global-concurrency-setting.md
Normal file
5
.changeset/expose-global-concurrency-setting.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@gsxdsm/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Expose the system-wide global concurrency limit in dashboard settings.
|
||||||
11
.changeset/project-engine-reconciliation.md
Normal file
11
.changeset/project-engine-reconciliation.md
Normal file
@@ -0,0 +1,11 @@
|
|||||||
|
---
|
||||||
|
"@gsxdsm/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Decouple project execution startup from dashboard page access
|
||||||
|
|
||||||
|
Previously, project task execution (triage, scheduling, in-progress transitions) would only start for projects registered after boot if a user navigated to the Projects view in the dashboard UI, which triggered the `onProjectFirstAccessed` lazy-start callback.
|
||||||
|
|
||||||
|
This fix adds a background reconciliation loop in `ProjectEngineManager` that periodically checks for newly registered projects and starts their engines without requiring any UI access. Projects registered via `fn project add` after the dashboard or headless node starts will now have their engines started automatically within the next reconciliation interval (30 seconds by default).
|
||||||
|
|
||||||
|
The `onProjectFirstAccessed` callback remains wired as a fast-path fallback for potential optimization, but it is no longer required for correctness.
|
||||||
@@ -38,8 +38,6 @@ vi.mock("../../api", () => ({
|
|||||||
fetchSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
fetchSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
||||||
updateSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
updateSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
||||||
updateGlobalSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
updateGlobalSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
||||||
fetchGlobalConcurrency: vi.fn(() => Promise.resolve({ globalMaxConcurrent: 4, currentlyActive: 0, queuedCount: 0, projectsActive: {} })),
|
|
||||||
updateGlobalConcurrency: vi.fn(() => Promise.resolve({ globalMaxConcurrent: 4, currentlyActive: 0, queuedCount: 0, projectsActive: {} })),
|
|
||||||
fetchAuthStatus: vi.fn(() => Promise.resolve({ providers: [{ id: "anthropic", name: "Anthropic", authenticated: false }] })),
|
fetchAuthStatus: vi.fn(() => Promise.resolve({ providers: [{ id: "anthropic", name: "Anthropic", authenticated: false }] })),
|
||||||
loginProvider: vi.fn(() => Promise.resolve({ url: "https://auth.example.com/login" })),
|
loginProvider: vi.fn(() => Promise.resolve({ url: "https://auth.example.com/login" })),
|
||||||
logoutProvider: vi.fn(() => Promise.resolve({ success: true })),
|
logoutProvider: vi.fn(() => Promise.resolve({ success: true })),
|
||||||
@@ -66,8 +64,6 @@ vi.mock("../../api", () => ({
|
|||||||
fetchPluginSettings: vi.fn(() => Promise.resolve({})),
|
fetchPluginSettings: vi.fn(() => Promise.resolve({})),
|
||||||
updatePluginSettings: vi.fn(() => Promise.resolve({})),
|
updatePluginSettings: vi.fn(() => Promise.resolve({})),
|
||||||
reloadPlugin: vi.fn(() => Promise.resolve({ id: "test-plugin", name: "Test Plugin", version: "1.0.0", state: "started" as const, enabled: true, settings: {}, settingsSchema: {} })),
|
reloadPlugin: vi.fn(() => Promise.resolve({ id: "test-plugin", name: "Test Plugin", version: "1.0.0", state: "started" as const, enabled: true, settings: {}, settingsSchema: {} })),
|
||||||
fetchGlobalConcurrency: vi.fn(() => Promise.resolve({ globalMaxConcurrent: 4, currentlyActive: 0, queuedCount: 0, projectsActive: {} })),
|
|
||||||
updateGlobalConcurrency: vi.fn(() => Promise.resolve({ globalMaxConcurrent: 4, currentlyActive: 0, queuedCount: 0, projectsActive: {} })),
|
|
||||||
fetchBackups: vi.fn(() => Promise.resolve({ backups: [], totalSize: 0 })),
|
fetchBackups: vi.fn(() => Promise.resolve({ backups: [], totalSize: 0 })),
|
||||||
createBackup: vi.fn(() => Promise.resolve({ success: true })),
|
createBackup: vi.fn(() => Promise.resolve({ success: true })),
|
||||||
exportSettings: vi.fn(() => Promise.resolve({ version: 1, exportedAt: new Date().toISOString(), global: undefined, project: {} })),
|
exportSettings: vi.fn(() => Promise.resolve({ version: 1, exportedAt: new Date().toISOString(), global: undefined, project: {} })),
|
||||||
|
|||||||
@@ -37,7 +37,6 @@ vi.mock("../../api", () => ({
|
|||||||
fetchSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
fetchSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
||||||
updateSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
updateSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
||||||
updateGlobalSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
updateGlobalSettings: vi.fn(() => Promise.resolve({ ...defaultSettings })),
|
||||||
fetchGlobalConcurrency: vi.fn(() => Promise.resolve({ globalMaxConcurrent: 4, currentlyActive: 0, queuedCount: 0, projectsActive: {} })),
|
|
||||||
fetchAuthStatus: vi.fn(() => Promise.resolve({ providers: [{ id: "anthropic", name: "Anthropic", authenticated: false }] })),
|
fetchAuthStatus: vi.fn(() => Promise.resolve({ providers: [{ id: "anthropic", name: "Anthropic", authenticated: false }] })),
|
||||||
loginProvider: vi.fn(() => Promise.resolve({ url: "https://auth.example.com/login" })),
|
loginProvider: vi.fn(() => Promise.resolve({ url: "https://auth.example.com/login" })),
|
||||||
logoutProvider: vi.fn(() => Promise.resolve({ success: true })),
|
logoutProvider: vi.fn(() => Promise.resolve({ success: true })),
|
||||||
|
|||||||
@@ -145,8 +145,14 @@ export interface ServerOptions {
|
|||||||
chatManager?: import("./chat.js").ChatManager;
|
chatManager?: import("./chat.js").ChatManager;
|
||||||
/**
|
/**
|
||||||
* Called once when a secondary project (identified by projectId query param)
|
* Called once when a secondary project (identified by projectId query param)
|
||||||
* is first accessed via a project-scoped API or SSE request. Use this to
|
* is first accessed via a project-scoped API or SSE request.
|
||||||
* lazily start an engine for that project.
|
*
|
||||||
|
* @deprecated This callback is a fast-path fallback for immediate engine
|
||||||
|
* startup on project access. ProjectEngineManager.startReconciliation() is
|
||||||
|
* the primary mechanism for ensuring all registered projects have engines
|
||||||
|
* started — it runs without requiring any UI or API access. This callback
|
||||||
|
* is NOT required for correctness; it only provides a potential optimization
|
||||||
|
* for projects that are accessed before the next reconciliation tick.
|
||||||
*/
|
*/
|
||||||
onProjectFirstAccessed?: (projectId: string) => void;
|
onProjectFirstAccessed?: (projectId: string) => void;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||||
|
|
||||||
// Mock ProjectEngine before importing the manager
|
// Mock ProjectEngine before importing the manager
|
||||||
vi.mock("../project-engine.js", () => {
|
vi.mock("../project-engine.js", () => {
|
||||||
@@ -313,4 +313,224 @@ describe("ProjectEngineManager", () => {
|
|||||||
expect(ProjectEngine).not.toHaveBeenCalled();
|
expect(ProjectEngine).not.toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("startReconciliation / stopReconciliation", () => {
|
||||||
|
beforeEach(() => {
|
||||||
|
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||||
|
});
|
||||||
|
|
||||||
|
afterEach(() => {
|
||||||
|
vi.useRealTimers();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("starts reconciliation and detects new projects on interval", async () => {
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
|
||||||
|
// Start reconciliation with a short interval
|
||||||
|
manager.startReconciliation(1000);
|
||||||
|
|
||||||
|
// Advance time to trigger the first reconciliation tick
|
||||||
|
await vi.advanceTimersByTimeAsync(1000);
|
||||||
|
|
||||||
|
// Should have started engines for all registered projects
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
expect(manager.getEngine("proj_bbb")).toBeDefined();
|
||||||
|
expect(manager.getEngine("proj_ccc")).toBeDefined();
|
||||||
|
|
||||||
|
manager.stopReconciliation();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("runs an immediate reconciliation tick on startReconciliation", async () => {
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
|
||||||
|
// Clear any previous calls
|
||||||
|
vi.clearAllMocks();
|
||||||
|
|
||||||
|
// Start reconciliation - should run immediately
|
||||||
|
manager.startReconciliation(60000);
|
||||||
|
|
||||||
|
// Don't advance time - the immediate tick should have run
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
manager.stopReconciliation();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("starts engines for newly registered projects on reconciliation tick", async () => {
|
||||||
|
// Create a manager with no initial projects
|
||||||
|
const projectMap = new Map<string, RegisteredProject>();
|
||||||
|
const emptyCentralCore = {
|
||||||
|
init: vi.fn().mockResolvedValue(undefined),
|
||||||
|
close: vi.fn().mockResolvedValue(undefined),
|
||||||
|
listProjects: vi.fn().mockResolvedValue([]),
|
||||||
|
getProject: vi.fn().mockImplementation((id: string) =>
|
||||||
|
Promise.resolve(projectMap.get(id) ?? null),
|
||||||
|
),
|
||||||
|
getProjectByPath: vi.fn().mockResolvedValue(null),
|
||||||
|
} as unknown as CentralCore;
|
||||||
|
const manager = new ProjectEngineManager(emptyCentralCore);
|
||||||
|
|
||||||
|
manager.startReconciliation(1000);
|
||||||
|
|
||||||
|
// Advance time to trigger the immediate tick (no projects yet)
|
||||||
|
await vi.advanceTimersByTimeAsync(1000);
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeUndefined();
|
||||||
|
|
||||||
|
// Simulate a new project being registered
|
||||||
|
projectMap.set("proj_aaa", projectA);
|
||||||
|
(emptyCentralCore.listProjects as ReturnType<typeof vi.fn>).mockResolvedValue([projectA]);
|
||||||
|
|
||||||
|
// Advance time to trigger reconciliation - should find the new project
|
||||||
|
await vi.advanceTimersByTimeAsync(1000);
|
||||||
|
|
||||||
|
// Should have started engine for the new project
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
manager.stopReconciliation();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("retries failed project starts on subsequent reconciliation ticks", async () => {
|
||||||
|
// Track how many times start() is called to fail only the FIRST set
|
||||||
|
let startCallCount = 0;
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
|
||||||
|
// Make starts fail on the first 3 calls (one per project in the first tick)
|
||||||
|
(ProjectEngine as unknown as ReturnType<typeof vi.fn>).mockImplementation(
|
||||||
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||||
|
(config: any) => ({
|
||||||
|
start: vi.fn().mockImplementation(async () => {
|
||||||
|
startCallCount++;
|
||||||
|
// Fail only the first 3 calls (one per project in first reconciliation tick)
|
||||||
|
if (startCallCount <= 3) {
|
||||||
|
throw new Error("transient failure");
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
stop: vi.fn().mockResolvedValue(undefined),
|
||||||
|
getTaskStore: vi.fn().mockReturnValue({ projectId: config.projectId }),
|
||||||
|
_config: config,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
// Start reconciliation (runs immediate tick which fails all 3)
|
||||||
|
manager.startReconciliation(1000);
|
||||||
|
|
||||||
|
// Wait for the immediate tick to complete
|
||||||
|
await vi.advanceTimersByTimeAsync(100);
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 10)); // Let promises settle
|
||||||
|
|
||||||
|
// After immediate tick: all should have failed
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeUndefined();
|
||||||
|
expect(manager.getEngine("proj_bbb")).toBeUndefined();
|
||||||
|
expect(manager.getEngine("proj_ccc")).toBeUndefined();
|
||||||
|
|
||||||
|
// First scheduled tick (after 1000ms): should retry and succeed
|
||||||
|
await vi.advanceTimersByTimeAsync(1000);
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 10)); // Let promises settle
|
||||||
|
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
expect(manager.getEngine("proj_bbb")).toBeDefined();
|
||||||
|
expect(manager.getEngine("proj_ccc")).toBeDefined();
|
||||||
|
|
||||||
|
manager.stopReconciliation();
|
||||||
|
|
||||||
|
// Reset mock for other tests
|
||||||
|
(ProjectEngine as unknown as ReturnType<typeof vi.fn>).mockImplementation(
|
||||||
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||||
|
(config: any) => ({
|
||||||
|
start: vi.fn().mockResolvedValue(undefined),
|
||||||
|
stop: vi.fn().mockResolvedValue(undefined),
|
||||||
|
getTaskStore: vi.fn().mockReturnValue({ projectId: config.projectId }),
|
||||||
|
_config: config,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("is idempotent - multiple calls to startReconciliation don't create multiple intervals", async () => {
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
|
||||||
|
// Call multiple times
|
||||||
|
manager.startReconciliation();
|
||||||
|
manager.startReconciliation();
|
||||||
|
manager.startReconciliation();
|
||||||
|
|
||||||
|
// Advance time
|
||||||
|
await vi.advanceTimersByTimeAsync(35000);
|
||||||
|
|
||||||
|
// Should only have started engines once (not 3 times each)
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
expect(manager.getEngine("proj_bbb")).toBeDefined();
|
||||||
|
expect(manager.getEngine("proj_ccc")).toBeDefined();
|
||||||
|
|
||||||
|
// stopReconciliation should clean up without errors
|
||||||
|
manager.stopReconciliation();
|
||||||
|
manager.stopReconciliation(); // idempotent
|
||||||
|
});
|
||||||
|
|
||||||
|
it("stopReconciliation stops the interval and prevents future ticks", async () => {
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
|
||||||
|
manager.startReconciliation(1000);
|
||||||
|
|
||||||
|
// Let the first tick run
|
||||||
|
await vi.advanceTimersByTimeAsync(1000);
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
|
||||||
|
// Stop reconciliation
|
||||||
|
manager.stopReconciliation();
|
||||||
|
|
||||||
|
// Clear mocks to track new calls
|
||||||
|
vi.clearAllMocks();
|
||||||
|
(ProjectEngine as unknown as ReturnType<typeof vi.fn>).mockImplementation(
|
||||||
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||||
|
(config: any) => ({
|
||||||
|
start: vi.fn().mockResolvedValue(undefined),
|
||||||
|
stop: vi.fn().mockResolvedValue(undefined),
|
||||||
|
getTaskStore: vi.fn().mockReturnValue({ projectId: config.projectId }),
|
||||||
|
_config: config,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
// Advance more time - no new engines should be started
|
||||||
|
await vi.advanceTimersByTimeAsync(1000);
|
||||||
|
expect(ProjectEngine).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("stopAll stops reconciliation as part of shutdown", async () => {
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
|
||||||
|
manager.startReconciliation(1000);
|
||||||
|
await manager.startAll();
|
||||||
|
|
||||||
|
// Verify engines started
|
||||||
|
expect(manager.getEngine("proj_aaa")).toBeDefined();
|
||||||
|
|
||||||
|
// stopAll should stop reconciliation
|
||||||
|
await manager.stopAll();
|
||||||
|
|
||||||
|
// Calling startReconciliation after stopAll should be no-op
|
||||||
|
vi.clearAllMocks();
|
||||||
|
manager.startReconciliation();
|
||||||
|
await vi.advanceTimersByTimeAsync(35000);
|
||||||
|
|
||||||
|
// No new engines because manager is stopped
|
||||||
|
expect(ProjectEngine).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("interval is unref'd so it doesn't prevent process exit", async () => {
|
||||||
|
vi.clearAllTimers();
|
||||||
|
vi.useRealTimers();
|
||||||
|
|
||||||
|
const manager = new ProjectEngineManager(centralCore);
|
||||||
|
manager.startReconciliation(60000);
|
||||||
|
|
||||||
|
// The interval should be unref'd - we can't easily test this directly,
|
||||||
|
// but we verify startReconciliation doesn't throw
|
||||||
|
expect(() => manager.stopReconciliation()).not.toThrow();
|
||||||
|
|
||||||
|
await manager.stopAll();
|
||||||
|
});
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -7,7 +7,8 @@
|
|||||||
*
|
*
|
||||||
* The manager is the single owner of all engines. It handles:
|
* The manager is the single owner of all engines. It handles:
|
||||||
* - Eager startup of all registered projects via `startAll()`
|
* - Eager startup of all registered projects via `startAll()`
|
||||||
* - Lazy startup of newly-accessed projects via `ensureEngine()`
|
* - Background reconciliation of newly registered projects via `startReconciliation()`
|
||||||
|
* - Lazy startup of newly-accessed projects via `ensureEngine()` and `onProjectAccessed()`
|
||||||
* - Deduplication of concurrent start requests for the same project
|
* - Deduplication of concurrent start requests for the same project
|
||||||
* - Graceful shutdown of all engines via `stopAll()`
|
* - Graceful shutdown of all engines via `stopAll()`
|
||||||
*/
|
*/
|
||||||
@@ -34,6 +35,9 @@ export interface EngineManagerOptions {
|
|||||||
onInsightRunProcessed?: ProjectEngineOptions["onInsightRunProcessed"];
|
onInsightRunProcessed?: ProjectEngineOptions["onInsightRunProcessed"];
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Default interval for background reconciliation (30 seconds). */
|
||||||
|
export const DEFAULT_RECONCILIATION_INTERVAL_MS = 30_000;
|
||||||
|
|
||||||
export class ProjectEngineManager {
|
export class ProjectEngineManager {
|
||||||
private engines = new Map<string, ProjectEngine>();
|
private engines = new Map<string, ProjectEngine>();
|
||||||
private starting = new Map<string, Promise<ProjectEngine>>();
|
private starting = new Map<string, Promise<ProjectEngine>>();
|
||||||
@@ -48,6 +52,10 @@ export class ProjectEngineManager {
|
|||||||
private currentGlobalLimit = 4;
|
private currentGlobalLimit = 4;
|
||||||
private concurrencyListener?: (...args: unknown[]) => void;
|
private concurrencyListener?: (...args: unknown[]) => void;
|
||||||
|
|
||||||
|
/** Reconciliation state for background project startup. */
|
||||||
|
private reconciliationInterval: ReturnType<typeof setInterval> | null = null;
|
||||||
|
private reconciliationStopped = false;
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
private centralCore: CentralCore,
|
private centralCore: CentralCore,
|
||||||
private options: EngineManagerOptions = {},
|
private options: EngineManagerOptions = {},
|
||||||
@@ -163,9 +171,16 @@ export class ProjectEngineManager {
|
|||||||
runtimeLog.log(`Engine startup complete: ${started} started, ${failed} failed`);
|
runtimeLog.log(`Engine startup complete: ${started} started, ${failed} failed`);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Gracefully stop all engines. */
|
/** Gracefully stop all engines and reconciliation. */
|
||||||
async stopAll(): Promise<void> {
|
async stopAll(): Promise<void> {
|
||||||
this.stopped = true;
|
this.stopped = true;
|
||||||
|
this.reconciliationStopped = true;
|
||||||
|
|
||||||
|
// Stop reconciliation interval
|
||||||
|
if (this.reconciliationInterval !== null) {
|
||||||
|
clearInterval(this.reconciliationInterval);
|
||||||
|
this.reconciliationInterval = null;
|
||||||
|
}
|
||||||
|
|
||||||
// Remove concurrency change listener
|
// Remove concurrency change listener
|
||||||
if (this.concurrencyListener && typeof this.centralCore.off === "function") {
|
if (this.concurrencyListener && typeof this.centralCore.off === "function") {
|
||||||
@@ -202,6 +217,96 @@ export class ProjectEngineManager {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── Background Reconciliation ────────────────────────────────────────
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Start background reconciliation to detect and start engines for
|
||||||
|
* newly registered projects without requiring UI access.
|
||||||
|
*
|
||||||
|
* This runs on an interval, checking for projects that have been
|
||||||
|
* registered but don't have running engines yet.
|
||||||
|
*
|
||||||
|
* Idempotent — safe to call multiple times. Reconciliation stops
|
||||||
|
* when `stopReconciliation()` or `stopAll()` is called.
|
||||||
|
*
|
||||||
|
* @param intervalMs How often to check for new projects (default: 30 seconds)
|
||||||
|
*/
|
||||||
|
startReconciliation(intervalMs: number = DEFAULT_RECONCILIATION_INTERVAL_MS): void {
|
||||||
|
if (this.stopped || this.reconciliationStopped) return;
|
||||||
|
if (this.reconciliationInterval !== null) return; // Already running
|
||||||
|
|
||||||
|
runtimeLog.log(`Starting project engine reconciliation (interval: ${intervalMs}ms)`);
|
||||||
|
|
||||||
|
// Run an immediate reconciliation tick, then schedule periodic checks
|
||||||
|
this.reconcile().catch((err) => {
|
||||||
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
|
runtimeLog.warn(`Reconciliation tick failed: ${message}`);
|
||||||
|
});
|
||||||
|
|
||||||
|
this.reconciliationInterval = setInterval(() => {
|
||||||
|
if (this.reconciliationStopped) {
|
||||||
|
this.stopReconciliation();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
this.reconcile().catch((err) => {
|
||||||
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
|
runtimeLog.warn(`Reconciliation tick failed: ${message}`);
|
||||||
|
});
|
||||||
|
}, intervalMs);
|
||||||
|
|
||||||
|
// Prevent the interval from keeping the process alive
|
||||||
|
this.reconciliationInterval.unref?.();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Stop background reconciliation.
|
||||||
|
* Idempotent — safe to call even if reconciliation is not running.
|
||||||
|
*/
|
||||||
|
stopReconciliation(): void {
|
||||||
|
this.reconciliationStopped = true;
|
||||||
|
if (this.reconciliationInterval !== null) {
|
||||||
|
clearInterval(this.reconciliationInterval);
|
||||||
|
this.reconciliationInterval = null;
|
||||||
|
runtimeLog.log("Stopped project engine reconciliation");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Check for registered projects that don't have engines and start them.
|
||||||
|
* This is the core reconciliation logic used by both `startReconciliation`
|
||||||
|
* and `startAll()`.
|
||||||
|
*/
|
||||||
|
private async reconcile(): Promise<void> {
|
||||||
|
if (this.stopped || this.reconciliationStopped) return;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const projects = await this.centralCore.listProjects();
|
||||||
|
if (projects.length === 0) return;
|
||||||
|
|
||||||
|
// Find projects that don't have running or pending engines
|
||||||
|
const missing = projects.filter((p) => !this.has(p.id));
|
||||||
|
if (missing.length === 0) return;
|
||||||
|
|
||||||
|
runtimeLog.log(
|
||||||
|
`Reconciliation: found ${missing.length} project(s) without engines`,
|
||||||
|
);
|
||||||
|
|
||||||
|
// Start engines for missing projects (fire-and-forget)
|
||||||
|
for (const project of missing) {
|
||||||
|
if (this.stopped || this.reconciliationStopped) break;
|
||||||
|
this.ensureEngine(project.id).catch((err) => {
|
||||||
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
|
runtimeLog.warn(
|
||||||
|
`Failed to start engine for project ${project.id}: ${message}`,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
|
runtimeLog.warn(`Reconciliation failed: ${message}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ── Internal ──
|
// ── Internal ──
|
||||||
|
|
||||||
private async createAndStart(
|
private async createAndStart(
|
||||||
|
|||||||
Reference in New Issue
Block a user