feat: shared Postgres multi-node — retire mesh data-plane replication (#2130)
## Summary - Treat **shared PostgreSQL** (`DATABASE_URL`) as the multi-node durable data plane; mesh HTTP is membership + optional auth, not task/settings replication. - **Peer exchange**: under Postgres backend mode, write queue is **topology/auth-only**; non-topology pending rows fail rather than replaying multi-leader task/settings payloads. - **Mesh routes**: task-ID reserve/commit/abort always hit local shared allocator rows (ignore remote `coordinatorNodeId`); mesh sync ignores settings and only exchanges `authMaterial`. - **Docs**: rewrite multi-project runbook, shared cluster protocol, and architecture mesh sections for shared-Postgres + claims/leases. ## Context Follows the SQLite→Postgres cutover. Multiple Fusion nodes can share one external Postgres while keeping **per-node execution** (worktrees, processes, claims via `central.task_claims`). Explicit non-goals remain: scheduler failover and live process migration. Plan: `docs/plans/2026-07-15-001-refactor-mesh-shared-postgres-multinode-plan.md` ## Test plan - [x] `pnpm --filter @fusion/engine exec vitest run src/__tests__/peer-exchange-service.test.ts` - [x] `pnpm --filter @fusion/dashboard exec vitest run src/__tests__/mesh-routes.test.ts` - [x] `pnpm --filter @fusion/core exec vitest run src/__tests__/shared-mesh-state.test.ts` - [ ] CI gate (lint/typecheck/build/gate) - [ ] Manual (optional): two processes, same `DATABASE_URL`, create task on A visible on B; settings change without mesh settings sync; claim exclusivity ## Operator note Multi-node shared board requires **external** `DATABASE_URL` on every node. Default embedded Postgres is still single-host. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Improved multi-node deployments using shared PostgreSQL as the durable source of execution state. * Task ID reservation/commit/abort now run locally (no remote coordinator forwarding). * Mesh syncing now prioritizes topology visibility and authentication material; settings replication is disabled in shared-Postgres mode. * **Bug Fixes** * Prevented task/settings replication over mesh HTTP in shared-Postgres deployments. * Refined lease ownership, recovery, and reconciliation to converge via shared-database primitives. * **Documentation** * Updated architecture and shared-mesh protocol guidance, including multi-node setup and lease/task-ID allocation behavior. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
gsxdsm
7
.changeset/friendly-shims-fix.md
Normal file
7
.changeset/friendly-shims-fix.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Fix clean-CI typechecking for bundled plugins that use PostgreSQL schemas.
|
||||
category: fix
|
||||
dev: Bundle the core schema through a runtime-only shim instead of requiring an unbuilt core dist artifact.
|
||||
7
.changeset/mesh-shared-postgres-cutover.md
Normal file
7
.changeset/mesh-shared-postgres-cutover.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Multi-node fleets on shared Postgres no longer replicate tasks or settings over mesh HTTP.
|
||||
category: internal
|
||||
dev: Peer exchange queues topology/auth only; task-ID routes always allocate against shared rows; docs describe DATABASE_URL multi-node + claims.
|
||||
@@ -61,7 +61,7 @@ Planner oversight (FN-7508 → FN-7583) is fully documented in Settings Referenc
|
||||
| [Storage](./storage.md) | PostgreSQL runtime storage, archive, migration compatibility, and file-backed payloads |
|
||||
| [DAG Architecture Deliverables](./dag/) | Milestone A DAG architecture documents plus Milestone B prototype scaffold docs (schema migration plan, DagCoordinator design, implementation checklist) |
|
||||
| [Dev Server Module Audit](./dev-server-modules.md) | Analysis of parallel dashboard dev-server module families, production wiring, and consolidation guidance |
|
||||
| [Shared Mesh Replication Protocol](./shared-mesh-protocol.md) | Canonical multi-leader replication/write-coordination contract (versioning, quorum, leases/fencing, queue/replay, reconciliation, and degraded-read semantics) |
|
||||
| [Shared Cluster Protocol](./shared-mesh-protocol.md) | Shared PostgreSQL multi-node contract: claims/leases, membership, auth, and retired multi-leader mesh replication |
|
||||
| [Signals Connectors](./signals-connectors.md) | HMAC-signed external signal connectors for setup, payload mapping, and security notes across Sentry, Datadog, PagerDuty, and generic webhooks |
|
||||
| [Multi-Project Sequencing and Dependency Analysis](./multi-project-sequencing.md) | Sequencing guidance for FN-3448/FN-3449/FN-3503/FN-3182, including identity boundaries and recommended board dependency edges |
|
||||
| [Contributing](./contributing.md) | Local development setup, testing, release flow, and contributor conventions |
|
||||
|
||||
@@ -779,15 +779,20 @@ Implemented in `agent-heartbeat.ts`:
|
||||
- `WakeContext` / per-agent runtime config support
|
||||
|
||||
### Node/mesh runtime services
|
||||
- `NodeHealthMonitor` (`node-health-monitor.ts`) — remote node liveness/metrics checks
|
||||
- `PeerExchangeService` (`peer-exchange-service.ts`) — peer sync orchestration
|
||||
- `MeshLeaseManager` (`mesh-lease-manager.ts`) — canonical abandoned-lease detection + recovery path
|
||||
<!--
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Under shared PostgreSQL, mesh services own membership/auth/claims — not task replication. Durable board state is the shared database; PeerExchange no longer multi-masters tasks or settings over HTTP.
|
||||
-->
|
||||
- `NodeHealthMonitor` (`node-health-monitor.ts`) — remote node liveness/metrics checks for handoff and recovery
|
||||
- `PeerExchangeService` (`peer-exchange-service.ts`) — membership gossip + optional auth material; settings/task HTTP replication disabled when nodes share Postgres
|
||||
- `MeshLeaseManager` (`mesh-lease-manager.ts`) — canonical abandoned-lease detection + recovery path (`central.task_claims` then task-row mirror)
|
||||
|
||||
### Outage ownership boundaries (degraded reads + queued write replay)
|
||||
- `CentralCore` owns durable outage state in central persistence (`meshSharedSnapshots` + `meshWriteQueue`) and exposes stable assertion methods: `recordMeshSnapshot`, `getLatestMeshSnapshot`, `enqueueMeshWrite`, `listPendingMeshWrites`, `markMeshWriteReplayStarted`, `markMeshWriteApplied`, `markMeshWriteFailed`, and `getMeshDegradedReadState`.
|
||||
- `PeerExchangeService` owns retryability classification for sync/apply failures, queue insertion for retryable failures, replay execution (`replayPendingWritesForNode(targetNodeId)`), and observable sync results (`queuedWriteId`, `replaySummary`) for partition/replay assertions.
|
||||
- `NodeHealthMonitor` provides liveness transitions as replay hints only via deterministic recovery callback `onNodeRecovered(nodeId, previousStatus)`; `online` is a trigger to attempt replay, not proof that replay succeeded.
|
||||
- Dashboard mesh routes (`register-mesh-routes.ts`) preserve `GET /api/mesh/state` array shape and attach per-node degraded `readState` metadata so stale fallback data is explicit during partitions.
|
||||
### Outage ownership boundaries (topology/auth only)
|
||||
- Durable **task** state does not queue offline over mesh when Postgres is down. If the shared database is unavailable, nodes do not invent alternate local task truth.
|
||||
- `CentralCore` still exposes `meshSharedSnapshots` + `meshWriteQueue` for **topology / auth** retry and degraded membership reads (`recordMeshSnapshot`, `getLatestMeshSnapshot`, `enqueueMeshWrite`, `listPendingMeshWrites`, `markMeshWrite*`, `getMeshDegradedReadState`).
|
||||
- `PeerExchangeService` classifies retryable **membership/auth** sync failures, may enqueue those narrow scopes, and replays them via `replayPendingWritesForNode(targetNodeId)`. Task/settings payloads are not queued for multi-leader replay under Postgres.
|
||||
- `NodeHealthMonitor` liveness transitions may trigger membership/auth replay attempts (`onNodeRecovered`); `online` is not proof that replay succeeded.
|
||||
- Dashboard mesh routes (`register-mesh-routes.ts`) keep `GET /api/mesh/state` and attach degraded `readState` metadata when peer probes fail.
|
||||
|
||||
### Mesh task lease ownership and recovery
|
||||
|
||||
@@ -808,26 +813,14 @@ A lease is recoverable only when there is **no active local executor session for
|
||||
|
||||
1. the owning node is `offline` or `error`, or
|
||||
2. the owner heartbeat/run age exceeds `max(agentHeartbeatTimeoutMs * 2, 120_000)` measured against the most recent lease renewal timestamp.
|
||||
- Canonical replication/write-coordination contract: [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md)
|
||||
- Defines protocol versioning, write classes, quorum/ack semantics, lease epochs/fencing, offline queue/replay, reconciliation outcomes, restart recovery hooks, and degraded-read staleness metadata.
|
||||
- Existing `/api/mesh/sync` and settings-sync payloads remain the active exchange primitives while follow-on runtime tasks implement full v1 coordinator/quorum behavior.
|
||||
- Distributed task-ID allocation (`packages/core/src/distributed-task-id.ts`) is the first mesh-aware coordinated write primitive.
|
||||
- Durable state lives in PostgreSQL tables `distributed_task_id_state` (prefix sequence + authoritative committed count) and `distributed_task_id_reservations` (reservation lifecycle rows).
|
||||
- Reserve/commit/abort execute through the async allocator inside PostgreSQL transactions. Lazy reservation expiry cleanup runs in the same transaction model, and task creation commits the reservation flip with the authoritative `project.tasks` insert so both share one durability point.
|
||||
- Default reservation TTL is `15 * 60 * 1000` ms (15 minutes). Expired/aborted reservations are **burned IDs** and are never reissued. If a post-insert create step fails after the reservation was committed (for example `task.json`/`PROMPT.md` disk materialization, file-scope validation, or duplicate-intake tombstone checks), the failed-create rollback deletes the just-created task row/partial directory, moves the reservation to `aborted`, recomputes committed reservation counters, and emits `task:reservation-commit-rolled-back`; the sequence stays burned for FN-5105 ID permanence.
|
||||
- `committedClusterTaskCount` from allocator state is the only authoritative cluster-wide committed-task count. Local task-row counts and ID suffix math are not authoritative.
|
||||
- Store open reconciles every known prefix in `distributed_task_id_state` to `max(current nextSequence, max(tasks suffix)+1, max(archivedTasks suffix)+1, max(reservation sequence)+1)`. This self-heals stale counters before ordinary task creation resumes.
|
||||
- Mesh allocator write routes (`/api/mesh/task-ids/reserve|commit|abort`) return `503` when the coordinator node is unreachable; they never fall back to local-only cluster ID issuance.
|
||||
- Cluster task creation now uses a strong-write reserve → create → replicate → commit/abort sequence.
|
||||
- Ordinary local task creation (`TaskStore.createTask()`, duplicate, and refine flows) now allocates IDs through the same distributed reserve/commit/abort lifecycle owned by `TaskStore`; the invariant is `distributed_task_id_reservations.status = 'committed'` iff a live durable `tasks` row and task directory landed for that ID. `applyReplicatedTaskCreate(...)` remains a direct reserved-ID apply path and does not require a local reservation row.
|
||||
- `POST /api/tasks` uses the store-owned allocator path for local creates rather than maintaining a separate route-local allocator implementation.
|
||||
- `POST /api/tasks` reserves a distributed ID, creates the authoritative local task with that reserved ID, then POSTs authenticated replication payloads to peer nodes.
|
||||
- All create-class writes use conflict-raising inserts, not upserts. Existing PostgreSQL task rows and `.fusion/tasks/{id}` contents always win over stale counters or colliding reservations.
|
||||
- Local create paths perform a final active+archived existence check immediately before insert. If a reserved `FN-*` still collides, the reservation is aborted/burned and the create fails loudly instead of rewriting the existing task.
|
||||
- Creation self-heals stale overlap state at the route layer: if a reserved `FN-*` collides with an existing task (`Task ID already exists...` or replicated-create collision), the route aborts that reservation, cleans up partial local state, reserves the next ID, and retries up to a bounded limit.
|
||||
- Replica apply uses `TaskStore.applyReplicatedTaskCreate(...)`, which is idempotent by task ID: replaying the same payload returns the existing task without creating duplicates.
|
||||
- If an incoming replicated payload conflicts with a different existing task record for the same ID, the apply path returns a deterministic collision error instead of overwriting data.
|
||||
- Any replication/coordinator failure aborts the reservation and returns write failure (`503`), so this path does not report success for local-only partial writes.
|
||||
- Canonical multi-node contract: [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md) (shared Postgres + claims; multi-leader HTTP task replication retired).
|
||||
- Distributed task-ID allocation is a **shared-database** primitive (not a remote mesh coordinator):
|
||||
- Durable state lives in Postgres `project.distributed_task_id_state` / `distributed_task_id_reservations` (async allocator path).
|
||||
- Reserve/commit/abort run against those shared rows under the project’s transactions. Default reservation TTL is `15 * 60 * 1000` ms. Expired/aborted reservations are **burned** and never reissued. Failed creates abort the still-reserved reservation (or roll back a just-committed reservation via `task:reservation-commit-rolled-back` when create materialization fails after the commit flip); once a reservation is fully committed with a durable task row, it is final and not re-aborted as an ordinary abort.
|
||||
- `committedClusterTaskCount` is the authoritative committed-task count for a prefix within a project partition.
|
||||
- Store open reconciles each prefix high-water mark past existing live/archived/reservation sequences.
|
||||
- `/api/mesh/task-ids/*` always uses the local allocator under Postgres (shared rows are the coordinator). Remote coordinator forwarding is disabled.
|
||||
- Task creation: reserve → insert task (shared DB) → commit reservation (or abort on failure). HTTP peer task replication and `applyReplicatedTaskCreate` multi-node fan-out are not part of the Postgres multi-node path.
|
||||
- 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.
|
||||
|
||||
@@ -18,17 +18,33 @@ Use multi-project mode when you need to:
|
||||
|
||||
Multi-project metadata is stored in the PostgreSQL `central` schema. Embedded mode uses Fusion's managed PostgreSQL data directory; external mode uses `DATABASE_URL`.
|
||||
|
||||
Core tables:
|
||||
<!--
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Multi-node durable state lives in shared PostgreSQL (central + project schemas), not per-node SQLite files or HTTP task replication. Embedded Postgres is per-machine only; multi-node shared boards require external DATABASE_URL.
|
||||
-->
|
||||
|
||||
Fusion stores multi-project and multi-node coordination state in **PostgreSQL**:
|
||||
|
||||
| Schema | Role |
|
||||
|---|---|
|
||||
| `central` | Project registry, nodes, path mappings, global concurrency, **task claims**, mesh topology helpers |
|
||||
| `project` | Tasks, agents, settings, workflows, missions, distributed task IDs (row-isolated by `project_id` + RLS) |
|
||||
| `archive` | Cold archive storage |
|
||||
|
||||
**Default (single machine):** unset `DATABASE_URL` → embedded Postgres under `~/.fusion/embedded-postgres/`. That data directory is **local to the host**. Two laptops each running embedded Postgres do **not** share a board.
|
||||
|
||||
**Multi-node (shared board):** every Fusion node sets the **same external** `DATABASE_URL` (and `DATABASE_MIGRATION_URL` when the runtime URL is a transaction pooler). All nodes share one database; execution (worktrees, agent processes) stays per node.
|
||||
|
||||
Core `central` tables (names as exposed by the data layer; SQL uses snake_case):
|
||||
|
||||
- `projects`
|
||||
- `projectHealth`
|
||||
- `centralActivityLog`
|
||||
- `globalConcurrency`
|
||||
- `nodes`
|
||||
- `peerNodes`
|
||||
- `settingsSyncState`
|
||||
- `taskClaims` (authoritative cross-node task checkout claims keyed by `(projectId, taskId)`)
|
||||
- `__meta`
|
||||
- `project_health`
|
||||
- `central_activity_log`
|
||||
- `global_concurrency`
|
||||
- `nodes` / `peer_nodes`
|
||||
- `project_node_path_mappings`
|
||||
- `task_claims` (authoritative cross-node checkout mutex keyed by `(project_id, task_id)`)
|
||||
- Topology helpers: `mesh_shared_snapshots`, `mesh_write_queue` (membership/auth retry only — **not** task-state replication)
|
||||
|
||||
Per-project task data is keyed by `projectId` in PostgreSQL's `project` schema. Each repo keeps `.fusion/project.json` as its filesystem identity marker; `.fusion/fusion.db` is read only by the one-time legacy migrator.
|
||||
|
||||
@@ -36,30 +52,47 @@ Use PostgreSQL-native backup/restore tooling for authoritative runtime data. Leg
|
||||
|
||||
`taskClaims` is the central cross-node lease mutex introduced by FN-4819 §2: claim acquisition/renewal/release happen in PostgreSQL, while per-project lease fields mirror the central winner for local scheduler/runtime consumption.
|
||||
|
||||
Peer/mesh coordination spans core + engine, with startup ownership in CLI process entrypoints:
|
||||
Legacy SQLite paths (`~/.fusion/fusion-central.db`, `<repo>/.fusion/fusion.db`) are migration/input only. Runtime writes go through the PostgreSQL schemas above.
|
||||
|
||||
- Topology visibility is now cluster-wide from any connected node: dashboard mesh reads aggregate remote local snapshots and dedupe by `nodeId`, with fallback to last-known local mesh state when a peer is temporarily unreachable.
|
||||
- Outage tolerance persistence is central and project-scoped: degraded mesh snapshots and queued write replay rows are stored with `projectId` keys so partitions in one project do not blur reconciliation state across other registered projects.
|
||||
- `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.
|
||||
- `MeshLeaseManager` in `@fusion/engine` is the single authority for stale lease detection and abandoned-work recovery across nodes.
|
||||
- Canonical replication semantics live in [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md). That protocol separates strongly coordinated shared state from append-only streams, queued replay classes, and node-local runtime state.
|
||||
- Distributed task-ID allocation is one strongly coordinated shared-state path: reserve/commit/abort are coordinator-mediated writes, and cluster-wide committed task totals come from allocator `committedClusterTaskCount` state (not per-node local task counts).
|
||||
- `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.
|
||||
`task_claims` is the cross-node lease mutex (FN-4819 §2): claim acquire/renew/release hit `central.task_claims` first; per-task lease columns on the project task row mirror the winner for scheduler/UI.
|
||||
|
||||
### Shared Postgres multi-node runbook
|
||||
|
||||
1. Provision one Postgres (local Docker, RDS, Supabase, etc.).
|
||||
2. On **every** Fusion node: `export DATABASE_URL=...` (same URL). If you use PgBouncer/Supavisor in transaction mode, also set `DATABASE_MIGRATION_URL` to a direct (non-pooled) connection for schema work.
|
||||
3. Register projects and nodes so they appear in shared `central.projects` / `central.nodes`.
|
||||
4. For each host, set `project_node_path_mappings` so that host’s absolute checkout path is recorded for each project.
|
||||
5. Run `fn serve` / the engine on each node. Task IDs and settings are shared via Postgres; checkout exclusivity uses `task_claims`; abandoned-owner recovery uses `MeshLeaseManager`.
|
||||
6. Keep provider credentials (`auth.json`) in mind: they are still file-local unless you use auth-sync. Task filesystem blobs under `.fusion/tasks/{ID}/` remain on the node that materializes them until a later blob strategy.
|
||||
|
||||
What is **not** multi-node via shared DB alone:
|
||||
|
||||
- Live agent/executor process migration mid-task
|
||||
- Scheduler failover of another node’s tick loop
|
||||
- Embedded Postgres sharing across machines
|
||||
|
||||
Canonical ownership / control-plane contract: [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md).
|
||||
|
||||
### Cluster membership and process ownership
|
||||
|
||||
- Topology visibility is cluster-wide: dashboard mesh reads aggregate node registry state (and optional remote health probes), with degraded fallback metadata when a peer HTTP probe fails.
|
||||
- `mesh_write_queue` / `mesh_shared_snapshots` are **not** a multi-leader task write log. Under shared Postgres they are limited to topology/auth retry and degraded membership reads. Task durability is the database commit itself.
|
||||
- `NodeDiscovery` and `NodeConnection` in `@fusion/core` handle discovery and remote connectivity/auth probes.
|
||||
- `PeerExchangeService` in `@fusion/engine` gossips membership (and optional `authMaterial`); it does **not** replicate tasks/settings over HTTP when nodes share Postgres.
|
||||
- `MeshLeaseManager` is the single authority for stale lease detection and abandoned-work recovery.
|
||||
- Distributed task-ID allocation uses shared `project.distributed_task_id_*` rows. Under Postgres, reserve/commit/abort always hit the local allocator against those shared rows — never a remote “coordinator” hop.
|
||||
- `runServe()` / `runDashboard()` own process-level peer-exchange + discovery lifecycle (one instance per process, after the HTTP port is known).
|
||||
- `InProcessRuntime` stays project-scoped and does **not** start mesh services.
|
||||
|
||||
## Mesh lease recovery in multi-node execution
|
||||
|
||||
Task ownership is shared as persisted lease metadata (`checkedOutBy`, `checkedOutAt`, `checkoutNodeId`, `checkoutRunId`, `checkoutLeaseRenewedAt`, `checkoutLeaseEpoch`) through the canonical mesh sync payloads.
|
||||
Task ownership is durable lease metadata on the shared task row (`checkedOutBy`, `checkedOutAt`, `checkoutNodeId`, `checkoutRunId`, `checkoutLeaseRenewedAt`, `checkoutLeaseEpoch`) plus the authoritative `central.task_claims` row.
|
||||
|
||||
When a node disappears or stops renewing ownership, recovery is routed only through `MeshLeaseManager.recoverAbandonedLease(...)`. The manager now performs a two-write release: it releases the authoritative central `taskClaims` row first, then clears per-project owner fields (`checkedOutBy`, `checkoutNodeId`, `checkoutRunId`, `checkoutLeaseRenewedAt`, `checkedOutAt`) and bumps `checkoutLeaseEpoch` locally.
|
||||
When a node disappears or stops renewing ownership, recovery is routed only through `MeshLeaseManager.recoverAbandonedLease(...)`. The manager performs a two-write release: release the central `task_claims` row first, then clear per-task owner fields and bump `checkoutLeaseEpoch`.
|
||||
|
||||
If one side succeeds and the other fails, the next scheduler/self-healing tick runs `reconcileLeaseRow(taskId)` to deterministically converge local and central lease state without a side queue. Recovery/reconciliation paths emit `task:auto-recover-lease-*` run-audit events (`...-released`, `...-already-healed`, `...-foreign-owner`, `...-central-unavailable`, `...-partial-write`, `...-reconciled`) for traceability.
|
||||
If one side succeeds and the other fails, the next scheduler/self-healing tick runs `reconcileLeaseRow(taskId)` to converge claim and task-row state. Recovery emits `task:auto-recover-lease-*` run-audit events for traceability.
|
||||
|
||||
This fencing prevents double-claims: a restarted or delayed stale owner cannot reclaim work once central ownership has been released and lease generation has advanced.
|
||||
This fencing prevents double-claims: a restarted or delayed stale owner cannot reclaim work once central ownership has been released and the lease generation has advanced.
|
||||
|
||||
## Recovering a missing central project row
|
||||
|
||||
@@ -69,7 +102,7 @@ If a project's PostgreSQL central-registry row is deleted, Fusion recovers it on
|
||||
2. If missing, it reads `<project>/.fusion/project.json` (or imports a legacy SQLite identity once).
|
||||
3. If present, central reattaches that exact `projectId` instead of creating a new one.
|
||||
|
||||
This prevents “empty workspace” regressions where project data still exists locally but is keyed to an older `projectId`.
|
||||
This prevents “empty workspace” regressions where project data still exists but is keyed to an older `projectId`.
|
||||
|
||||
PostgreSQL backups remain the first-line protection strategy, but this identity reattach path restores the path-to-project mapping without minting a new ID.
|
||||
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
# Plan: Mesh → Shared Postgres Multi-Node
|
||||
|
||||
**Date:** 2026-07-15
|
||||
**Branch:** `feature/migrate-mesh`
|
||||
**Status:** Implementation in progress (S1–S3 landed)
|
||||
|
||||
## Context
|
||||
|
||||
Fusion multi-node was built as multi-leader mesh over per-node SQLite. PostgreSQL cutover makes durable state shareable via one external `DATABASE_URL`. Mesh HTTP should own membership, optional auth material, and execution claims — not task/settings replication.
|
||||
|
||||
## Delivery slices
|
||||
|
||||
| Slice | Status | Outcome |
|
||||
|---|---|---|
|
||||
| S1 Docs + runbook | **Done** | multi-project, shared-mesh-protocol, architecture rewritten |
|
||||
| S2 Dead-code removal | **Done** | remote task-id coordinator hop removed; settings mesh path retired on live routes |
|
||||
| S3 Topology/auth-only queue | **Done** | PeerExchange enqueues/replays topology/auth only under backendMode |
|
||||
| S4 Presence simplification | Deferred | optional PG heartbeats replacing gossip |
|
||||
| S5 Claim e2e | Existing | PG claim tests in central-archive-secrets |
|
||||
| S6 Auth/blob | Deferred | later |
|
||||
|
||||
## Non-goals
|
||||
|
||||
- Scheduler failover
|
||||
- Live process migration
|
||||
- Multi-leader task writes when Postgres is down
|
||||
|
||||
## Verification
|
||||
|
||||
- `peer-exchange-service.test.ts` (34)
|
||||
- `mesh-routes.test.ts` (27)
|
||||
- `shared-mesh-state.test.ts` (4)
|
||||
@@ -1,202 +1,135 @@
|
||||
# Shared Mesh Replication Protocol (v1)
|
||||
# Shared Cluster Protocol (Postgres multi-node)
|
||||
|
||||
[← Docs index](./README.md)
|
||||
|
||||
This document is the canonical contract for Fusion multi-leader mesh replication.
|
||||
<!--
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
This document supersedes the multi-leader SQLite mesh replication contract. Durable project state is shared PostgreSQL; mesh HTTP is membership, optional auth material, and execution ownership — not a second database.
|
||||
-->
|
||||
|
||||
This document is the canonical contract for Fusion **multi-node operation on shared PostgreSQL**.
|
||||
|
||||
The historical multi-leader SQLite mesh (HTTP task replication, settings gossip, strong-write quorum, offline task write queues) is **retired**. Nodes that share `DATABASE_URL` already share durable state at the database layer.
|
||||
|
||||
## 1. Goals and non-goals
|
||||
|
||||
### Goals
|
||||
- Preserve one shared durable project state across multiple nodes.
|
||||
- Keep task and planning state strongly coordinated by default.
|
||||
- Allow local progress during peer outages via durable queues.
|
||||
- Support deterministic replay/reconciliation after recovery.
|
||||
- Expose read staleness so clients can decide whether to trust last-known global state.
|
||||
|
||||
### Runtime scope and non-goals (v1, updated for FN-4772/FN-4813)
|
||||
- `HybridExecutor` is the canonical multi-project/multi-node runtime orchestration path.
|
||||
- Scheduler failover (a peer node taking over another node's live scheduler tick loop) is an explicit non-goal.
|
||||
- Live-process state migration (moving in-memory executor/session state between nodes mid-task) is an explicit non-goal.
|
||||
- Supported alternative: lease handoff under `OwningNodeHandoffPolicy` (`park`, `reassign-to-local`, `reassign-any-healthy`) so tasks resume from durable state on the picking node.
|
||||
- Immediate global consistency for every data class remains a non-goal.
|
||||
- One shared durable project + central state across multiple Fusion nodes.
|
||||
- Exclusive execution ownership per task via central claims + lease epochs.
|
||||
- Per-node worktrees, processes, and path mappings without live process migration.
|
||||
- Explicit degraded topology reads when peer **HTTP** health probes fail (membership visibility), without inventing divergent local task truth.
|
||||
|
||||
### Non-goals
|
||||
|
||||
- Scheduler failover (a peer does not take over another node’s live scheduler tick loop).
|
||||
- Live-process / in-memory session migration mid-task.
|
||||
- Multi-leader task writes when Postgres is unavailable (if the DB is down, nodes do not queue alternate task realities over HTTP).
|
||||
- Treating embedded Postgres as a multi-host shared backend (embedded is per-machine only).
|
||||
|
||||
Supported recovery model: **lease handoff** under `OwningNodeHandoffPolicy` (`park`, `reassign-to-local`, `reassign-any-healthy`) so a healthy node resumes from **durable** task state.
|
||||
|
||||
## 2. Terms
|
||||
|
||||
- **Node**: A Fusion runtime instance participating in mesh sync.
|
||||
- **Coordinator**: Node currently responsible for committing a write intent.
|
||||
- **Intent**: Durable write proposal before global ack quorum completes.
|
||||
- **Envelope**: Wire record carrying replication metadata + payload.
|
||||
- **Epoch**: Monotonic lease/fencing generation for coordinator authority.
|
||||
- **Fence token**: `epoch + coordinatorNodeId + sequence` token that invalidates stale coordinators.
|
||||
- **Queue entry**: Durable locally-accepted write waiting for replay.
|
||||
- **Node**: A Fusion runtime/API process with a registered `central.nodes` row and local execution capacity.
|
||||
- **Shared database**: One Postgres cluster (schemas `project`, `central`, `archive`) reached via the same `DATABASE_URL` on every participating node.
|
||||
- **Claim**: Authoritative ownership row in `central.task_claims` keyed by `(projectId, taskId)`.
|
||||
- **Lease epoch**: Monotonic fencing generation on the task row that invalidates stale owners after recovery.
|
||||
- **Membership gossip**: Optional peer HTTP exchange of known peers / metrics; does not carry task or settings payloads under Postgres.
|
||||
- **Auth material**: Provider credentials in per-machine `auth.json` (not in the shared DB by default); optional secure HTTP sync remains.
|
||||
|
||||
## 3. Versioning
|
||||
|
||||
- Protocol id: `fusion.shared-mesh`
|
||||
- Initial version: `1.0`
|
||||
- All envelopes must include `{ protocol, version }`.
|
||||
- Minor versions (`1.x`) are backward-compatible additive.
|
||||
- Major versions (`2.0+`) may change semantics and require explicit compatibility checks.
|
||||
|
||||
## 4. Data-class coordination matrix
|
||||
## 3. Data-class matrix (current truth)
|
||||
|
||||
| Data class | Mode | Notes |
|
||||
|---|---|---|
|
||||
| Tasks (core fields, deps, steps, column transitions) | Strongly coordinated | Quorum-acked intent/commit path; replayable with fencing |
|
||||
| Task metadata (priority, model overrides, docs metadata refs) | Strongly coordinated | Same write path as tasks |
|
||||
| Missions/milestones/slices/features | Strongly coordinated | Ordered writes preserve hierarchy invariants |
|
||||
| Agent definitions/configuration | Strongly coordinated | Durable config replicated; runtime process handles excluded |
|
||||
| Agent runtime state (heartbeat ticks, local process internals, worktree paths) | Node-local only | Exposed as local telemetry, not global truth |
|
||||
| Project settings | Strongly coordinated | Existing settings payloads remain canonical payload shape |
|
||||
| Auth material / provider credentials | Queued-for-later (secured transport only) | Explicit auth snapshot channel (`sharedState.authMaterial`); never merged as ordinary settings payload |
|
||||
| Execution runs / live activity streams | Node-local + queued summary | Live events local; durable run outcomes appended later |
|
||||
| Audit / event streams (`activityLog`, `runAuditEvents`) | Append-only replicated | Immutable event replication with origin metadata |
|
||||
| Filesystem blobs (`.fusion/tasks/*` prompts/logs/attachments) | Queued-for-later | Metadata in replicated records, blob transfer out-of-band |
|
||||
| Tasks, deps, steps, columns | **Shared Postgres** | Commit is cluster-visible; no HTTP task replication |
|
||||
| Missions / agents config / workflows / audit | **Shared Postgres** | Same |
|
||||
| Project + global settings | **Shared Postgres** | Settings HTTP push/pull between nodes is disabled (`409`) |
|
||||
| Distributed task IDs | **Shared Postgres** | `distributed_task_id_state` / `_reservations`; always local allocator against shared rows |
|
||||
| Checkout ownership | **Central claim + task mirror** | `task_claims` then task lease columns |
|
||||
| Agent runtime / worktrees / live sessions | **Node-local** | Paths may differ via `project_node_path_mappings` |
|
||||
| Auth credentials (`auth.json`) | **Node-local + optional sync** | `sharedState.authMaterial` / auth routes only |
|
||||
| FS blobs (`.fusion/tasks/*`) | **Node-local** | Metadata may be in PG; bytes on the materializing host |
|
||||
| Topology / peer metrics | **Registry + probes** | `central.nodes` / peers; optional gossip + health HTTP |
|
||||
|
||||
## 5. Write classes
|
||||
## 4. Execution ownership
|
||||
|
||||
- **`strong`**: Requires coordinator fence + quorum ack before `committed`.
|
||||
- **`append-only`**: Event-style immutable replication; dedupe by event id.
|
||||
- **`queued`**: Accept locally when peers unavailable; replay later.
|
||||
- **`local`**: Never replicated globally.
|
||||
### Claim path
|
||||
|
||||
## 6. Replication envelope
|
||||
1. `AgentStore.checkoutTask` → `CentralClaimStore.tryClaimTask` (`central.task_claims`).
|
||||
2. Mirror winner onto the task row (`tryClaimCheckout`: `checkedOutBy`, `checkoutNodeId`, `checkoutRunId`, `checkoutLeaseRenewedAt`, `checkoutLeaseEpoch`).
|
||||
3. Scheduler/executor on the winning node run locally; other nodes must not start a second exclusive execution lane for the same claim.
|
||||
|
||||
Every replicated record uses:
|
||||
- `protocol`, `version`
|
||||
- `recordId`, `entityType`, `entityId`
|
||||
- `originNodeId`, `originSeq`
|
||||
- `writeClass`
|
||||
- `leaseEpoch`, `fenceToken`
|
||||
- `intentId` and `state` (`intent` | `committed` | `rejected` | `queued` | `reconciled`)
|
||||
- `createdAt`, `committedAt?`
|
||||
- `payload`
|
||||
- `precondition?` (base revision / expected epoch)
|
||||
### Recovery path
|
||||
|
||||
`PeerSyncRequest` / `PeerSyncResponse` remain mesh exchange carriers. v1 envelopes are payloads exchanged through current mesh sync infrastructure and follow-on sync endpoints.
|
||||
Only `MeshLeaseManager.recoverAbandonedLease(...)`:
|
||||
|
||||
### Auth snapshot contract (v1)
|
||||
1. Prove recoverable (owner offline/error, or lease/heartbeat stale; not active local execution).
|
||||
2. Apply handoff policy when configured.
|
||||
3. Release **central claim first**, then clear task lease fields and **bump epoch**.
|
||||
4. Requeue to `todo` (preserve progress when appropriate).
|
||||
5. Partial split-brain → `reconcileLeaseRow` on a later tick.
|
||||
|
||||
Auth replication uses `AuthMaterialSnapshot` (`version`, `exportedAt`, `checksum`, `payload`) with:
|
||||
- `payload.providerAuth: Record<string, ProviderAuthEntry>`
|
||||
- `ProviderAuthEntry.type`: `api_key | oauth`
|
||||
- `api_key` fields: `key`
|
||||
- `oauth` fields: `accessToken`, `refreshToken`, `expires`, optional `accountId`
|
||||
Run-audit: `task:auto-recover-lease-*`, `node:lease:*`, `node:handoff:*` as applicable.
|
||||
|
||||
Transport paths:
|
||||
- Mesh shared-state channel: `POST /api/mesh/sync` (`sharedState.authMaterial`)
|
||||
- Explicit node auth channel: `POST /api/nodes/:id/auth/sync` and inbound `POST /api/settings/auth-receive` / `GET /api/settings/auth-export`
|
||||
## 5. Membership and HTTP mesh surfaces
|
||||
|
||||
Security/redaction rules:
|
||||
- Auth snapshots are only exchanged over API-key-authenticated node links.
|
||||
- Raw secrets (`key`, `accessToken`, `refreshToken`, bearer headers) MUST NOT be logged.
|
||||
- Route diagnostics may emit provider names/counts only.
|
||||
Still useful under shared Postgres:
|
||||
|
||||
## 7. Quorum and acknowledgements
|
||||
| Surface | Role |
|
||||
|---|---|
|
||||
| `GET /api/mesh/state` | Topology snapshot for dashboard Nodes UI |
|
||||
| `POST /api/mesh/sync` | Peer gossip: `knownPeers` (+ optional `authMaterial` only) |
|
||||
| `POST/GET /api/mesh/task-ids/*` | Local allocator against shared ID tables (no remote coordinator hop) |
|
||||
| Auth sync routes | Optional credential fan-out for file-local auth |
|
||||
| mDNS discovery | Join convenience, not task SoT |
|
||||
| Docker mesh config generator | Provision managed peers |
|
||||
|
||||
For `strong` writes:
|
||||
1. Coordinator accepts intent locally.
|
||||
2. Coordinator requests acknowledgements from peers in current membership view.
|
||||
3. Commit requires `quorum = floor(eligibleVoters / 2) + 1` including coordinator.
|
||||
4. If quorum fails before timeout, intent becomes `queued` with retry metadata.
|
||||
Removed / disabled:
|
||||
|
||||
`append-only` writes can be accepted locally and replicated asynchronously, but must preserve origin ordering `(originNodeId, originSeq)`.
|
||||
| Surface | Status |
|
||||
|---|---|
|
||||
| `POST /api/mesh/tasks/create` | Removed — DB is the replication plane |
|
||||
| Task/agent/mission/audit shared-state domains | Removed |
|
||||
| Settings gossip / node settings push-pull | Disabled on Postgres (`409`) |
|
||||
| Remote task-ID coordinator forwarding | Disabled on Postgres |
|
||||
|
||||
## 8. Lease epochs and fencing
|
||||
## 6. Write queue and degraded topology (narrowed)
|
||||
|
||||
- Coordinator authority is leased with a monotonic `leaseEpoch`.
|
||||
- Any write with stale epoch/fence must be rejected (`fenced`).
|
||||
- Restarted nodes must reacquire lease and increment epoch before coordinating strong writes.
|
||||
- Replay workers must carry original fence metadata; reconciler can reject stale queued entries after epoch advancement.
|
||||
Historical multi-leader design used `meshWriteQueue` for offline **task** write replay and `meshSharedSnapshots` for last-known global task state.
|
||||
|
||||
## 9. Offline queueing and replay
|
||||
Under shared Postgres:
|
||||
|
||||
When a strong/queued write cannot reach quorum:
|
||||
- Persist queue entry durably in `meshWriteQueue` (`status`: `pending | replaying | applied | failed`).
|
||||
- Retryable queueing is limited to transport/outage failures: HTTP `502/503/504`, timeout/abort, and transport errors (`TypeError` fetch/network rejections, or Node-style `ECONNREFUSED`, `ENOTFOUND`, `ETIMEDOUT`, `ECONNRESET`).
|
||||
- Non-retryable HTTP failures (`400/401/403/404/409/422`) are recorded as immediate failures and are not queued for replay.
|
||||
- `applied` and `failed` rows are retained as durable reconciliation history (no auto-cleanup in v1).
|
||||
- **Do not** invent local task commits when Postgres is unavailable.
|
||||
- `meshWriteQueue` is limited to **topology / auth** retry classes (membership sync / auth material), not task or settings payloads.
|
||||
- `meshSharedSnapshots` support **degraded membership/topology** reads only; they are not a substitute board store.
|
||||
- `PeerExchangeService.replayPendingWritesForNode` replays only those narrow scopes.
|
||||
|
||||
Replay ordering:
|
||||
1. Sort by `(createdAt asc, id asc)`.
|
||||
2. Transition idempotently on the same row (`pending → replaying → applied|failed`) while incrementing `attemptCount` on each attempt.
|
||||
3. Re-validate preconditions/fencing and record deterministic outcome.
|
||||
If Postgres is down, operators fix the database; nodes do not multi-master task rows over HTTP.
|
||||
|
||||
## 10. Reconciliation
|
||||
## 7. Process lifecycle
|
||||
|
||||
Reconciliation outcomes are explicit:
|
||||
- `applied` — replayed successfully.
|
||||
- `noop_already_applied` — idempotent duplicate.
|
||||
- `superseded` — newer committed revision already exists.
|
||||
- `conflict_requires_merge` — semantic conflict; requires policy/agent/manual resolution.
|
||||
- `rejected_fenced` — stale epoch/fence.
|
||||
- `fn serve` / `fn dashboard` start one process-wide `PeerExchangeService` and call `CentralCore.startDiscovery()` after the HTTP server binds the real port.
|
||||
- `InProcessRuntime` is project-scoped (scheduler/executor/heartbeat) and does **not** start mesh services.
|
||||
- `HybridExecutor` remains the multi-project / multi-node orchestration path when the hybrid gate enables it.
|
||||
|
||||
Conflict policy must never silently downgrade strong writes to local-only updates.
|
||||
## 8. Security boundary
|
||||
|
||||
## 11. Restart recovery hooks
|
||||
- Peer HTTP (sync, auth, remote isolation runtime) requires node API-key authentication when configured.
|
||||
- Never log raw secrets from auth snapshots.
|
||||
- Database credentials in `DATABASE_URL` must not appear in logs (redaction helpers in the Postgres connection layer).
|
||||
|
||||
On node startup:
|
||||
1. Load durable queue.
|
||||
2. Rebuild last known lease epoch / origin sequence.
|
||||
3. Mark in-flight intents without terminal state as `queued` recovery candidates.
|
||||
4. Start replay loop only after mesh membership snapshot and lease status are known.
|
||||
## 9. Operator checklist
|
||||
|
||||
## 12. Degraded reads and staleness
|
||||
See the **Shared Postgres multi-node runbook** in [`docs/multi-project.md`](./multi-project.md).
|
||||
|
||||
Mesh state reads expose a canonical `MeshDegradedReadState` contract:
|
||||
Short form:
|
||||
|
||||
```ts
|
||||
type MeshDegradedReadState = {
|
||||
mode: "fresh" | "degraded";
|
||||
asOf: string;
|
||||
sourceNodeId: string | null;
|
||||
snapshotVersion: string | null;
|
||||
stalenessMs: number;
|
||||
queueDepth: number;
|
||||
pendingWriteCount: number;
|
||||
failedWriteCount: number;
|
||||
};
|
||||
```
|
||||
1. Same external `DATABASE_URL` on every node.
|
||||
2. Register nodes/projects + path mappings per host.
|
||||
3. Run engines; claims enforce exclusive execution.
|
||||
4. Expect worktrees/auth/blobs to remain node-local unless you opt into auth-sync or a future blob store.
|
||||
|
||||
Rules:
|
||||
- `mode="fresh"` only when data came from the live mesh read path.
|
||||
- `mode="degraded"` when read falls back to `meshSharedSnapshots`.
|
||||
- `asOf` comes from snapshot `capturedAt`; `stalenessMs = Date.now() - new Date(asOf).getTime()`.
|
||||
- `snapshotVersion` is the stored 64-char SHA-256 hex digest of snapshot payload.
|
||||
- `queueDepth` is computed from queue rows where `status IN ('pending','replaying','failed')`.
|
||||
## 10. Historical note
|
||||
|
||||
API behavior must never hide fallback mode: degraded reads are explicit so clients can distinguish stale last-known state from fresh cluster state.
|
||||
|
||||
Concrete exported/runtime surfaces:
|
||||
- Core types are exported from `@fusion/core`: `MeshSnapshotQuery`, `MeshSnapshotRecord`, `MeshSnapshotRecordInput`, `MeshWriteQueueStatus`, `MeshWriteQueueEntry`, `MeshWriteQueueInput`, `MeshWriteQueueFilter`, `MeshWriteApplyResult`, `MeshWriteFailureResult`, `MeshWriteReplaySummary`, and `MeshDegradedReadState`.
|
||||
- `CentralCore` persistence/assertion methods: `recordMeshSnapshot`, `getLatestMeshSnapshot`, `enqueueMeshWrite`, `listPendingMeshWrites`, `markMeshWriteReplayStarted`, `markMeshWriteApplied`, `markMeshWriteFailed`, and `getMeshDegradedReadState`.
|
||||
- Runtime replay/assertion methods: `PeerExchangeService.replayPendingWritesForNode(targetNodeId)` and `NodeHealthMonitor` recovery callback `onNodeRecovered(nodeId, previousStatus)`.
|
||||
|
||||
## 13. End-to-end v1 write path
|
||||
|
||||
1. **Intent creation**: Node creates write intent + envelope.
|
||||
2. **Coordinator selection**: Node routes to current coordinator lease holder for the entity scope.
|
||||
3. **Commit/ack**:
|
||||
- strong: quorum commit
|
||||
- append-only: local append + async replication
|
||||
4. **Fallback**: if unreachable/quorum-fail, persist queue entry (`queued`).
|
||||
5. **Replay**: on recovery, replay durable queue in canonical order with fencing checks.
|
||||
6. **Reconciliation**: produce explicit outcome and update entity revision state.
|
||||
|
||||
## 14. Contract for FN-3449 through FN-3456
|
||||
|
||||
Follow-on tasks must implement against this contract and not redefine it:
|
||||
- **FN-3449**: distributed ids/origin sequence allocation + monotonic ordering.
|
||||
- **FN-3450**: coordinator selection and lease management runtime.
|
||||
- **FN-3451**: strong-write commit path + quorum ack handling.
|
||||
- **FN-3452**: durable offline queue persistence and replay engine.
|
||||
- **FN-3453**: reconciliation executor + conflict outcome handling.
|
||||
- **FN-3454**: restart recovery bootstrap and in-flight intent recovery.
|
||||
- **FN-3455**: degraded-read APIs exposing staleness metadata.
|
||||
- **FN-3456**: partition behavior policy, observability, and operator controls.
|
||||
|
||||
## 15. Security boundary
|
||||
|
||||
- Mesh transport authentication (node API keys / trust) is mandatory for replication traffic.
|
||||
- Auth credential replication is explicit and separately controlled from ordinary settings replication.
|
||||
- Sensitive payloads must be redacted from non-secure logs and diagnostics.
|
||||
Earlier revisions of this file described protocol id `fusion.shared-mesh` v1 with strong/queued/append-only write classes and quorum acks for multi-leader SQLite. That contract is archived by this rewrite. Implementation remnants that still mention multi-leader envelopes are compatibility shims and must not reintroduce HTTP task replication.
|
||||
|
||||
@@ -43,6 +43,17 @@ describe("plugin-sdk export surface", () => {
|
||||
expect(tsupRaw).toContain("/^@fusion\\//");
|
||||
});
|
||||
|
||||
it("uses a runtime-only core shim that bundles schema source without requiring core dist", () => {
|
||||
const tsupPath = join(workspaceRoot, "packages", "cli", "tsup.config.ts");
|
||||
const tsupRaw = readFileSync(tsupPath, "utf-8");
|
||||
const shimPath = join(workspaceRoot, "packages", "cli", "src", "plugin-sdk-core-runtime-shim.mjs");
|
||||
const shimRaw = readFileSync(shimPath, "utf-8");
|
||||
|
||||
expect(tsupRaw).toContain('"plugin-sdk-core-runtime-shim.mjs"');
|
||||
expect(shimRaw).toContain('from "../../core/src/postgres/schema/index.js"');
|
||||
expect(shimRaw).not.toContain("../../core/dist/");
|
||||
});
|
||||
|
||||
it("has no @fusion runtime specifiers in built plugin-sdk artifact when present", () => {
|
||||
const distPath = join(workspaceRoot, "packages", "cli", "dist", "plugin-sdk", "index.js");
|
||||
if (!existsSync(distPath)) {
|
||||
|
||||
27
packages/cli/src/plugin-sdk-core-runtime-shim.mjs
Normal file
27
packages/cli/src/plugin-sdk-core-runtime-shim.mjs
Normal file
@@ -0,0 +1,27 @@
|
||||
/*
|
||||
* FNXC:BundledPlugins 2026-07-15-13:11:
|
||||
* Clean CI typechecks the CLI before @fusion/core emits dist, but published bundled plugins still need postgresSchema runtime values. Keep this alias implementation in an untyped .mjs module so CLI tsc does not cross the package rootDir boundary; esbuild follows the core source import and inlines the schema into each bundled.js artifact, leaving no private @fusion/core runtime dependency.
|
||||
*/
|
||||
import * as postgresSchema from "../../core/src/postgres/schema/index.js";
|
||||
|
||||
export { postgresSchema };
|
||||
|
||||
export const WORKFLOW_EXTENSION_SCHEMA_VERSION = 1;
|
||||
|
||||
export function workflowExtensionRegistryId(pluginId, extensionId) {
|
||||
return `plugin:${pluginId}:${extensionId}`;
|
||||
}
|
||||
|
||||
export function createBoardActionServices(store) {
|
||||
return {
|
||||
moveTask(input) {
|
||||
return store.moveTask(input.taskId, input.column, {
|
||||
preserveProgress: input.preserveProgress,
|
||||
moveSource: input.source ?? "user",
|
||||
});
|
||||
},
|
||||
updateTask(input) {
|
||||
return store.updateTask(input.taskId, input.updates);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,39 +0,0 @@
|
||||
import type { BoardActionTaskStore, ColumnId, Task } from "@fusion/core";
|
||||
|
||||
/**
|
||||
* FNXC:BundledPlugins 2026-07-15-00:00:
|
||||
* Reports and CLI Printing Press import the `postgresSchema` runtime namespace while the CLI bundler aliases `@fusion/core` to this shim for published plugin bundles. Re-export the concrete core schema build artifact here so npm-installed `bundled.js` files resolve `postgresSchema.plugin` without keeping a bare private `@fusion/core` runtime specifier or crashing with `Cannot find package '@fusion/core'`; using dist keeps @runfusion/fusion typecheck inside its package root while esbuild still bundles the schema runtime values.
|
||||
*/
|
||||
export * as postgresSchema from "../../core/dist/postgres/schema/index.js";
|
||||
|
||||
export const WORKFLOW_EXTENSION_SCHEMA_VERSION = 1 as const;
|
||||
|
||||
export function workflowExtensionRegistryId(pluginId: string, extensionId: string): string {
|
||||
return `plugin:${pluginId}:${extensionId}`;
|
||||
}
|
||||
|
||||
export interface MoveBoardTaskInput {
|
||||
taskId: string;
|
||||
column: ColumnId;
|
||||
preserveProgress?: boolean;
|
||||
source?: "user" | "engine" | "scheduler";
|
||||
}
|
||||
|
||||
export interface UpdateBoardTaskInput {
|
||||
taskId: string;
|
||||
updates: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export function createBoardActionServices(store: BoardActionTaskStore) {
|
||||
return {
|
||||
moveTask(input: MoveBoardTaskInput): Promise<Task> {
|
||||
return store.moveTask(input.taskId, input.column, {
|
||||
preserveProgress: input.preserveProgress,
|
||||
moveSource: input.source ?? "user",
|
||||
});
|
||||
},
|
||||
updateTask(input: UpdateBoardTaskInput): Promise<Task> {
|
||||
return store.updateTask(input.taskId, input.updates);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -66,7 +66,7 @@ const compoundEngineeringPluginSrc = join(__dirname, "..", "..", "plugins", "fus
|
||||
const compoundEngineeringPluginDest = join(__dirname, "dist", "plugins", "fusion-plugin-compound-engineering");
|
||||
const linearImportPluginSrc = join(__dirname, "..", "..", "plugins", "fusion-plugin-linear-import");
|
||||
const linearImportPluginDest = join(__dirname, "dist", "plugins", "fusion-plugin-linear-import");
|
||||
const pluginSdkCoreRuntimeShim = join(__dirname, "src", "plugin-sdk-core-runtime-shim.ts");
|
||||
const pluginSdkCoreRuntimeShim = join(__dirname, "src", "plugin-sdk-core-runtime-shim.mjs");
|
||||
const dashboardClientStub = `<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
|
||||
@@ -9,10 +9,9 @@ import {
|
||||
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-12:
|
||||
Task/state mesh replication is REMOVED (replication is handled at the
|
||||
PostgreSQL level), so only the surviving settings-adjacent snapshot pair —
|
||||
projectSettings and authMaterial — plus the envelope/checksum plumbing are
|
||||
covered here.
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Task/state mesh replication is REMOVED. Live mesh routes exchange authMaterial
|
||||
only; projectSettings helpers remain for legacy envelope tests/checksum plumbing.
|
||||
*/
|
||||
describe("shared-mesh-state", () => {
|
||||
const exportedAt = "2026-05-04T00:00:00.000Z";
|
||||
|
||||
@@ -9,14 +9,12 @@ import type {
|
||||
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-12:
|
||||
Task/state mesh replication is REMOVED — all replication is handled at the
|
||||
PostgreSQL level (nodes share the database). The task-metadata, agent,
|
||||
agent-run, activity-log, run-audit, and mission-hierarchy snapshot types and
|
||||
creators that used to live here are gone with it. What remains is the
|
||||
settings-adjacent pair still exchanged over the mesh: projectSettings (legacy
|
||||
sqlite topology settings sync) and authMaterial (auth.json is per-machine
|
||||
file state — the one domain that does not live in the database), plus the
|
||||
shared envelope/checksum plumbing they ride on.
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Task/state mesh replication is REMOVED — shared PostgreSQL is the SoT.
|
||||
createProjectSettingsSnapshot remains for legacy helpers/tests only; live mesh
|
||||
routes no longer apply or emit projectSettings. authMaterial stays on the wire
|
||||
because auth.json is per-machine file state, plus the shared envelope/checksum
|
||||
plumbing.
|
||||
*/
|
||||
|
||||
export const SHARED_STATE_DEFAULT_LIMIT = 10_000;
|
||||
|
||||
@@ -5598,12 +5598,13 @@ export interface MeshDegradedReadState {
|
||||
export interface SharedMeshStatePayload {
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-12:
|
||||
Task/state mesh replication is REMOVED — replication is handled at the
|
||||
PostgreSQL level (nodes share the database). Only the settings-adjacent
|
||||
domains remain on the wire: projectSettings (legacy sqlite settings sync)
|
||||
and authMaterial (per-machine auth.json). Receivers ignore any other
|
||||
domain a legacy peer may still send.
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Task/state mesh replication is REMOVED — shared PostgreSQL is the SoT.
|
||||
projectSettings is deprecated on the wire (ignored by receivers; settings
|
||||
live in the shared DB). authMaterial remains (per-machine auth.json).
|
||||
Receivers ignore any other domain a legacy peer may still send.
|
||||
*/
|
||||
/** @deprecated Ignored under shared Postgres; kept for wire compatibility with old peers. */
|
||||
projectSettings?: SnapshotBase & { payload: { global: GlobalSettings; projects?: Record<string, ProjectSettings> } };
|
||||
authMaterial?: SnapshotBase & { payload: { providerAuth?: Record<string, ProviderAuthEntry> } };
|
||||
}
|
||||
|
||||
@@ -33,6 +33,7 @@ const mockAbortDistributedTaskIdReservation = vi.fn();
|
||||
const mockGetDistributedTaskIdState = vi.fn();
|
||||
const mockApplyReplicatedTaskCreate = vi.fn();
|
||||
const mockApplyAuthMaterialSnapshot = vi.fn();
|
||||
const mockGetAuthMaterialSnapshot = vi.fn();
|
||||
|
||||
// Mock GlobalSettingsStore
|
||||
const mockGetSettings = vi.fn().mockResolvedValue({});
|
||||
@@ -58,6 +59,7 @@ vi.mock("@fusion/core", async () => {
|
||||
getSettingsForSync: mockGetSettingsForSync,
|
||||
applyRemoteSettings: mockApplyRemoteSettings,
|
||||
applyAuthMaterialSnapshot: mockApplyAuthMaterialSnapshot,
|
||||
getAuthMaterialSnapshot: mockGetAuthMaterialSnapshot,
|
||||
}; }),
|
||||
// FNXC:PostgresCutover 2026-07-10: the mesh sync response path constructs a
|
||||
// REAL AgentStore for the agents/agentRuns shared-state snapshots; the
|
||||
@@ -102,17 +104,20 @@ vi.mock("@fusion/engine", async (importOriginal) => {
|
||||
});
|
||||
|
||||
class MockStore extends EventEmitter {
|
||||
// FNXC:PostgresCutover 2026-07-10: createServer resolves the chat/session
|
||||
// layers via store.getAsyncLayer(); null = legacy mode for this mock (this
|
||||
// single missing method had the whole file red since the cutover).
|
||||
getAsyncLayer(): null {
|
||||
return null;
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-10-00:00:
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
createServer requires a project PostgreSQL AsyncDataLayer for ChatStore /
|
||||
AgentStore construction. Mesh route tests only exercise HTTP topology and
|
||||
allocator routes, so a minimal stub layer is enough — real query paths are
|
||||
covered by pg harness suites.
|
||||
*/
|
||||
getAsyncLayer(): { projectId: string } {
|
||||
return { projectId: "mesh-routes-test-project" };
|
||||
}
|
||||
|
||||
// Settings sync stays enabled for this mock (sqlite topology); the PG-mode
|
||||
// 409 gating is covered in routes-system.test.ts.
|
||||
get backendMode(): boolean {
|
||||
return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
getRootDir(): string {
|
||||
@@ -196,6 +201,29 @@ function makeNodeConfig(overrides: Partial<Record<string, unknown>> = {}) {
|
||||
};
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
createServer now boots ChatStore/AiSessionStore against the project PG layer and
|
||||
fire-and-forgets recoverStaleSessions. Mesh route unit tests inject inert stores
|
||||
so createServer does not touch a stub layer's query builders.
|
||||
*/
|
||||
function createMeshTestServer(store: TaskStore, extra: Record<string, unknown> = {}) {
|
||||
const chatStore = Object.assign(new EventEmitter(), {
|
||||
deleteSessionsForAgentId: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
const aiSessionStore = Object.assign(new EventEmitter(), {
|
||||
recoverStaleSessions: vi.fn().mockResolvedValue(undefined),
|
||||
rehydrateFromStore: vi.fn().mockResolvedValue(0),
|
||||
stopScheduledCleanup: vi.fn(),
|
||||
cleanupStaleSessions: vi.fn().mockResolvedValue({ terminalDeleted: 0, orphanedDeleted: 0 }),
|
||||
});
|
||||
return createServer(store, {
|
||||
chatStore: chatStore as never,
|
||||
aiSessionStore: aiSessionStore as never,
|
||||
...extra,
|
||||
});
|
||||
}
|
||||
|
||||
type RuntimeLogEntry = {
|
||||
level: "info" | "warn" | "error";
|
||||
scope: string;
|
||||
@@ -289,7 +317,7 @@ describe("POST /api/mesh/sync", () => {
|
||||
]);
|
||||
|
||||
const store = new MockStore();
|
||||
app = createServer(store as unknown as TaskStore);
|
||||
app = createMeshTestServer(store as unknown as TaskStore);
|
||||
});
|
||||
|
||||
it("should merge peers and return sync response", async () => {
|
||||
@@ -533,7 +561,12 @@ describe("POST /api/mesh/sync", () => {
|
||||
expect(newPeerIds).not.toContain("node_b");
|
||||
});
|
||||
|
||||
describe("settings sync", () => {
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Mesh settings gossip is retired. Shared PostgreSQL is the settings SoT; mesh
|
||||
sync ignores inbound settings payloads and never echoes settings in responses.
|
||||
*/
|
||||
describe("settings sync (retired under shared PostgreSQL)", () => {
|
||||
function makeSettingsPayload(overrides: Partial<Record<string, unknown>> = {}) {
|
||||
return {
|
||||
exportedAt: "2026-04-01T00:00:00.000Z",
|
||||
@@ -548,23 +581,19 @@ describe("POST /api/mesh/sync", () => {
|
||||
mockGetSettingsForSync.mockReset();
|
||||
mockApplyRemoteSettings.mockReset();
|
||||
mockGetSettings.mockReset();
|
||||
mockGetAuthMaterialSnapshot.mockReset();
|
||||
mockGetAuthMaterialSnapshot.mockReturnValue(undefined);
|
||||
});
|
||||
|
||||
it("should apply settings when remote checksum differs", async () => {
|
||||
it("ignores inbound settings and does not apply or echo them", async () => {
|
||||
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
|
||||
const localPayload = makeSettingsPayload({ checksum: "local-checksum" });
|
||||
|
||||
mockGetSettings.mockResolvedValue({});
|
||||
mockGetSettingsForSync.mockResolvedValue(localPayload);
|
||||
mockApplyRemoteSettings.mockResolvedValue({
|
||||
success: true,
|
||||
globalCount: 5,
|
||||
projectCount: 2,
|
||||
authCount: 1,
|
||||
const runtimeHarness = createRuntimeLoggerHarness();
|
||||
const appWithLogger = createMeshTestServer(new MockStore() as unknown as TaskStore, {
|
||||
runtimeLogger: runtimeHarness.logger,
|
||||
});
|
||||
|
||||
const response = await request(
|
||||
app,
|
||||
appWithLogger,
|
||||
"POST",
|
||||
"/api/mesh/sync",
|
||||
JSON.stringify({
|
||||
@@ -577,71 +606,26 @@ describe("POST /api/mesh/sync", () => {
|
||||
{ "Content-Type": "application/json" }
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(mockApplyRemoteSettings).toHaveBeenCalledWith(remotePayload);
|
||||
expect(mockGetSettingsForSync).toHaveBeenCalled();
|
||||
expect((response.body as any).settings).toBeDefined();
|
||||
expect((response.body as any).settings.checksum).toBe("local-checksum");
|
||||
});
|
||||
|
||||
it("should skip applying settings when checksums match", async () => {
|
||||
const samePayload = makeSettingsPayload({ checksum: "same-checksum" });
|
||||
|
||||
mockGetSettings.mockResolvedValue({});
|
||||
mockGetSettingsForSync.mockResolvedValue(samePayload);
|
||||
|
||||
const response = await request(
|
||||
app,
|
||||
"POST",
|
||||
"/api/mesh/sync",
|
||||
JSON.stringify({
|
||||
senderNodeId: "node_remote",
|
||||
senderNodeUrl: "https://remote.example.com",
|
||||
knownPeers: [],
|
||||
timestamp: "2026-04-01T12:00:00.000Z",
|
||||
settings: samePayload,
|
||||
}),
|
||||
{ "Content-Type": "application/json" }
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(mockApplyRemoteSettings).not.toHaveBeenCalled();
|
||||
expect((response.body as any).settings).toBeDefined();
|
||||
});
|
||||
|
||||
it("should respond with settings when request includes settings", async () => {
|
||||
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
|
||||
const localPayload = makeSettingsPayload({ checksum: "local-checksum" });
|
||||
|
||||
mockGetSettings.mockResolvedValue({});
|
||||
mockGetSettingsForSync.mockResolvedValue(localPayload);
|
||||
mockApplyRemoteSettings.mockResolvedValue({
|
||||
success: true,
|
||||
globalCount: 1,
|
||||
projectCount: 0,
|
||||
authCount: 0,
|
||||
});
|
||||
|
||||
const response = await request(
|
||||
app,
|
||||
"POST",
|
||||
"/api/mesh/sync",
|
||||
JSON.stringify({
|
||||
senderNodeId: "node_remote",
|
||||
senderNodeUrl: "https://remote.example.com",
|
||||
knownPeers: [],
|
||||
timestamp: "2026-04-01T12:00:00.000Z",
|
||||
settings: remotePayload,
|
||||
expect(mockGetSettingsForSync).not.toHaveBeenCalled();
|
||||
expect((response.body as any).settings).toBeUndefined();
|
||||
expect(mockMergePeers).toHaveBeenCalled();
|
||||
expect(runtimeHarness.entries).toContainEqual(
|
||||
expect.objectContaining({
|
||||
level: "info",
|
||||
scope: "test:routes:remote-route:mesh-sync",
|
||||
message: "Ignored inbound settings payload — settings live in shared PostgreSQL",
|
||||
context: expect.objectContaining({
|
||||
nodeId: "node_remote",
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "settings-sync",
|
||||
}),
|
||||
}),
|
||||
{ "Content-Type": "application/json" }
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect((response.body as any).settings).toBeDefined();
|
||||
expect((response.body as any).settings.checksum).toBe("local-checksum");
|
||||
});
|
||||
|
||||
it("should NOT include settings in response when request does not include settings", async () => {
|
||||
it("does not include settings in response when request has no settings", async () => {
|
||||
const response = await request(
|
||||
app,
|
||||
"POST",
|
||||
@@ -659,104 +643,6 @@ describe("POST /api/mesh/sync", () => {
|
||||
expect((response.body as any).settings).toBeUndefined();
|
||||
expect(mockGetSettingsForSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should not fail sync when settings apply fails", async () => {
|
||||
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
|
||||
const localPayload = makeSettingsPayload({ checksum: "local-checksum" });
|
||||
const runtimeHarness = createRuntimeLoggerHarness();
|
||||
const appWithLogger = createServer(new MockStore() as unknown as TaskStore, {
|
||||
runtimeLogger: runtimeHarness.logger,
|
||||
});
|
||||
|
||||
mockGetSettings.mockResolvedValue({});
|
||||
mockGetSettingsForSync.mockResolvedValue(localPayload);
|
||||
mockApplyRemoteSettings.mockResolvedValue({
|
||||
success: false,
|
||||
globalCount: 0,
|
||||
projectCount: 0,
|
||||
authCount: 0,
|
||||
error: "Checksum mismatch",
|
||||
});
|
||||
|
||||
const response = await request(
|
||||
appWithLogger,
|
||||
"POST",
|
||||
"/api/mesh/sync",
|
||||
JSON.stringify({
|
||||
senderNodeId: "node_remote",
|
||||
senderNodeUrl: "https://remote.example.com",
|
||||
knownPeers: [],
|
||||
timestamp: "2026-04-01T12:00:00.000Z",
|
||||
settings: remotePayload,
|
||||
}),
|
||||
{ "Content-Type": "application/json" }
|
||||
);
|
||||
|
||||
// Sync should still succeed even if settings apply failed
|
||||
expect(response.status).toBe(200);
|
||||
expect(mockMergePeers).toHaveBeenCalled();
|
||||
expect((response.body as any).knownPeers).toBeDefined();
|
||||
expect(runtimeHarness.entries).toContainEqual(
|
||||
expect.objectContaining({
|
||||
level: "warn",
|
||||
scope: "test:routes:remote-route:mesh-sync",
|
||||
message: "Failed to apply remote settings payload",
|
||||
context: expect.objectContaining({
|
||||
nodeId: "node_remote",
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "apply-remote-settings",
|
||||
transportClassification: "unexpected",
|
||||
errorClass: "Error",
|
||||
errorMessage: "Checksum mismatch",
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("should not fail sync when getSettingsForSync throws", async () => {
|
||||
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
|
||||
const runtimeHarness = createRuntimeLoggerHarness();
|
||||
const appWithLogger = createServer(new MockStore() as unknown as TaskStore, {
|
||||
runtimeLogger: runtimeHarness.logger,
|
||||
});
|
||||
|
||||
mockGetSettings.mockRejectedValue(new Error("Settings unavailable"));
|
||||
|
||||
const response = await request(
|
||||
appWithLogger,
|
||||
"POST",
|
||||
"/api/mesh/sync",
|
||||
JSON.stringify({
|
||||
senderNodeId: "node_remote",
|
||||
senderNodeUrl: "https://remote.example.com",
|
||||
knownPeers: [],
|
||||
timestamp: "2026-04-01T12:00:00.000Z",
|
||||
settings: remotePayload,
|
||||
}),
|
||||
{ "Content-Type": "application/json" }
|
||||
);
|
||||
|
||||
// Sync should still succeed even if getting settings failed
|
||||
expect(response.status).toBe(200);
|
||||
expect(mockMergePeers).toHaveBeenCalled();
|
||||
expect((response.body as any).knownPeers).toBeDefined();
|
||||
expect((response.body as any).settings).toBeUndefined();
|
||||
expect(runtimeHarness.entries).toContainEqual(
|
||||
expect.objectContaining({
|
||||
level: "error",
|
||||
scope: "test:routes:remote-route:mesh-sync",
|
||||
message: "Settings sync operation failed",
|
||||
context: expect.objectContaining({
|
||||
nodeId: "node_remote",
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "settings-sync",
|
||||
transportClassification: "unexpected",
|
||||
errorClass: "Error",
|
||||
errorMessage: "Settings unavailable",
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ── FN-7647 Symptom Verification: auth-material shared-state sync ─────────────
|
||||
@@ -861,7 +747,7 @@ describe("/api/mesh/task-ids routes", () => {
|
||||
mockCommitDistributedTaskIdReservation.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, committedClusterTaskCount: 1, committedAt: "2030-01-01T00:00:00.000Z" });
|
||||
mockAbortDistributedTaskIdReservation.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, committedClusterTaskCount: 0, abortedAt: "2030-01-01T00:00:00.000Z" });
|
||||
mockGetDistributedTaskIdState.mockResolvedValue({ nextSequence: 2, committedClusterTaskCount: 1, activeReservationCount: 0, burnedReservationCount: 0, lastCommittedTaskId: "FN-001" });
|
||||
app = createServer(new MockStore() as unknown as TaskStore);
|
||||
app = createMeshTestServer(new MockStore() as unknown as TaskStore);
|
||||
});
|
||||
|
||||
it("reserves distributed task ids locally", async () => {
|
||||
@@ -894,12 +780,42 @@ describe("/api/mesh/task-ids routes", () => {
|
||||
expect(response.status).toBe(401);
|
||||
});
|
||||
|
||||
it("returns 503 when coordinator is unreachable for writes", async () => {
|
||||
mockGetNode.mockResolvedValue(makeNodeConfig({ id: "node_remote_1", url: "https://remote.example.com", apiKey: "secret" }));
|
||||
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("network down")));
|
||||
const response = await request(app, "POST", "/api/mesh/task-ids/commit", JSON.stringify({ reservationId: "res-1", nodeId: "node-a", coordinatorNodeId: "node_remote_1" }), { "Content-Type": "application/json" });
|
||||
expect(response.status).toBe(503);
|
||||
vi.unstubAllGlobals();
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Remote coordinator hops are retired on all three mutating allocator routes.
|
||||
Assert the invariant across reserve/commit/abort, not only commit.
|
||||
*/
|
||||
it.each([
|
||||
{
|
||||
name: "reserve",
|
||||
method: "POST" as const,
|
||||
path: "/api/mesh/task-ids/reserve",
|
||||
body: { prefix: "FN", nodeId: "node-a", coordinatorNodeId: "node_remote_1" },
|
||||
mock: mockReserveDistributedTaskId,
|
||||
expectedArgs: { prefix: "FN", nodeId: "node-a", ttlMs: undefined },
|
||||
},
|
||||
{
|
||||
name: "commit",
|
||||
method: "POST" as const,
|
||||
path: "/api/mesh/task-ids/commit",
|
||||
body: { reservationId: "res-1", nodeId: "node-a", coordinatorNodeId: "node_remote_1" },
|
||||
mock: mockCommitDistributedTaskIdReservation,
|
||||
expectedArgs: { reservationId: "res-1", nodeId: "node-a" },
|
||||
},
|
||||
{
|
||||
name: "abort",
|
||||
method: "POST" as const,
|
||||
path: "/api/mesh/task-ids/abort",
|
||||
body: { reservationId: "res-1", nodeId: "node-a", reason: "abort", coordinatorNodeId: "node_remote_1" },
|
||||
mock: mockAbortDistributedTaskIdReservation,
|
||||
expectedArgs: { reservationId: "res-1", nodeId: "node-a", reason: "abort" },
|
||||
},
|
||||
])("ignores coordinatorNodeId on $name and allocates locally", async ({ method, path, body, mock, expectedArgs }) => {
|
||||
mockGetNode.mockClear();
|
||||
const response = await request(app, method, path, JSON.stringify(body), { "Content-Type": "application/json" });
|
||||
expect(response.status).toBe(200);
|
||||
expect(mock).toHaveBeenCalledWith(expectedArgs);
|
||||
expect(mockGetNode).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -913,7 +829,7 @@ describe("GET /api/mesh/state", () => {
|
||||
mockInit.mockResolvedValue(undefined);
|
||||
mockClose.mockResolvedValue(undefined);
|
||||
mockGetNode.mockResolvedValue(undefined);
|
||||
app = createServer(new MockStore() as unknown as TaskStore);
|
||||
app = createMeshTestServer(new MockStore() as unknown as TaskStore);
|
||||
});
|
||||
|
||||
it("returns local-only mesh snapshot when includeRemote=false", async () => {
|
||||
@@ -951,7 +867,7 @@ describe("GET /api/mesh/state", () => {
|
||||
};
|
||||
|
||||
const store = new MockStore();
|
||||
const sharedApp = createServer(store as unknown as TaskStore, { centralCore: sharedCentral as never });
|
||||
const sharedApp = createMeshTestServer(store as unknown as TaskStore, { centralCore: sharedCentral as never });
|
||||
const response = await request(sharedApp, "GET", "/api/mesh/state?includeRemote=false");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
@@ -1060,7 +976,7 @@ describe("PostgreSQL backend mode: task mesh replication disabled", () => {
|
||||
mockUpdateNode.mockResolvedValue({ id: "node_remote", status: "online" });
|
||||
mockReserveDistributedTaskId.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, expiresAt: "2030-01-01T00:00:00.000Z", committedClusterTaskCount: 0 });
|
||||
store = new BackendModeMockStore();
|
||||
app = createServer(store as unknown as TaskStore);
|
||||
app = createMeshTestServer(store as unknown as TaskStore);
|
||||
});
|
||||
|
||||
it("POST /api/mesh/tasks/create no longer exists (route removed — replication is the database)", async () => {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { validateSnapshotEnvelope } from "@fusion/core";
|
||||
import { createFusionAuthStorage } from "@fusion/engine";
|
||||
import { ApiError, badRequest } from "../api-error.js";
|
||||
import type { ApiRouteRegistrar } from "./types.js";
|
||||
import { fetchFromRemoteNode } from "./register-settings-sync-helpers.js";
|
||||
|
||||
export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
const { router, store, options, emitRemoteRouteDiagnostic, rethrowAsApiError } = ctx;
|
||||
@@ -23,41 +23,6 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
}
|
||||
};
|
||||
|
||||
const resolveAllocator = async (coordinatorNodeId?: string) => {
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-12:
|
||||
Task mesh replication is REMOVED on the PostgreSQL backend: every node
|
||||
connects to the same shared database, so the shared
|
||||
`distributed_task_id_state` rows ARE the coordinator. Never forward a
|
||||
reservation to a remote coordinator node — the local allocator commits
|
||||
atomically against the shared rows, and a remote hop only adds a failure
|
||||
mode (and double-reservation risk against the same table).
|
||||
*/
|
||||
if (store.backendMode) {
|
||||
return { mode: "local" as const };
|
||||
}
|
||||
return withCentralCore(async (central) => {
|
||||
if (!coordinatorNodeId) {
|
||||
return { mode: "local" as const };
|
||||
}
|
||||
const coordinator = await central.getNode(coordinatorNodeId);
|
||||
if (coordinator?.type === "local") {
|
||||
return { mode: "local" as const };
|
||||
}
|
||||
if (!coordinator) {
|
||||
throw new ApiError(503, "Allocator coordinator is unavailable");
|
||||
}
|
||||
return { mode: "remote" as const, coordinator };
|
||||
});
|
||||
};
|
||||
|
||||
const mapCoordinatorWriteError = (err: unknown): never => {
|
||||
if (err instanceof ApiError && [502, 504].includes(err.statusCode)) {
|
||||
throw new ApiError(503, "Allocator coordinator is unavailable");
|
||||
}
|
||||
throw err;
|
||||
};
|
||||
|
||||
const requireMeshAuth = async (
|
||||
req: { headers: { authorization?: string } },
|
||||
res: { status: (code: number) => { json: (payload: unknown) => void } },
|
||||
@@ -200,29 +165,18 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
*/
|
||||
router.post("/mesh/task-ids/reserve", async (req, res) => {
|
||||
try {
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Always allocate against shared distributed_task_id_* rows. coordinatorNodeId is ignored.
|
||||
*/
|
||||
const prefix = String(req.body?.prefix ?? "").trim();
|
||||
const nodeId = String(req.body?.nodeId ?? "").trim();
|
||||
const ttlMs = req.body?.ttlMs;
|
||||
const coordinatorNodeId = typeof req.body?.coordinatorNodeId === "string" ? req.body.coordinatorNodeId : undefined;
|
||||
const senderNodeId = typeof req.body?.senderNodeId === "string" ? req.body.senderNodeId : undefined;
|
||||
if (!prefix) throw badRequest("prefix is required");
|
||||
if (!nodeId) throw badRequest("nodeId is required");
|
||||
if (!(await requireMeshAuth(req, res, senderNodeId))) return;
|
||||
|
||||
const target = await resolveAllocator(coordinatorNodeId);
|
||||
if (target.mode === "remote") {
|
||||
try {
|
||||
const remote = await fetchFromRemoteNode(target.coordinator, "/api/mesh/task-ids/reserve", {
|
||||
method: "POST",
|
||||
body: { prefix, nodeId, ttlMs },
|
||||
});
|
||||
res.json(remote);
|
||||
return;
|
||||
} catch (err) {
|
||||
mapCoordinatorWriteError(err);
|
||||
}
|
||||
}
|
||||
|
||||
const result = await store.getDistributedTaskIdAllocator().reserveDistributedTaskId({ prefix, nodeId, ttlMs });
|
||||
res.json(result);
|
||||
} catch (err: unknown) {
|
||||
@@ -235,26 +189,11 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
try {
|
||||
const reservationId = String(req.body?.reservationId ?? "").trim();
|
||||
const nodeId = String(req.body?.nodeId ?? "").trim();
|
||||
const coordinatorNodeId = typeof req.body?.coordinatorNodeId === "string" ? req.body.coordinatorNodeId : undefined;
|
||||
const senderNodeId = typeof req.body?.senderNodeId === "string" ? req.body.senderNodeId : undefined;
|
||||
if (!reservationId) throw badRequest("reservationId is required");
|
||||
if (!nodeId) throw badRequest("nodeId is required");
|
||||
if (!(await requireMeshAuth(req, res, senderNodeId))) return;
|
||||
|
||||
const target = await resolveAllocator(coordinatorNodeId);
|
||||
if (target.mode === "remote") {
|
||||
try {
|
||||
const remote = await fetchFromRemoteNode(target.coordinator, "/api/mesh/task-ids/commit", {
|
||||
method: "POST",
|
||||
body: { reservationId, nodeId },
|
||||
});
|
||||
res.json(remote);
|
||||
return;
|
||||
} catch (err) {
|
||||
mapCoordinatorWriteError(err);
|
||||
}
|
||||
}
|
||||
|
||||
const result = await store.getDistributedTaskIdAllocator().commitDistributedTaskIdReservation({ reservationId, nodeId });
|
||||
res.json(result);
|
||||
} catch (err: unknown) {
|
||||
@@ -271,7 +210,6 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
const reservationId = String(req.body?.reservationId ?? "").trim();
|
||||
const nodeId = String(req.body?.nodeId ?? "").trim();
|
||||
const reason = req.body?.reason;
|
||||
const coordinatorNodeId = typeof req.body?.coordinatorNodeId === "string" ? req.body.coordinatorNodeId : undefined;
|
||||
const senderNodeId = typeof req.body?.senderNodeId === "string" ? req.body.senderNodeId : undefined;
|
||||
if (!reservationId) throw badRequest("reservationId is required");
|
||||
if (!nodeId) throw badRequest("nodeId is required");
|
||||
@@ -280,20 +218,6 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
}
|
||||
if (!(await requireMeshAuth(req, res, senderNodeId))) return;
|
||||
|
||||
const target = await resolveAllocator(coordinatorNodeId);
|
||||
if (target.mode === "remote") {
|
||||
try {
|
||||
const remote = await fetchFromRemoteNode(target.coordinator, "/api/mesh/task-ids/abort", {
|
||||
method: "POST",
|
||||
body: { reservationId, nodeId, reason },
|
||||
});
|
||||
res.json(remote);
|
||||
return;
|
||||
} catch (err) {
|
||||
mapCoordinatorWriteError(err);
|
||||
}
|
||||
}
|
||||
|
||||
const result = await store.getDistributedTaskIdAllocator().abortDistributedTaskIdReservation({ reservationId, nodeId, reason });
|
||||
res.json(result);
|
||||
} catch (err: unknown) {
|
||||
@@ -384,22 +308,17 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
// Get local node info
|
||||
const localPeer = await central.getLocalPeerInfo();
|
||||
|
||||
// ── Settings sync: handle incoming settings and prepare response ──
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-10:
|
||||
Node settings sync is REMOVED on the PostgreSQL backend: nodes connect to
|
||||
the same shared PostgreSQL database, so mesh-level settings replication is
|
||||
redundant and can only introduce churn/clobber against the shared rows.
|
||||
Inbound settings payloads are ignored (with a diagnostic) and no settings
|
||||
are included in the response. The legacy SQLite topology (one DB file per
|
||||
node) keeps the sync path unchanged.
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Settings and projectSettings mesh sync are retired: shared Postgres is the
|
||||
settings SoT. Inbound settings/projectSettings are ignored. Only
|
||||
authMaterial (per-machine auth.json) is applied/offered over mesh HTTP.
|
||||
*/
|
||||
let responseSettings: import("@fusion/core").SettingsSyncPayload | undefined;
|
||||
const remoteSettings = store.backendMode ? undefined : req.body?.settings;
|
||||
if (store.backendMode && req.body?.settings) {
|
||||
if (req.body?.settings) {
|
||||
emitRemoteRouteDiagnostic({
|
||||
route: "mesh-sync",
|
||||
message: "Ignored inbound settings payload — settings sync is disabled on the PostgreSQL backend (nodes share the database)",
|
||||
message: "Ignored inbound settings payload — settings live in shared PostgreSQL",
|
||||
nodeId: senderNodeId,
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "settings-sync",
|
||||
@@ -407,82 +326,16 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
});
|
||||
}
|
||||
|
||||
if (remoteSettings) {
|
||||
try {
|
||||
// Get local settings from the dashboard's GlobalSettingsStore
|
||||
const localGlobal = await store.getGlobalSettingsStore().getSettings();
|
||||
const localPayload = await central.getSettingsForSync(localGlobal);
|
||||
const localChecksum = localPayload.checksum;
|
||||
|
||||
// Apply remote settings if checksum differs (remote is newer/different)
|
||||
if (remoteSettings.checksum !== localChecksum) {
|
||||
const applyResult = await central.applyRemoteSettings(remoteSettings);
|
||||
|
||||
if (applyResult.success) {
|
||||
emitRemoteRouteDiagnostic({
|
||||
route: "mesh-sync",
|
||||
message: "Applied remote settings payload",
|
||||
nodeId: senderNodeId,
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "apply-remote-settings",
|
||||
level: "info",
|
||||
context: {
|
||||
globalCount: applyResult.globalCount,
|
||||
projectCount: applyResult.projectCount,
|
||||
authCount: applyResult.authCount,
|
||||
},
|
||||
});
|
||||
} else {
|
||||
emitRemoteRouteDiagnostic({
|
||||
route: "mesh-sync",
|
||||
message: "Failed to apply remote settings payload",
|
||||
nodeId: senderNodeId,
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "apply-remote-settings",
|
||||
level: "warn",
|
||||
error: new Error(applyResult.error ?? "Unknown applyRemoteSettings failure"),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Always respond with our settings if sender included theirs
|
||||
responseSettings = localPayload;
|
||||
} catch (err) {
|
||||
// Log but don't fail the sync - peers are more important
|
||||
emitRemoteRouteDiagnostic({
|
||||
route: "mesh-sync",
|
||||
message: "Settings sync operation failed",
|
||||
nodeId: senderNodeId,
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "settings-sync",
|
||||
error: err,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ── Shared state sync (settings/auth only) ──
|
||||
const { validateSnapshotEnvelope } = await import("@fusion/core");
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-12:
|
||||
Task/state mesh replication is REMOVED: the task-metadata,
|
||||
mission-hierarchy, agents, agent-runs, activity-log, and run-audit
|
||||
shared-state domains (and the store snapshot machinery behind them) are
|
||||
gone — all replication is handled at the PostgreSQL level (nodes share
|
||||
the database). What remains of sharedState is the settings-adjacent
|
||||
pair: projectSettings (legacy sqlite settings sync only; ignored on the
|
||||
PostgreSQL backend like the rest of settings sync) and authMaterial
|
||||
(auth.json is per-machine file state — the one domain not in the
|
||||
database, kept on both backends).
|
||||
*/
|
||||
// ── Shared state: auth material only ──
|
||||
const rawSharedState = req.body?.sharedState;
|
||||
let sharedState = rawSharedState;
|
||||
if (rawSharedState && typeof rawSharedState === "object") {
|
||||
const allowedDomains = store.backendMode ? ["authMaterial"] : ["projectSettings", "authMaterial"];
|
||||
const allowedDomains = ["authMaterial"];
|
||||
const ignoredDomains = Object.keys(rawSharedState).filter((domain) => !allowedDomains.includes(domain));
|
||||
if (ignoredDomains.length > 0) {
|
||||
emitRemoteRouteDiagnostic({
|
||||
route: "mesh-sync",
|
||||
message: `Ignored inbound shared-state domains [${ignoredDomains.join(", ")}] — task/state mesh replication is removed (replication is handled at the PostgreSQL level)`,
|
||||
message: `Ignored inbound shared-state domains [${ignoredDomains.join(", ")}] — only authMaterial is exchanged over mesh (durable state is shared PostgreSQL)`,
|
||||
nodeId: senderNodeId,
|
||||
upstreamPath: "/api/mesh/sync",
|
||||
operationStage: "shared-state-sync",
|
||||
@@ -512,15 +365,6 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
}
|
||||
};
|
||||
|
||||
await applyDomain("project-settings", async () => {
|
||||
if (!sharedState.projectSettings) return;
|
||||
validateSnapshotEnvelope(sharedState.projectSettings);
|
||||
const result = await central.applyProjectSettingsSnapshot(sharedState.projectSettings as Parameters<typeof central.applyProjectSettingsSnapshot>[0]);
|
||||
if (!result.success) {
|
||||
throw new Error(result.error ?? "applyProjectSettingsSnapshot failed");
|
||||
}
|
||||
});
|
||||
|
||||
await applyDomain("auth-material", async () => {
|
||||
if (!sharedState.authMaterial) return;
|
||||
validateSnapshotEnvelope(sharedState.authMaterial);
|
||||
@@ -554,7 +398,7 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
|
||||
}
|
||||
|
||||
// Build shared-state response from fresh local snapshots per request.
|
||||
// Build shared-state response: authMaterial only (file-local credentials).
|
||||
const responseSharedState: Record<string, unknown> = {};
|
||||
const collectSnapshot = async (domain: string, fn: () => Promise<unknown>): Promise<void> => {
|
||||
try {
|
||||
@@ -584,18 +428,6 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
}
|
||||
};
|
||||
|
||||
/*
|
||||
FNXC:PostgresCutover 2026-07-12:
|
||||
Outbound shared-state mirrors the inbound surface: the database-backed
|
||||
domain snapshots are removed; projectSettings is offered only on the
|
||||
legacy sqlite topology; authMaterial is offered on both backends.
|
||||
*/
|
||||
if (!store.backendMode) {
|
||||
await collectSnapshot("projectSettings", async () => {
|
||||
const localGlobal = await store.getGlobalSettingsStore().getSettings();
|
||||
return central.getProjectSettingsSnapshot(localGlobal);
|
||||
});
|
||||
}
|
||||
await collectSnapshot("authMaterial", async () => {
|
||||
const authPathsModule = await import("./register-settings-sync-helpers.js");
|
||||
const allProviders = await authPathsModule.readStoredAuthProvidersFromDisk();
|
||||
@@ -604,7 +436,7 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
|
||||
await central.close();
|
||||
|
||||
// Return sync response
|
||||
// Return sync response (membership + optional auth material only)
|
||||
const response: Record<string, unknown> = {
|
||||
senderNodeId: localPeer.nodeId,
|
||||
senderNodeUrl: localPeer.nodeUrl,
|
||||
@@ -613,10 +445,6 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
// Include settings in response if sender sent settings
|
||||
if (responseSettings) {
|
||||
response.settings = responseSettings;
|
||||
}
|
||||
if (Object.keys(responseSharedState).length > 0) {
|
||||
response.sharedState = responseSharedState;
|
||||
}
|
||||
|
||||
@@ -51,6 +51,7 @@ describe("PeerExchangeService", () => {
|
||||
let mockApplyRemoteSettings: ReturnType<typeof vi.fn>;
|
||||
let mockGetProjectSettingsSnapshot: ReturnType<typeof vi.fn>;
|
||||
let mockGetAuthMaterialSnapshot: ReturnType<typeof vi.fn>;
|
||||
let mockApplyAuthMaterialSnapshot: ReturnType<typeof vi.fn>;
|
||||
let mockApplyProjectSettingsSnapshot: ReturnType<typeof vi.fn>;
|
||||
let mockEnqueueMeshWrite: ReturnType<typeof vi.fn>;
|
||||
let mockListPendingMeshWrites: ReturnType<typeof vi.fn>;
|
||||
@@ -72,6 +73,7 @@ describe("PeerExchangeService", () => {
|
||||
mockApplyRemoteSettings = vi.fn();
|
||||
mockGetProjectSettingsSnapshot = vi.fn();
|
||||
mockGetAuthMaterialSnapshot = vi.fn();
|
||||
mockApplyAuthMaterialSnapshot = vi.fn();
|
||||
mockApplyProjectSettingsSnapshot = vi.fn();
|
||||
mockEnqueueMeshWrite = vi.fn();
|
||||
mockListPendingMeshWrites = vi.fn();
|
||||
@@ -81,6 +83,8 @@ describe("PeerExchangeService", () => {
|
||||
mockGetNode = vi.fn();
|
||||
|
||||
mockCentralCore = {
|
||||
// Default: not shared-Postgres backendMode so legacy settings-sync tests remain valid.
|
||||
backendMode: false,
|
||||
listNodes: mockListNodes,
|
||||
getAllKnownPeerInfo: mockGetAllKnownPeerInfo,
|
||||
mergePeers: mockMergePeers,
|
||||
@@ -89,6 +93,7 @@ describe("PeerExchangeService", () => {
|
||||
applyRemoteSettings: mockApplyRemoteSettings,
|
||||
getProjectSettingsSnapshot: mockGetProjectSettingsSnapshot,
|
||||
getAuthMaterialSnapshot: mockGetAuthMaterialSnapshot,
|
||||
applyAuthMaterialSnapshot: mockApplyAuthMaterialSnapshot,
|
||||
applyProjectSettingsSnapshot: mockApplyProjectSettingsSnapshot,
|
||||
enqueueMeshWrite: mockEnqueueMeshWrite,
|
||||
listPendingMeshWrites: mockListPendingMeshWrites,
|
||||
@@ -408,6 +413,104 @@ describe("PeerExchangeService", () => {
|
||||
});
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Under shared Postgres, settings gossip is forced off, queue rows are
|
||||
topology/auth-scoped, and non-topology pending writes are failed rather than
|
||||
replayed as multi-leader task/settings payloads.
|
||||
*/
|
||||
describe("shared PostgreSQL backendMode", () => {
|
||||
beforeEach(() => {
|
||||
Object.defineProperty(mockCentralCore, "backendMode", { value: true, configurable: true });
|
||||
});
|
||||
|
||||
it("force-disables settingsSyncEnabled even when the option is true", async () => {
|
||||
mockGetSettingsForSync.mockResolvedValue(makeSettingsPayload());
|
||||
const service = new PeerExchangeService(mockCentralCore, { settingsSyncEnabled: true });
|
||||
setupSuccessfulSync();
|
||||
await service.syncWithNode(makeNode());
|
||||
expect(mockGetSettingsForSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("enqueues retryable peer failures as mesh.topology / topology-sync", async () => {
|
||||
const node = makeNode();
|
||||
mockListNodes.mockResolvedValue([makeNode({ id: "node_local", type: "local", status: "online" })]);
|
||||
mockGetAllKnownPeerInfo.mockResolvedValue([]);
|
||||
mockReportMeshState.mockResolvedValue({});
|
||||
mockFetch.mockResolvedValue({ ok: false, status: 503, statusText: "Service Unavailable" });
|
||||
mockEnqueueMeshWrite.mockResolvedValue({ id: "mq-topo-1" });
|
||||
|
||||
const service = new PeerExchangeService(mockCentralCore);
|
||||
const result = await service.syncWithNode(node);
|
||||
|
||||
expect(result.queuedWriteId).toBe("mq-topo-1");
|
||||
expect(mockEnqueueMeshWrite).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
scope: "mesh.topology",
|
||||
entityType: "topology-sync",
|
||||
}),
|
||||
);
|
||||
const payload = mockEnqueueMeshWrite.mock.calls[0][0].payload.request;
|
||||
expect(payload.settings).toBeUndefined();
|
||||
expect(payload.sharedState?.projectSettings).toBeUndefined();
|
||||
});
|
||||
|
||||
it("replays topology scopes and fails retired task/settings queue rows", async () => {
|
||||
const node = makeNode();
|
||||
setupSuccessfulSync(node);
|
||||
mockListPendingMeshWrites.mockResolvedValue([
|
||||
{
|
||||
id: "mq-topo",
|
||||
originNodeId: "node_local",
|
||||
targetNodeId: node.id,
|
||||
projectId: null,
|
||||
scope: "mesh.topology",
|
||||
entityType: "topology-sync",
|
||||
entityId: node.id,
|
||||
operation: "sync",
|
||||
payload: { request: { senderNodeId: "node_local", knownPeers: [], senderNodeUrl: "", timestamp: "2026-04-01T12:00:00.000Z" } },
|
||||
intentVersion: "1.0",
|
||||
status: "pending",
|
||||
attemptCount: 0,
|
||||
createdAt: "2026-04-01T12:00:00.000Z",
|
||||
updatedAt: "2026-04-01T12:00:00.000Z",
|
||||
},
|
||||
{
|
||||
id: "mq-task",
|
||||
originNodeId: "node_local",
|
||||
targetNodeId: node.id,
|
||||
projectId: "proj-1",
|
||||
scope: "task.strong",
|
||||
entityType: "task-create",
|
||||
entityId: "FN-1",
|
||||
operation: "create",
|
||||
payload: { request: {} },
|
||||
intentVersion: "1.0",
|
||||
status: "pending",
|
||||
attemptCount: 0,
|
||||
createdAt: "2026-04-01T12:00:00.000Z",
|
||||
updatedAt: "2026-04-01T12:00:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
const service = new PeerExchangeService(mockCentralCore);
|
||||
const result = await service.syncWithNode(node);
|
||||
|
||||
expect(result.replaySummary).toEqual({
|
||||
replayed: 1,
|
||||
applied: 1,
|
||||
failed: 0,
|
||||
queuedWriteIds: ["mq-topo"],
|
||||
});
|
||||
expect(mockMarkMeshWriteFailed).toHaveBeenCalledWith(
|
||||
"mq-task",
|
||||
expect.objectContaining({ lastError: expect.stringContaining("retired") }),
|
||||
);
|
||||
expect(mockMarkMeshWriteReplayStarted).toHaveBeenCalledWith("mq-topo");
|
||||
expect(mockMarkMeshWriteApplied).toHaveBeenCalledWith("mq-topo", {});
|
||||
});
|
||||
});
|
||||
|
||||
describe("triggerSync()", () => {
|
||||
it("should trigger sync when called", async () => {
|
||||
mockListNodes.mockResolvedValue([
|
||||
@@ -460,6 +563,60 @@ describe("PeerExchangeService", () => {
|
||||
expect(result.settingsApplied).toBeUndefined();
|
||||
expect(result.settingsVersion).toBeUndefined();
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-15-00:15:
|
||||
Auth material must still travel when settings gossip is off (Postgres path).
|
||||
*/
|
||||
it("includes and applies authMaterial when settingsSyncAuth is enabled without settings gossip", async () => {
|
||||
Object.defineProperty(mockCentralCore, "backendMode", { value: true, configurable: true });
|
||||
mockGetAuthMaterialSnapshot.mockReturnValue({
|
||||
version: 1,
|
||||
exportedAt: "2026-04-01T00:00:00.000Z",
|
||||
checksum: "auth-checksum",
|
||||
payload: { providerAuth: { anthropic: { type: "api_key", key: "sk-ant" } } },
|
||||
});
|
||||
mockApplyAuthMaterialSnapshot.mockReturnValue({
|
||||
success: true,
|
||||
authCount: 1,
|
||||
providerAuth: { anthropic: { type: "api_key", key: "sk-ant" } },
|
||||
});
|
||||
const node = makeNode();
|
||||
setupSuccessfulSync(node);
|
||||
mockFetch.mockResolvedValue({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
senderNodeId: node.id,
|
||||
senderNodeUrl: node.url,
|
||||
knownPeers: [],
|
||||
newPeers: [],
|
||||
timestamp: "2026-04-01T12:00:00.000Z",
|
||||
sharedState: {
|
||||
authMaterial: {
|
||||
version: 1,
|
||||
exportedAt: "2026-04-01T00:00:00.000Z",
|
||||
checksum: "remote-auth",
|
||||
payload: { providerAuth: { anthropic: { type: "api_key", key: "sk-remote" } } },
|
||||
},
|
||||
},
|
||||
}),
|
||||
});
|
||||
|
||||
const service = new PeerExchangeService(mockCentralCore, {
|
||||
settingsSyncEnabled: true, // forced off by backendMode
|
||||
settingsSyncAuth: true,
|
||||
providerAuth: { anthropic: { type: "api_key", key: "sk-ant" } },
|
||||
});
|
||||
await service.syncWithNode(node);
|
||||
|
||||
expect(mockGetSettingsForSync).not.toHaveBeenCalled();
|
||||
const body = JSON.parse(mockFetch.mock.calls[0][1].body);
|
||||
expect(body.settings).toBeUndefined();
|
||||
expect(body.sharedState?.authMaterial).toBeDefined();
|
||||
expect(body.sharedState?.projectSettings).toBeUndefined();
|
||||
expect(mockApplyAuthMaterialSnapshot).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("settings sync - when enabled", () => {
|
||||
|
||||
@@ -51,8 +51,9 @@ export interface SyncResult {
|
||||
/**
|
||||
* Background service that implements the peer gossip protocol.
|
||||
*
|
||||
* Periodically exchanges peer information with connected remote nodes
|
||||
* to keep the mesh state up-to-date across all nodes.
|
||||
* Periodically exchanges peer membership with connected remote nodes.
|
||||
* Under shared PostgreSQL, this is membership (+ optional auth material) only —
|
||||
* not a task/settings multi-leader replication engine.
|
||||
*/
|
||||
export class PeerExchangeService {
|
||||
private centralCore: CentralCore;
|
||||
@@ -60,7 +61,7 @@ export class PeerExchangeService {
|
||||
private interval: ReturnType<typeof setInterval> | null = null;
|
||||
private activeSync: Promise<void> | null = null;
|
||||
private running = false;
|
||||
/** Whether settings sync is enabled. Default: false. */
|
||||
/** Whether settings sync is enabled. Default: false. Forced false under Postgres. */
|
||||
private settingsSyncEnabled: boolean;
|
||||
/** Minimum interval between settings syncs with the same node in ms. Default: 5 minutes. */
|
||||
private settingsSyncThrottleMs: number;
|
||||
@@ -92,6 +93,10 @@ export class PeerExchangeService {
|
||||
the same shared PostgreSQL database, so gossip-level settings replication is
|
||||
redundant. Force-disable regardless of the caller's option when the central
|
||||
core runs in backend mode.
|
||||
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Shared Postgres is the durable SoT. Mesh HTTP stays for membership (+ optional
|
||||
auth.json sync). meshWriteQueue is topology/auth-only — never a task multi-master log.
|
||||
*/
|
||||
this.settingsSyncEnabled = centralCore.backendMode ? false : (options.settingsSyncEnabled ?? false);
|
||||
this.settingsSyncThrottleMs = options.settingsSyncThrottleMs ?? 300_000; // 5 minutes default
|
||||
@@ -100,6 +105,11 @@ export class PeerExchangeService {
|
||||
this.providerAuth = options.providerAuth;
|
||||
}
|
||||
|
||||
/** True when durable state is shared Postgres (no multi-leader task/settings mesh). */
|
||||
private get sharedPostgresMode(): boolean {
|
||||
return this.centralCore.backendMode === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the global settings used for settings sync.
|
||||
* Call this when global settings change to ensure fresh data is included in the next sync.
|
||||
@@ -289,7 +299,13 @@ export class PeerExchangeService {
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
|
||||
// ── Settings sync: decide whether to include settings in request ──
|
||||
// ── Settings sync (legacy) + auth material (independent of settings gossip) ──
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-15-00:15:
|
||||
Auth material lives in per-machine auth.json and must still sync when
|
||||
settings gossip is force-disabled under shared Postgres. Attach auth-only
|
||||
sharedState whenever settingsSyncAuth is on, without requiring settingsSyncEnabled.
|
||||
*/
|
||||
let shouldIncludeSettings = false;
|
||||
|
||||
if (this.settingsSyncEnabled) {
|
||||
@@ -333,7 +349,6 @@ export class PeerExchangeService {
|
||||
|
||||
if (shouldIncludeSettings) {
|
||||
request.settings = this.cachedSettingsPayload;
|
||||
request.sharedState = await this.getSharedStateSettingsBundle();
|
||||
}
|
||||
} catch (err) {
|
||||
// Log error but continue with peer sync
|
||||
@@ -342,6 +357,18 @@ export class PeerExchangeService {
|
||||
}
|
||||
}
|
||||
|
||||
if (shouldIncludeSettings || this.settingsSyncAuth) {
|
||||
try {
|
||||
const sharedState = await this.getSharedStateSettingsBundle();
|
||||
if (sharedState) {
|
||||
request.sharedState = sharedState;
|
||||
}
|
||||
} catch (err) {
|
||||
const error = err instanceof Error ? err : String(err);
|
||||
peerExchangeLog.warn(`Failed to build shared-state bundle for ${node.name}: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Build headers
|
||||
const headers: Record<string, string> = {
|
||||
"Content-Type": "application/json",
|
||||
@@ -386,7 +413,7 @@ export class PeerExchangeService {
|
||||
// This ensures we get updates for existing peers too
|
||||
const mergeResult = await this.centralCore.mergePeers(peerResponse.knownPeers);
|
||||
|
||||
// ── Process remote settings if included in response ──
|
||||
// ── Process remote settings (legacy) and/or auth sharedState ──
|
||||
if (this.settingsSyncEnabled && (peerResponse.sharedState || peerResponse.settings)) {
|
||||
const remoteChecksum =
|
||||
peerResponse.sharedState?.projectSettings?.checksum ??
|
||||
@@ -427,6 +454,23 @@ export class PeerExchangeService {
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
}
|
||||
} else if (this.settingsSyncAuth && peerResponse.sharedState?.authMaterial) {
|
||||
// Auth-only path when settings gossip is disabled (shared Postgres default).
|
||||
try {
|
||||
const applyResult = await this.applyRemoteSharedState(peerResponse.sharedState, undefined);
|
||||
if (applyResult.success && applyResult.authCount > 0) {
|
||||
settingsApplied = true;
|
||||
this.cachedSharedStatePayload = null;
|
||||
peerExchangeLog.log(
|
||||
`Applied remote auth material from ${node.name} (auth providers: ${applyResult.authCount})`,
|
||||
);
|
||||
} else if (!applyResult.success) {
|
||||
peerExchangeLog.warn(`Failed to apply remote auth material from ${node.name}: ${applyResult.error}`);
|
||||
}
|
||||
} catch (err) {
|
||||
const error = err instanceof Error ? err.message : String(err);
|
||||
peerExchangeLog.warn(`Auth material sync error with ${node.name}: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
peerExchangeLog.log(
|
||||
@@ -470,42 +514,94 @@ export class PeerExchangeService {
|
||||
}
|
||||
|
||||
const pending = await this.centralCore.listPendingMeshWrites({ targetNodeId, status: "pending" });
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Under shared Postgres only topology/auth mesh queue scopes are replayable.
|
||||
Legacy task/settings queue rows (if any) are marked failed instead of
|
||||
reintroducing multi-leader task writes over HTTP.
|
||||
*/
|
||||
const eligible = this.sharedPostgresMode
|
||||
? pending.filter((entry) => this.isTopologyOrAuthMeshScope(entry.scope, entry.entityType))
|
||||
: pending;
|
||||
if (this.sharedPostgresMode) {
|
||||
for (const entry of pending) {
|
||||
if (this.isTopologyOrAuthMeshScope(entry.scope, entry.entityType)) continue;
|
||||
await this.centralCore.markMeshWriteFailed(entry.id, {
|
||||
lastError: "Skipped: task/settings mesh write queues are retired under shared PostgreSQL",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let applied = 0;
|
||||
let failed = 0;
|
||||
|
||||
for (const entry of pending) {
|
||||
for (const entry of eligible) {
|
||||
await this.centralCore.markMeshWriteReplayStarted(entry.id);
|
||||
try {
|
||||
const headers: Record<string, string> = { "Content-Type": "application/json" };
|
||||
if (node.apiKey) headers.Authorization = `Bearer ${node.apiKey}`;
|
||||
const response = await fetch(`${node.url}/api/mesh/sync`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(entry.payload?.request ?? {}),
|
||||
});
|
||||
if (!response.ok) {
|
||||
await this.centralCore.markMeshWriteFailed(entry.id, { lastError: `HTTP ${response.status}: ${response.statusText}` });
|
||||
failed += 1;
|
||||
continue;
|
||||
// Strip legacy settings payloads before replay under shared Postgres.
|
||||
const rawRequest = (entry.payload?.request ?? {}) as PeerSyncRequest;
|
||||
const body: PeerSyncRequest = this.sharedPostgresMode
|
||||
? {
|
||||
senderNodeId: rawRequest.senderNodeId,
|
||||
senderNodeUrl: rawRequest.senderNodeUrl,
|
||||
knownPeers: rawRequest.knownPeers ?? [],
|
||||
timestamp: rawRequest.timestamp ?? new Date().toISOString(),
|
||||
sharedState: rawRequest.sharedState?.authMaterial
|
||||
? { authMaterial: rawRequest.sharedState.authMaterial }
|
||||
: undefined,
|
||||
}
|
||||
: rawRequest;
|
||||
// FNXC:SharedPostgresMultiNode 2026-07-15-00:15: Replay uses the same 10s abort as live peer sync so stop() cannot hang on a stalled peer.
|
||||
const controller = new AbortController();
|
||||
const timeoutId = setTimeout(() => controller.abort(), 10_000);
|
||||
try {
|
||||
const response = await fetch(`${node.url}/api/mesh/sync`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(body),
|
||||
signal: controller.signal,
|
||||
});
|
||||
if (!response.ok) {
|
||||
await this.centralCore.markMeshWriteFailed(entry.id, { lastError: `HTTP ${response.status}: ${response.statusText}` });
|
||||
failed += 1;
|
||||
continue;
|
||||
}
|
||||
await this.centralCore.markMeshWriteApplied(entry.id, {});
|
||||
applied += 1;
|
||||
} finally {
|
||||
clearTimeout(timeoutId);
|
||||
}
|
||||
await this.centralCore.markMeshWriteApplied(entry.id, {});
|
||||
applied += 1;
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
await this.centralCore.markMeshWriteFailed(entry.id, {
|
||||
lastError: error instanceof Error ? error.message : String(error),
|
||||
lastError: message.includes("abort") ? "Timeout (10s)" : message,
|
||||
});
|
||||
failed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
replayed: pending.length,
|
||||
replayed: eligible.length,
|
||||
applied,
|
||||
failed,
|
||||
queuedWriteIds: pending.map((entry) => entry.id),
|
||||
queuedWriteIds: eligible.map((entry) => entry.id),
|
||||
};
|
||||
}
|
||||
|
||||
private isTopologyOrAuthMeshScope(scope: string | null | undefined, entityType: string | null | undefined): boolean {
|
||||
const s = (scope ?? "").toLowerCase();
|
||||
const e = (entityType ?? "").toLowerCase();
|
||||
if (s === "mesh.sync" || s === "mesh.topology" || s === "mesh.auth") return true;
|
||||
if (e === "peer-sync" || e === "topology-sync" || e === "auth-material-sync") return true;
|
||||
// Legacy generic "shared-state-sync" rows are membership retries only when
|
||||
// we re-enqueue them under shared Postgres with topology scope; treat them
|
||||
// as eligible so pre-cutover peer-sync rows still retry once as membership.
|
||||
if (e === "shared-state-sync" && (s === "mesh.sync" || s === "mesh.topology")) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
private async enqueueRetryableSyncWrite(
|
||||
node: NodeConfig,
|
||||
request: PeerSyncRequest | undefined,
|
||||
@@ -521,16 +617,33 @@ export class PeerExchangeService {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:SharedPostgresMultiNode 2026-07-14-23:45:
|
||||
Prefer topology/auth-only queue rows. Under shared Postgres, strip settings
|
||||
and non-auth sharedState so a later replay cannot reapply DB-backed domains.
|
||||
*/
|
||||
const queueRequest: PeerSyncRequest | Record<string, unknown> = this.sharedPostgresMode
|
||||
? {
|
||||
senderNodeId: request?.senderNodeId ?? localNode.id,
|
||||
senderNodeUrl: request?.senderNodeUrl ?? localNode.url ?? "",
|
||||
knownPeers: request?.knownPeers ?? [],
|
||||
timestamp: request?.timestamp ?? new Date().toISOString(),
|
||||
...(request?.sharedState?.authMaterial
|
||||
? { sharedState: { authMaterial: request.sharedState.authMaterial } }
|
||||
: {}),
|
||||
}
|
||||
: (request ?? {});
|
||||
|
||||
const entry = await this.centralCore.enqueueMeshWrite({
|
||||
originNodeId: localNode.id,
|
||||
targetNodeId: node.id,
|
||||
projectId: null,
|
||||
scope: "mesh.sync",
|
||||
entityType: "shared-state-sync",
|
||||
scope: this.sharedPostgresMode ? "mesh.topology" : "mesh.sync",
|
||||
entityType: this.sharedPostgresMode ? "topology-sync" : "shared-state-sync",
|
||||
entityId: node.id,
|
||||
operation: "sync",
|
||||
payload: {
|
||||
request: request ?? {},
|
||||
request: queueRequest,
|
||||
error: message,
|
||||
},
|
||||
intentVersion: "1.0",
|
||||
@@ -550,11 +663,21 @@ export class PeerExchangeService {
|
||||
providerAuth?: Record<string, { type: "api_key" | "oauth"; key?: string; accessToken?: string; authenticated?: boolean }>,
|
||||
) => SharedMeshStatePayload["authMaterial"];
|
||||
};
|
||||
if (!core.getProjectSettingsSnapshot || !core.getAuthMaterialSnapshot) {
|
||||
if (!core.getAuthMaterialSnapshot && !core.getProjectSettingsSnapshot) {
|
||||
return undefined;
|
||||
}
|
||||
// Shared Postgres: never ship projectSettings over mesh (already in DB).
|
||||
const projectSettings =
|
||||
!this.sharedPostgresMode && core.getProjectSettingsSnapshot
|
||||
? await core.getProjectSettingsSnapshot(globalSettings)
|
||||
: undefined;
|
||||
const authMaterial =
|
||||
this.settingsSyncAuth && core.getAuthMaterialSnapshot
|
||||
? core.getAuthMaterialSnapshot(this.providerAuth)
|
||||
: undefined;
|
||||
if (!projectSettings && !authMaterial) {
|
||||
return undefined;
|
||||
}
|
||||
const projectSettings = await core.getProjectSettingsSnapshot(globalSettings);
|
||||
const authMaterial = this.settingsSyncAuth ? core.getAuthMaterialSnapshot(this.providerAuth) : undefined;
|
||||
this.cachedSharedStatePayload = { projectSettings, authMaterial };
|
||||
return this.cachedSharedStatePayload;
|
||||
}
|
||||
@@ -568,6 +691,29 @@ export class PeerExchangeService {
|
||||
applyAuthMaterialSnapshot?: (snapshot: NonNullable<SharedMeshStatePayload["authMaterial"]>) => { success?: boolean; authCount?: number; error?: string; providerAuth?: Record<string, unknown> };
|
||||
};
|
||||
|
||||
// Auth-only apply is always allowed (auth.json is not the shared DB).
|
||||
if (this.sharedPostgresMode) {
|
||||
if (this.settingsSyncAuth && sharedState?.authMaterial && core.applyAuthMaterialSnapshot) {
|
||||
try {
|
||||
const authResult = core.applyAuthMaterialSnapshot(sharedState.authMaterial);
|
||||
const authCount =
|
||||
typeof authResult.authCount === "number"
|
||||
? authResult.authCount
|
||||
: Object.keys(sharedState.authMaterial.payload.providerAuth ?? {}).length;
|
||||
return { success: true, globalCount: 0, projectCount: 0, authCount };
|
||||
} catch (err) {
|
||||
return {
|
||||
success: false,
|
||||
globalCount: 0,
|
||||
projectCount: 0,
|
||||
authCount: 0,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
};
|
||||
}
|
||||
}
|
||||
return { success: true, globalCount: 0, projectCount: 0, authCount: 0 };
|
||||
}
|
||||
|
||||
if (sharedState?.projectSettings && core.applyProjectSettingsSnapshot) {
|
||||
const result = await core.applyProjectSettingsSnapshot(sharedState.projectSettings);
|
||||
if (this.settingsSyncAuth && sharedState.authMaterial && core.applyAuthMaterialSnapshot) {
|
||||
|
||||
Reference in New Issue
Block a user