diff --git a/backend/src/__tests__/mesh-service.test.ts b/backend/src/__tests__/mesh-service.test.ts index bdde05eb..78fb78dc 100644 --- a/backend/src/__tests__/mesh-service.test.ts +++ b/backend/src/__tests__/mesh-service.test.ts @@ -1,6 +1,6 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; -import { getSenchoIpFromSubnet } from '../services/MeshService'; +import { getSenchoIpFromSubnet, MeshError } from '../services/MeshService'; let tmpDir: string; let MeshService: typeof import('../services/MeshService').MeshService; @@ -347,7 +347,7 @@ describe('MeshService.optInStack guard rails (network setup)', () => { }); describe('MeshService.regenerateAllOverrides (F6: boot-time regen)', () => { - it('pushes every mesh_stacks row across the fleet', async () => { + it('pushes every mesh_stacks row across the fleet and returns a summary', async () => { const svc = MeshService.getInstance(); const db = DatabaseService.getInstance(); const localNodeId = db.getNodes()[0].id; @@ -362,22 +362,29 @@ describe('MeshService.regenerateAllOverrides (F6: boot-time regen)', () => { const pushSpy = vi.spyOn(svc, 'pushOverrideToNode').mockResolvedValue(undefined); try { - await (svc as unknown as { regenerateAllOverrides: () => Promise }).regenerateAllOverrides(); + const summary = await svc.regenerateAllOverrides(); + + expect(summary.skipped).toBe(false); + expect(summary.regenerated).toBe(2); + expect(summary.failures).toEqual([]); expect(pushSpy).toHaveBeenCalledTimes(2); expect(pushSpy).toHaveBeenCalledWith(localNodeId, 'audit-mesh-prod'); expect(pushSpy).toHaveBeenCalledWith(remoteNodeId, 'audit-mesh-pilot'); const activity = svc.getActivity({ limit: 100 }); - expect(activity.some((e) => e.message === 'boot regenerated 2 override(s)')).toBe(true); + expect(activity.some((e) => + e.message === 'mesh override regen complete: 2 succeeded, 0 failed across 0 node(s)', + )).toBe(true); } finally { db.deleteNode(remoteNodeId); } }); - it('skips entirely when senchoIp is null (network setup failed)', async () => { + it('skips entirely with a reason when senchoIp is null (network setup failed)', async () => { const svc = MeshService.getInstance(); (svc as unknown as { senchoIp: string | null }).senchoIp = null; + (svc as unknown as { networkSetupError: string | null }).networkSetupError = 'sencho_mesh subnet mismatch'; const db = DatabaseService.getInstance(); const localNodeId = db.getNodes()[0].id; @@ -385,12 +392,20 @@ describe('MeshService.regenerateAllOverrides (F6: boot-time regen)', () => { const pushSpy = vi.spyOn(svc, 'pushOverrideToNode').mockResolvedValue(undefined); - await (svc as unknown as { regenerateAllOverrides: () => Promise }).regenerateAllOverrides(); + const summary = await svc.regenerateAllOverrides(); + expect(summary.skipped).toBe(true); + expect(summary.reason).toBe('sencho_mesh subnet mismatch'); + expect(summary.regenerated).toBe(0); expect(pushSpy).not.toHaveBeenCalled(); + + const activity = svc.getActivity({ limit: 100 }); + expect(activity.some((e) => + e.level === 'warn' && /data plane unavailable/.test(e.message), + )).toBe(true); }); - it('logs a warning per stack when push fails but does not throw', async () => { + it('records per-stack failures in the summary and aggregates by node', async () => { const svc = MeshService.getInstance(); const db = DatabaseService.getInstance(); const localNodeId = db.getNodes()[0].id; @@ -399,13 +414,90 @@ describe('MeshService.regenerateAllOverrides (F6: boot-time regen)', () => { vi.spyOn(svc, 'pushOverrideToNode').mockRejectedValue(new Error('remote node offline')); - await expect( - (svc as unknown as { regenerateAllOverrides: () => Promise }).regenerateAllOverrides(), - ).resolves.toBeUndefined(); + const summary = await svc.regenerateAllOverrides(); + + expect(summary.skipped).toBe(false); + expect(summary.regenerated).toBe(0); + expect(summary.failures).toEqual([{ + nodeId: localNodeId, + stackName: 'audit-mesh-prod', + message: 'remote node offline', + }]); const activity = svc.getActivity({ limit: 100 }); expect(activity.some((e) => - e.level === 'warn' && /boot override regen failed for audit-mesh-prod/.test(e.message), + e.level === 'warn' && /mesh override regen failed for audit-mesh-prod/.test(e.message), + )).toBe(true); + expect(activity.some((e) => + /mesh override regen complete: 0 succeeded, 1 failed across 1 node\(s\)/.test(e.message), )).toBe(true); }); + + it('continues past version-skew 404 push failures and reports them in the summary', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const localNodeId = db.getNodes()[0].id; + const remoteNodeId = db.addNode({ + name: 'remote-old-pilot', type: 'remote', mode: 'pilot_agent', + compose_dir: '/tmp', is_default: false, api_url: '', api_token: '', + }); + + db.insertMeshStack(localNodeId, 'audit-mesh-prod', 'tester'); + db.insertMeshStack(remoteNodeId, 'audit-mesh-pilot', 'tester'); + + vi.spyOn(svc, 'pushOverrideToNode').mockImplementation(async (nodeId: number) => { + if (nodeId === remoteNodeId) { + throw new MeshError('push_failed', 'node remote-old-pilot does not support mesh override push (upgrade required)'); + } + }); + + try { + const summary = await svc.regenerateAllOverrides(); + + expect(summary.regenerated).toBe(1); + expect(summary.failures).toHaveLength(1); + expect(summary.failures[0].nodeId).toBe(remoteNodeId); + expect(summary.failures[0].stackName).toBe('audit-mesh-pilot'); + expect(summary.failures[0].message).toMatch(/upgrade required/); + } finally { + db.deleteNode(remoteNodeId); + } + }); + + it('completes both flows without throwing when regen and opt-in are scheduled concurrently', async () => { + // The race is benign by design: both writes target the same alias list + // with the same `senchoIp`, so even if regen reads `mesh_stacks` mid- + // opt-in and pushes a concurrent override for the new stack, the + // resulting on-disk file is identical. This test locks the no-throw + // contract; it does not attempt to force a specific interleave because + // `regenerateAllOverrides` snapshots the table synchronously before + // its first await, so under the JS event loop the two flows always + // resolve cleanly without a real read-after-write window. + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const localNodeId = db.getNodes()[0].id; + + db.insertMeshStack(localNodeId, 'existing-stack', 'tester'); + + vi.spyOn(svc as unknown as { inspectStackServices: (n: number, s: string) => Promise }, 'inspectStackServices') + .mockResolvedValue([{ service: 'web', ports: [8080] }]); + const pushSpy = vi.spyOn(svc, 'pushOverrideToNode').mockResolvedValue(undefined); + vi.spyOn(svc as unknown as { triggerRedeploy: (n: number, s: string, a: string) => void }, 'triggerRedeploy') + .mockImplementation(() => { /* noop */ }); + vi.spyOn(svc as unknown as { regenerateOverridesForNode: (n: number, skip?: string) => Promise }, 'regenerateOverridesForNode') + .mockResolvedValue(undefined); + + const [summary] = await Promise.all([ + svc.regenerateAllOverrides(), + svc.optInStack(localNodeId, 'concurrent-stack', 'tester'), + ]); + + expect(summary.regenerated).toBeGreaterThanOrEqual(1); + expect(db.isMeshStackEnabled(localNodeId, 'concurrent-stack')).toBe(true); + + const existingCalls = pushSpy.mock.calls.filter((c) => c[1] === 'existing-stack'); + expect(existingCalls.length).toBeGreaterThanOrEqual(1); + const concurrentCalls = pushSpy.mock.calls.filter((c) => c[1] === 'concurrent-stack'); + expect(concurrentCalls.length).toBeGreaterThanOrEqual(1); + }); }); diff --git a/backend/src/routes/mesh.ts b/backend/src/routes/mesh.ts index d987c8ad..f3f6d16f 100644 --- a/backend/src/routes/mesh.ts +++ b/backend/src/routes/mesh.ts @@ -1,7 +1,7 @@ import { Router, type Request, type Response } from 'express'; import { DatabaseService } from '../services/DatabaseService'; import { NodeRegistry } from '../services/NodeRegistry'; -import { MeshError, MeshService } from '../services/MeshService'; +import { MeshError, MeshService, type MeshRegenSummary } from '../services/MeshService'; import { requireAdmin, requireAdmiral } from '../middleware/tierGates'; import { sanitizeForLog } from '../utils/safeLog'; import { isValidStackName } from '../utils/validation'; @@ -24,6 +24,47 @@ meshRouter.get('/status', async (_req: Request, res: Response): Promise => } }); +/** + * Operator-triggered rerun of the boot-time override regeneration. Walks + * every `mesh_stacks` row across the fleet and re-pushes each override to + * its owning node. Useful when a remote node was offline at central boot + * and the override files there are stale; previously the only recovery + * path was opt-out + opt-in for every meshed stack on that node. + */ +meshRouter.post('/regen-overrides', async (req: Request, res: Response): Promise => { + if (!requireAdmiral(req, res)) return; + if (!requireAdmin(req, res)) return; + const actor = actorFor(req); + let summary: MeshRegenSummary | null = null; + let outcome: 'success' | 'skipped' | 'partial' | 'error' = 'error'; + try { + summary = await MeshService.getInstance().regenerateAllOverrides(); + outcome = summary.skipped ? 'skipped' : (summary.failures.length === 0 ? 'success' : 'partial'); + res.json(summary); + } catch (err) { + outcome = 'error'; + console.warn('[mesh] /regen-overrides failed:', sanitizeForLog((err as Error).message)); + res.status(500).json({ error: 'Failed to regenerate mesh overrides' }); + } finally { + try { + DatabaseService.getInstance().insertAuditLog({ + timestamp: Date.now(), + username: actor, + method: 'POST', + path: req.path, + status_code: res.statusCode, + node_id: null, + ip_address: req.ip ?? 'unknown', + summary: summary + ? `Mesh override regen ${outcome}: ${summary.regenerated} regenerated, ${summary.failures.length} failed` + : `Mesh override regen ${outcome}`, + }); + } catch (auditErr) { + console.error('[mesh] Audit log insert failed:', auditErr); + } + } +}); + meshRouter.post('/nodes/:nodeId/enable', async (req: Request, res: Response): Promise => { if (!requireAdmiral(req, res)) return; if (!requireAdmin(req, res)) return; diff --git a/backend/src/services/MeshService.ts b/backend/src/services/MeshService.ts index 66552b5e..bd649612 100644 --- a/backend/src/services/MeshService.ts +++ b/backend/src/services/MeshService.ts @@ -88,6 +88,19 @@ export interface MeshTarget { alias: string; } +export interface MeshRegenFailure { + nodeId: number; + stackName: string; + message: string; +} + +export interface MeshRegenSummary { + regenerated: number; + failures: MeshRegenFailure[]; + skipped: boolean; + reason?: string; +} + export interface MeshNodeStatus { nodeId: number; nodeName: string; @@ -192,8 +205,22 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { })); await this.setupMeshNetwork(); - await this.refreshAliasCache(); - await this.syncForwarderListeners(); + try { + await this.refreshAliasCache(); + } catch (err) { + this.logActivity({ + source: 'mesh', level: 'error', type: 'forwarder.error', + message: `boot refreshAliasCache failed: ${sanitizeForLog((err as Error).message)}`, + }); + } + try { + await this.syncForwarderListeners(); + } catch (err) { + this.logActivity({ + source: 'mesh', level: 'error', type: 'forwarder.error', + message: `boot syncForwarderListeners failed: ${sanitizeForLog((err as Error).message)}`, + }); + } await this.regenerateAllOverrides(); this.aliasRefreshTimer = setInterval(() => { void (async () => { @@ -206,9 +233,10 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { })(); }, ALIAS_REFRESH_INTERVAL_MS); + const dataPlane = this.senchoIp ? 'ok' : `unavailable (${this.networkSetupError ?? 'unknown'})`; this.logActivity({ - source: 'mesh', level: 'info', type: 'mesh.enable', - message: 'MeshService started', + source: 'mesh', level: this.senchoIp ? 'info' : 'warn', type: 'mesh.enable', + message: `MeshService started (data plane ${dataPlane})`, }); } @@ -651,32 +679,48 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { * Walk every `mesh_stacks` row across the fleet and re-push each override * to its owning node. Called once at boot so on-disk override files * survive a Sencho restart even if they were lost (image rebuild, volume - * reset, manual cleanup). Best-effort: failures are logged per-stack and - * other nodes still get regenerated. An offline remote node leaves stale - * overrides until the next opt-in / opt-out on that node. + * reset, manual cleanup). Also exposed as `POST /api/mesh/regen-overrides` + * so an operator can rerun it after fixing a remote node that was offline + * at boot. Best-effort: failures are logged per-stack and other nodes + * still get regenerated. An offline remote node leaves stale overrides + * until the next opt-in / opt-out on that node, or the next manual rerun. */ - private async regenerateAllOverrides(): Promise { - if (!this.senchoIp) return; + public async regenerateAllOverrides(): Promise { + if (!this.senchoIp) { + const reason = this.networkSetupError ?? 'mesh data plane unavailable'; + this.logActivity({ + source: 'mesh', level: 'warn', type: 'mesh.disable', + message: `mesh override regen skipped: data plane unavailable (${sanitizeForLog(reason)})`, + }); + return { regenerated: 0, failures: [], skipped: true, reason }; + } const db = DatabaseService.getInstance(); const stacks = db.listMeshStacks(); + const failures: MeshRegenFailure[] = []; await Promise.allSettled( stacks.map(async (s) => { try { await this.pushOverrideToNode(s.node_id, s.stack_name); } catch (err) { + const message = sanitizeForLog((err as Error).message); + failures.push({ nodeId: s.node_id, stackName: s.stack_name, message }); this.logActivity({ source: 'mesh', level: 'warn', type: 'forwarder.error', nodeId: s.node_id, - message: `boot override regen failed for ${s.stack_name}: ${sanitizeForLog((err as Error).message)}`, + message: `mesh override regen failed for ${s.stack_name}: ${message}`, details: { stackName: s.stack_name }, }); } }), ); + const succeeded = stacks.length - failures.length; + const failedNodeIds = Array.from(new Set(failures.map((f) => f.nodeId))).sort((a, b) => a - b); this.logActivity({ - source: 'mesh', level: 'info', type: 'mesh.enable', - message: `boot regenerated ${stacks.length} override(s)`, + source: 'mesh', level: failures.length === 0 ? 'info' : 'warn', type: 'mesh.enable', + message: `mesh override regen complete: ${succeeded} succeeded, ${failures.length} failed across ${failedNodeIds.length} node(s)`, + details: { succeeded, failed: failures.length, failedNodeIds }, }); + return { regenerated: succeeded, failures, skipped: false }; } // --- Alias aggregation --- @@ -1167,7 +1211,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { this.logActivity({ source: 'pilot', level: 'error', type: 'tunnel.fail', nodeId: target.nodeId, alias: target.alias, streamId: record.streamId, - message: err.message, + message: sanitizeForLog(err.message), }); this.activeStreams.delete(record.streamId); try { src.destroy(); } catch { /* ignore */ } @@ -1242,11 +1286,12 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { }); stream.once('error', (err: Error) => { clearTimeout(timer); + const sanitized = sanitizeForLog(err.message); this.logActivity({ source: 'mesh', level: 'error', type: 'probe.fail', - alias: target.host, message: err.message, + alias: target.host, message: sanitized, }); - resolve({ ok: false, where: 'agent_dial', code: 'unreachable', message: err.message }); + resolve({ ok: false, where: 'agent_dial', code: 'unreachable', message: sanitized }); }); }); } @@ -1266,7 +1311,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { resolve({ ok: false, where: 'target_port', code: 'timeout', message: 'connect timeout' }); }); sock.once('error', (err) => { - resolve({ ok: false, where: 'target_port', code: 'unreachable', message: err.message }); + resolve({ ok: false, where: 'target_port', code: 'unreachable', message: sanitizeForLog(err.message) }); }); }); }