feat(FN-3040): make peer exchange shutdown deterministic and improve dashbo
This merge lands seven features and fixes across the dashboard and engine. Notable changes: restructured TodoView rows with improved action row styling, added `/clear` command to Chat and QuickChat, persisted session banner dismissals with a hide-banner setting, made peer exchange shutdown determini Fusion-Task-Id: FN-3040
This commit is contained in:
5
.changeset/fn-3040-mesh-lifecycle.md
Normal file
5
.changeset/fn-3040-mesh-lifecycle.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Clarify and harden cross-node mesh lifecycle ownership in node startup paths. Peer exchange shutdown is now deterministic (idempotent and waits for in-flight sync), and docs/tests now codify that mesh discovery + peer exchange are owned by `fn serve`/`fn dashboard` process lifecycle rather than per-project runtime startup.
|
||||||
@@ -358,6 +358,10 @@ Implemented in `agent-heartbeat.ts`:
|
|||||||
### Node/mesh runtime services
|
### Node/mesh runtime services
|
||||||
- `NodeHealthMonitor` (`node-health-monitor.ts`) — remote node liveness/metrics checks
|
- `NodeHealthMonitor` (`node-health-monitor.ts`) — remote node liveness/metrics checks
|
||||||
- `PeerExchangeService` (`peer-exchange-service.ts`) — peer sync orchestration
|
- `PeerExchangeService` (`peer-exchange-service.ts`) — peer sync orchestration
|
||||||
|
- Process lifecycle ownership:
|
||||||
|
- `fn serve` / `fn dashboard` start a single process-level `PeerExchangeService` and stop it during shutdown.
|
||||||
|
- `CentralCore.startDiscovery()` is invoked from CLI startup only after HTTP bind completes so discovery advertises the actual listening port.
|
||||||
|
- `InProcessRuntime` stays project-scoped and intentionally does not own mesh startup/shutdown.
|
||||||
|
|
||||||
### Remote access runtime
|
### Remote access runtime
|
||||||
|
|
||||||
|
|||||||
@@ -31,9 +31,14 @@ Core tables:
|
|||||||
|
|
||||||
Per-project task data remains in each repo’s `.fusion/fusion.db`.
|
Per-project task data remains in each repo’s `.fusion/fusion.db`.
|
||||||
|
|
||||||
Peer/mesh coordination spans core + engine:
|
Peer/mesh coordination spans core + engine, with startup ownership in CLI process entrypoints:
|
||||||
- `NodeDiscovery` and `NodeConnection` in `@fusion/core` handle discovery and remote node connectivity/auth.
|
- `NodeDiscovery` and `NodeConnection` in `@fusion/core` handle discovery and remote node connectivity/auth primitives.
|
||||||
- `PeerExchangeService` in `@fusion/engine` coordinates node-to-node sync/exchange workflows.
|
- `PeerExchangeService` in `@fusion/engine` coordinates node-to-node sync/exchange workflows.
|
||||||
|
- `runServe()` and `runDashboard()` (CLI) own process-level mesh service lifecycle:
|
||||||
|
- start one process-wide `PeerExchangeService` instance
|
||||||
|
- call `CentralCore.startDiscovery()` only after the HTTP server is listening and the real bound port is known
|
||||||
|
- stop peer exchange + discovery on shutdown
|
||||||
|
- `InProcessRuntime` remains project-scoped (scheduler/executor/heartbeat/missions) and does **not** start mesh services, which avoids one peer-exchange instance per project.
|
||||||
|
|
||||||
## Registering and Managing Projects
|
## Registering and Managing Projects
|
||||||
|
|
||||||
|
|||||||
@@ -73,6 +73,8 @@ Defaults from `DEFAULT_GLOBAL_SETTINGS`; key scope from `GLOBAL_SETTINGS_KEYS`.
|
|||||||
| `settingsSyncInterval` | `number` | `900000` | Automatic sync interval in ms. Valid values: `300000`, `900000`, `1800000`, `3600000`. |
|
| `settingsSyncInterval` | `number` | `900000` | Automatic sync interval in ms. Valid values: `300000`, `900000`, `1800000`, `3600000`. |
|
||||||
| `settingsSyncConflictResolution` | `"last-write-wins" \| "always-ask" \| "keep-local" \| "keep-remote"` | `"last-write-wins"` | Conflict strategy for divergent synced settings. |
|
| `settingsSyncConflictResolution` | `"last-write-wins" \| "always-ask" \| "keep-local" \| "keep-remote"` | `"last-write-wins"` | Conflict strategy for divergent synced settings. |
|
||||||
| `dashboardCurrentNodeId` | `string` | `undefined` | Currently selected dashboard node ID. Restores the last-viewed node on fresh browser/PWA sessions. `undefined` means viewing the local node. |
|
| `dashboardCurrentNodeId` | `string` | `undefined` | Currently selected dashboard node ID. Restores the last-viewed node on fresh browser/PWA sessions. `undefined` means viewing the local node. |
|
||||||
|
|
||||||
|
> Mesh lifecycle note: settings sync is executed by the process-level `PeerExchangeService` started by `fn serve`/`fn dashboard`. `InProcessRuntime` does not instantiate settings-sync mesh services per project.
|
||||||
| `dashboardCurrentProjectIdByNode` | `Record<string, string>` | `undefined` | Map of node ID to last-selected project ID. Use key `"local"` for the local node. Persists project context across browser restarts and PWA sessions. |
|
| `dashboardCurrentProjectIdByNode` | `Record<string, string>` | `undefined` | Map of node ID to last-selected project ID. Use key `"local"` for the local node. Persists project context across browser restarts and PWA sessions. |
|
||||||
| `researchGlobalDefaults` | `ResearchGlobalDefaults` | `{ searchProvider: undefined, synthesisProvider: undefined, synthesisModelId: undefined, enabledSources: { webSearch: true, pageFetch: true, github: false, localDocs: true, llmSynthesis: true }, maxSourcesPerRun: 20, defaultExportFormat: "markdown" }` | Global Research defaults shared by all projects. Project overrides come from `researchSettings`. |
|
| `researchGlobalDefaults` | `ResearchGlobalDefaults` | `{ searchProvider: undefined, synthesisProvider: undefined, synthesisModelId: undefined, enabledSources: { webSearch: true, pageFetch: true, github: false, localDocs: true, llmSynthesis: true }, maxSourcesPerRun: 20, defaultExportFormat: "markdown" }` | Global Research defaults shared by all projects. Project overrides come from `researchSettings`. |
|
||||||
| `experimentalFeatures` | `Record<string, boolean>` | `{}` | Global-scoped experimental feature flags. Includes `experimentalFeatures.researchView` for standalone Research route visibility. |
|
| `experimentalFeatures` | `Record<string, boolean>` | `{}` | Global-scoped experimental feature flags. Includes `experimentalFeatures.researchView` for standalone Research route visibility. |
|
||||||
|
|||||||
@@ -293,6 +293,7 @@ vi.mock("@fusion/dashboard", () => ({
|
|||||||
getCliPackageVersion: mockGetCliPackageVersion,
|
getCliPackageVersion: mockGetCliPackageVersion,
|
||||||
getProjectSettingsPath: vi.fn().mockReturnValue("/tmp/project/.fusion/settings.json"),
|
getProjectSettingsPath: vi.fn().mockReturnValue("/tmp/project/.fusion/settings.json"),
|
||||||
loadTlsCredentialsFromEnv: vi.fn().mockReturnValue(undefined),
|
loadTlsCredentialsFromEnv: vi.fn().mockReturnValue(undefined),
|
||||||
|
stopAllDevServers: vi.fn().mockResolvedValue(undefined),
|
||||||
}));
|
}));
|
||||||
|
|
||||||
// ── Mock node:readline ──────────────────────────────────────────────
|
// ── Mock node:readline ──────────────────────────────────────────────
|
||||||
@@ -682,6 +683,10 @@ vi.mock("@fusion/engine", async (importOriginal) => {
|
|||||||
getPlugin: vi.fn(),
|
getPlugin: vi.fn(),
|
||||||
getLoadedPlugins: vi.fn().mockReturnValue([]),
|
getLoadedPlugins: vi.fn().mockReturnValue([]),
|
||||||
})),
|
})),
|
||||||
|
PeerExchangeService: vi.fn().mockImplementation(() => ({
|
||||||
|
start: vi.fn(),
|
||||||
|
stop: vi.fn().mockResolvedValue(undefined),
|
||||||
|
})),
|
||||||
scanIdleWorktrees: vi.fn().mockResolvedValue([]),
|
scanIdleWorktrees: vi.fn().mockResolvedValue([]),
|
||||||
cleanupOrphanedWorktrees: vi.fn().mockResolvedValue(0),
|
cleanupOrphanedWorktrees: vi.fn().mockResolvedValue(0),
|
||||||
});
|
});
|
||||||
@@ -2444,6 +2449,97 @@ describe("runDashboard — lifecycle listener cleanup", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("runDashboard — mesh lifecycle ownership", () => {
|
||||||
|
function getNewSignalHandler(
|
||||||
|
signal: "SIGINT" | "SIGTERM",
|
||||||
|
baseline: Array<(...args: any[]) => unknown>,
|
||||||
|
): () => void {
|
||||||
|
const added = process.listeners(signal).find((listener) => !baseline.includes(listener as (...args: any[]) => unknown));
|
||||||
|
expect(added).toBeDefined();
|
||||||
|
return added as () => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
it("starts peer exchange and discovery after the dashboard binds a port", async () => {
|
||||||
|
const { CentralCore } = await import("@fusion/core");
|
||||||
|
const { PeerExchangeService } = await import("@fusion/engine");
|
||||||
|
|
||||||
|
const startDiscovery = vi.fn().mockResolvedValue(undefined);
|
||||||
|
const updateNode = vi.fn().mockResolvedValue(undefined);
|
||||||
|
|
||||||
|
(CentralCore as unknown as ReturnType<typeof vi.fn>).mockImplementation(() => ({
|
||||||
|
init: vi.fn().mockResolvedValue(undefined),
|
||||||
|
close: vi.fn().mockResolvedValue(undefined),
|
||||||
|
getProjectByPath: vi.fn().mockResolvedValue({ id: "project-1" }),
|
||||||
|
listProjects: vi.fn().mockResolvedValue([{ id: "project-1", path: process.cwd() }]),
|
||||||
|
listNodes: vi.fn().mockResolvedValue([{ id: "node-local", type: "local", status: "offline" }]),
|
||||||
|
updateNode,
|
||||||
|
startDiscovery,
|
||||||
|
stopDiscovery: vi.fn(),
|
||||||
|
}));
|
||||||
|
|
||||||
|
const peerExchangeCtor = PeerExchangeService as unknown as ReturnType<typeof vi.fn>;
|
||||||
|
const baselineCalls = peerExchangeCtor.mock.calls.length;
|
||||||
|
|
||||||
|
const { dispose } = await runDashboard(0, { open: false });
|
||||||
|
await new Promise((resolve) => setImmediate(resolve));
|
||||||
|
await new Promise((resolve) => setImmediate(resolve));
|
||||||
|
|
||||||
|
expect(peerExchangeCtor.mock.calls.length).toBeGreaterThan(baselineCalls);
|
||||||
|
const peerExchangeInstance = peerExchangeCtor.mock.results.at(-1)?.value;
|
||||||
|
expect(peerExchangeInstance.start).toHaveBeenCalledTimes(1);
|
||||||
|
expect(startDiscovery).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
broadcast: true,
|
||||||
|
listen: true,
|
||||||
|
serviceType: "_fusion._tcp",
|
||||||
|
port: 0,
|
||||||
|
}));
|
||||||
|
expect(updateNode).toHaveBeenCalledWith("node-local", { status: "online" });
|
||||||
|
|
||||||
|
dispose();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("stops peer exchange and discovery during shutdown", async () => {
|
||||||
|
const { CentralCore } = await import("@fusion/core");
|
||||||
|
const { PeerExchangeService } = await import("@fusion/engine");
|
||||||
|
|
||||||
|
const stopDiscovery = vi.fn();
|
||||||
|
const updateNode = vi.fn().mockResolvedValue(undefined);
|
||||||
|
|
||||||
|
(CentralCore as unknown as ReturnType<typeof vi.fn>).mockImplementation(() => ({
|
||||||
|
init: vi.fn().mockResolvedValue(undefined),
|
||||||
|
close: vi.fn().mockResolvedValue(undefined),
|
||||||
|
getProjectByPath: vi.fn().mockResolvedValue({ id: "project-1" }),
|
||||||
|
listProjects: vi.fn().mockResolvedValue([{ id: "project-1", path: process.cwd() }]),
|
||||||
|
listNodes: vi.fn().mockResolvedValue([{ id: "node-local", type: "local", status: "offline" }]),
|
||||||
|
updateNode,
|
||||||
|
startDiscovery: vi.fn().mockResolvedValue(undefined),
|
||||||
|
stopDiscovery,
|
||||||
|
}));
|
||||||
|
|
||||||
|
const exitSpy = vi.spyOn(process, "exit").mockImplementation((() => undefined) as never);
|
||||||
|
const baselineSigintHandlers = process.listeners("SIGINT");
|
||||||
|
|
||||||
|
try {
|
||||||
|
await runDashboard(0, { open: false });
|
||||||
|
await new Promise((resolve) => setImmediate(resolve));
|
||||||
|
await new Promise((resolve) => setImmediate(resolve));
|
||||||
|
const sigintHandler = getNewSignalHandler("SIGINT", baselineSigintHandlers);
|
||||||
|
sigintHandler();
|
||||||
|
|
||||||
|
await new Promise((resolve) => setImmediate(resolve));
|
||||||
|
await new Promise((resolve) => setImmediate(resolve));
|
||||||
|
|
||||||
|
const peerExchangeInstance = (PeerExchangeService as unknown as ReturnType<typeof vi.fn>).mock.results.at(-1)?.value;
|
||||||
|
expect(peerExchangeInstance.stop).toHaveBeenCalledTimes(1);
|
||||||
|
expect(stopDiscovery).toHaveBeenCalledTimes(1);
|
||||||
|
expect(updateNode).toHaveBeenCalledWith("node-local", { status: "offline" });
|
||||||
|
expect(exitSpy).toHaveBeenCalledWith(0);
|
||||||
|
} finally {
|
||||||
|
exitSpy.mockRestore();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("runDashboard — CentralCore cleanup diagnostics", () => {
|
describe("runDashboard — CentralCore cleanup diagnostics", () => {
|
||||||
function getNewSignalHandler(
|
function getNewSignalHandler(
|
||||||
signal: "SIGINT" | "SIGTERM",
|
signal: "SIGINT" | "SIGTERM",
|
||||||
|
|||||||
@@ -48,7 +48,7 @@ export class PeerExchangeService {
|
|||||||
private syncIntervalMs: number;
|
private syncIntervalMs: number;
|
||||||
private interval: ReturnType<typeof setInterval> | null = null;
|
private interval: ReturnType<typeof setInterval> | null = null;
|
||||||
private activeSync: Promise<void> | null = null;
|
private activeSync: Promise<void> | null = null;
|
||||||
private stopped = false;
|
private running = false;
|
||||||
/** Whether settings sync is enabled. Default: false. */
|
/** Whether settings sync is enabled. Default: false. */
|
||||||
private settingsSyncEnabled: boolean;
|
private settingsSyncEnabled: boolean;
|
||||||
/** Minimum interval between settings syncs with the same node in ms. Default: 5 minutes. */
|
/** Minimum interval between settings syncs with the same node in ms. Default: 5 minutes. */
|
||||||
@@ -94,11 +94,13 @@ export class PeerExchangeService {
|
|||||||
* Begins periodic gossip with all online remote nodes.
|
* Begins periodic gossip with all online remote nodes.
|
||||||
*/
|
*/
|
||||||
start(): void {
|
start(): void {
|
||||||
if (this.stopped) {
|
if (this.running) {
|
||||||
peerExchangeLog.warn("Cannot start - service has been stopped");
|
peerExchangeLog.log("Peer exchange service already running");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
this.running = true;
|
||||||
|
|
||||||
// Get initial peer count for logging (async call)
|
// Get initial peer count for logging (async call)
|
||||||
this.centralCore.listNodes().then((nodes) => {
|
this.centralCore.listNodes().then((nodes) => {
|
||||||
const onlineRemoteCount = nodes.filter(
|
const onlineRemoteCount = nodes.filter(
|
||||||
@@ -112,6 +114,7 @@ export class PeerExchangeService {
|
|||||||
|
|
||||||
// Start periodic sync
|
// Start periodic sync
|
||||||
this.interval = setInterval(() => {
|
this.interval = setInterval(() => {
|
||||||
|
if (!this.running) return;
|
||||||
void this.syncWithAllPeers();
|
void this.syncWithAllPeers();
|
||||||
}, this.syncIntervalMs);
|
}, this.syncIntervalMs);
|
||||||
}
|
}
|
||||||
@@ -120,13 +123,28 @@ export class PeerExchangeService {
|
|||||||
* Stop the peer exchange service.
|
* Stop the peer exchange service.
|
||||||
* Clears the sync interval and prevents further syncs.
|
* Clears the sync interval and prevents further syncs.
|
||||||
*/
|
*/
|
||||||
stop(): void {
|
async stop(): Promise<void> {
|
||||||
|
if (!this.running) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.running = false;
|
||||||
|
|
||||||
if (this.interval) {
|
if (this.interval) {
|
||||||
clearInterval(this.interval);
|
clearInterval(this.interval);
|
||||||
this.interval = null;
|
this.interval = null;
|
||||||
}
|
}
|
||||||
|
|
||||||
this.stopped = true;
|
// If a sync cycle is already in flight, wait for it to settle so shutdown
|
||||||
|
// does not leave partially completed gossip work behind.
|
||||||
|
if (this.activeSync) {
|
||||||
|
try {
|
||||||
|
await this.activeSync;
|
||||||
|
} catch {
|
||||||
|
// best-effort shutdown; sync errors are already logged by run loop
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
peerExchangeLog.log("Stopped peer exchange service");
|
peerExchangeLog.log("Stopped peer exchange service");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -48,6 +48,12 @@ import { TriageProcessor } from "../triage.js";
|
|||||||
* - Graceful shutdown with configurable timeout
|
* - Graceful shutdown with configurable timeout
|
||||||
* - Automatic orphaned task recovery on startup
|
* - Automatic orphaned task recovery on startup
|
||||||
*
|
*
|
||||||
|
* Lifecycle boundary:
|
||||||
|
* - Mesh networking services (PeerExchangeService + mDNS discovery) are process-level
|
||||||
|
* concerns owned by CLI startup paths (`runServe`/`runDashboard`) because discovery
|
||||||
|
* requires the final bound HTTP port. InProcessRuntime remains project-scoped and
|
||||||
|
* intentionally does not start process-level mesh services.
|
||||||
|
*
|
||||||
* @example
|
* @example
|
||||||
* ```typescript
|
* ```typescript
|
||||||
* const config: ProjectRuntimeConfig = {
|
* const config: ProjectRuntimeConfig = {
|
||||||
|
|||||||
Reference in New Issue
Block a user