From 523179cba17d2f35469f247f5d8accdf0c036637 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sat, 26 Sep 2026 21:23:24 +0000 Subject: [PATCH] =?UTF-8?q?feat(desktop):=20Fleet=20Sync=20=E2=80=94=20pus?= =?UTF-8?q?h=20one=20path=20from=20a=20source=20host=20to=20many=20at=20on?= =?UTF-8?q?ce?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a "Fleet Sync" tab: pick a source (a server or this machine) + a path, tick destination servers, and it syncs that path out to all of them at once — e.g. a dir on dev1 to the other 7 servers. It is pure orchestration over the existing transfer engine: one transfer per destination (server-to-server when the source is remote, local-to-remote when it is this machine), run with bounded concurrency, each streaming its rsync progress on a per-connection channel. No new transfer logic — buildPlan/runPlan are reused wholesale. - contract: `fleet:sync` + `fleet:sync-cancel` channels, `event:fleet-sync`, `FleetSyncRequestSchema` (source EndpointRef, destinationConnectionIds, destinationPath, options, concurrency). `delete` is intentionally not exposed — a pruning mirror is a per-host previewed decision, not a fan-out default. - main: `services/fleet-sync.ts` `startFleetSync`/`cancelFleetSync`; a worker pool drains the destination queue, a per-destination failure never takes the fan-out down. `buildPlan` is exported from transfers.ts for reuse. - preload: `fleet.sync` / `fleet.cancelSync` / `events.onFleetSync`. - renderer: self-contained `FleetSyncPanel` (source + path, destination checklist, dest path, per-destination progress rows) on a new tab; the existing FleetView is untouched. Tests: fleet-sync.test.ts (schema validation + orchestration: one transfer per dest, per-connection begin/exit events, a done marker), with buildPlan/runPlan stubbed. Desktop suite 165 pass; renderer + electron typecheck clean. The UI is typecheck-verified, not yet visually verified in a running app. Co-Authored-By: Claude Opus 4.8 --- apps/desktop/electron/main/ipc.ts | 7 + .../electron/main/services/fleet-sync.test.ts | 73 ++++++ .../electron/main/services/fleet-sync.ts | 83 +++++++ .../electron/main/services/transfers.ts | 2 +- apps/desktop/electron/preload/index.ts | 8 + apps/desktop/electron/shared/contract.ts | 30 +++ apps/desktop/src/app/page.tsx | 12 +- .../src/components/fleet-sync-panel.tsx | 216 ++++++++++++++++++ apps/desktop/src/lib/api.ts | 10 + 9 files changed, 439 insertions(+), 2 deletions(-) create mode 100644 apps/desktop/electron/main/services/fleet-sync.test.ts create mode 100644 apps/desktop/electron/main/services/fleet-sync.ts create mode 100644 apps/desktop/src/components/fleet-sync-panel.tsx diff --git a/apps/desktop/electron/main/ipc.ts b/apps/desktop/electron/main/ipc.ts index ecc1fde..ecfa468 100644 --- a/apps/desktop/electron/main/ipc.ts +++ b/apps/desktop/electron/main/ipc.ts @@ -10,6 +10,8 @@ import { DeleteEntryRequestSchema, ExternalUrlSchema, FleetCheckRequestSchema, + FleetSyncRequestSchema, + FleetSyncIdSchema, FleetCommandSaveSchema, FleetListRenameSchema, FleetListSaveSchema, @@ -48,6 +50,7 @@ import { previewFleet, startFleet, } from './services/fleet.js' +import { startFleetSync, cancelFleetSync } from './services/fleet-sync.js' import { browserFor, dropSession, sessionFor } from './services/sessions.js' import { store } from './services/store.js' import { advanceSeries, handlersFor, openSeries, openWith, stopSeries } from './services/open-with.js' @@ -378,6 +381,10 @@ export function registerIpc(): void { handle(IPC.fleetCheck, FleetCheckRequestSchema, async (input) => checkFleetServers(input)) + handle(IPC.fleetSync, FleetSyncRequestSchema, async (request, event) => startFleetSync(request, event.sender)) + + handle(IPC.fleetSyncCancel, z.object({ syncId: FleetSyncIdSchema }), async ({ syncId }) => cancelFleetSync(syncId)) + handle(IPC.fleetRuns, z.object({ limit: z.number().int().min(1).max(200).default(25) }), async ({ limit }) => (await store()).listFleetRuns(limit), ) diff --git a/apps/desktop/electron/main/services/fleet-sync.test.ts b/apps/desktop/electron/main/services/fleet-sync.test.ts new file mode 100644 index 0000000..266c7c3 --- /dev/null +++ b/apps/desktop/electron/main/services/fleet-sync.test.ts @@ -0,0 +1,73 @@ +import { describe, it, expect, vi } from 'vitest' +import { FleetSyncRequestSchema } from '../../shared/contract.js' + +// The transfer engine is exercised elsewhere; here we test only the fan-out +// orchestration, so buildPlan/runPlan are stubbed. +vi.mock('./transfers.js', () => ({ + buildPlan: vi.fn(async () => ({ + plan: { display: '', controlDisplay: null, warnings: [] }, + cleanup: async () => {}, + })), +})) +vi.mock('@diskpush/rsync-core', () => ({ + runPlan: vi.fn(() => ({ + events: (async function* () { + yield { type: 'progress', progress: { percent: 50, bytesTransferred: 1, bytesPerSecond: 1, elapsedSeconds: 1 } } + yield { type: 'exit', code: 0, resumable: false, message: '' } + })(), + cancel: () => {}, + })), +})) + +const { startFleetSync } = await import('./fleet-sync.js') + +function fakeSender() { + const sends: Array<{ ch: string; payload: { syncId: string; connectionId: string | null; event: { type: string } } }> = [] + const sender = { + isDestroyed: () => false, + send: (ch: string, payload: unknown) => sends.push({ ch, payload: payload as never }), + } as never + return { sends, sender } +} + +const req = (dests: string[]) => + FleetSyncRequestSchema.parse({ + source: { type: 'ssh', connectionId: 'src', path: '/data' }, + destinationConnectionIds: dests, + destinationPath: '/data', + options: {}, + concurrency: 2, + }) + +describe('FleetSyncRequestSchema', () => { + it('requires at least one destination', () => { + expect(() => req([])).toThrow() + }) + it('defaults concurrency', () => { + const parsed = FleetSyncRequestSchema.parse({ + source: { type: 'ssh', connectionId: 'src', path: '/data' }, + destinationConnectionIds: ['a'], + destinationPath: '/data', + options: {}, + }) + expect(parsed.concurrency).toBe(4) + }) +}) + +describe('startFleetSync', () => { + it('runs one transfer per destination, streams per-connection events, then a done marker', async () => { + const { sends, sender } = fakeSender() + const res = await startFleetSync(req(['a', 'b']), sender) + expect(res.destinations).toEqual(['a', 'b']) + + // Let the async worker pool drain. + await new Promise((r) => setTimeout(r, 30)) + + for (const id of ['a', 'b']) { + const forId = sends.filter((s) => s.payload.connectionId === id) + expect(forId.some((s) => s.payload.event.type === 'begin'), `${id} begins`).toBe(true) + expect(forId.some((s) => s.payload.event.type === 'exit'), `${id} exits`).toBe(true) + } + expect(sends.some((s) => s.payload.connectionId === null && s.payload.event.type === 'done')).toBe(true) + }) +}) diff --git a/apps/desktop/electron/main/services/fleet-sync.ts b/apps/desktop/electron/main/services/fleet-sync.ts new file mode 100644 index 0000000..b716130 --- /dev/null +++ b/apps/desktop/electron/main/services/fleet-sync.ts @@ -0,0 +1,83 @@ +import { randomUUID } from 'node:crypto' +import type { WebContents } from 'electron' +import { runPlan } from '@diskpush/rsync-core' +import { IPC, type FleetSyncRequest, type TransferRequest } from '../../shared/contract.js' +import { buildPlan } from './transfers.js' + +/** + * Fleet sync — push one source path out to many hosts at once. + * + * Each destination is a normal transfer (server-to-server when the source is an + * SSH host, local-to-remote when it is this machine), so this is pure + * orchestration: build a plan per destination, run them with bounded + * concurrency, and stream each one's rsync events tagged with its connection id. + * The renderer renders a row per destination from that stream. Nothing about how + * a transfer runs lives here — it is reused wholesale from the transfer path. + */ + +type RunningSync = { cancels: Map void> } +const running = new Map() + +export type StartedFleetSync = { syncId: string; destinations: string[] } + +export async function startFleetSync(request: FleetSyncRequest, sender: WebContents): Promise { + const syncId = randomUUID() + const cancels = new Map void>() + running.set(syncId, { cancels }) + + const dests = [...request.destinationConnectionIds] + const emit = (connectionId: string | null, event: unknown) => { + if (!sender.isDestroyed()) sender.send(IPC.eventFleetSync, { syncId, connectionId, event }) + } + + const runOne = async (destId: string): Promise => { + const req: TransferRequest = { + source: request.source, + destination: { type: 'ssh', connectionId: destId, path: request.destinationPath }, + options: request.options, + deletesConfirmed: false, + selection: [], + } + let cleanup: () => Promise = async () => {} + emit(destId, { type: 'begin' }) + try { + const built = await buildPlan(req) + cleanup = built.cleanup + const handle = runPlan(built.plan) + cancels.set(destId, handle.cancel) + for await (const event of handle.events) emit(destId, event) + } catch (error) { + // A per-destination failure never takes the fan-out down; it is reported + // on that destination's row and the others carry on. + emit(destId, { type: 'error', message: error instanceof Error ? error.message : String(error) }) + } finally { + cancels.delete(destId) + await cleanup() + } + } + + // Bounded concurrency: a worker pool draining the destination queue. + void (async () => { + const queue = [...dests] + const worker = async (): Promise => { + for (;;) { + const destId = queue.shift() + if (destId === undefined) return + await runOne(destId) + } + } + const width = Math.max(1, Math.min(request.concurrency, dests.length)) + await Promise.all(Array.from({ length: width }, () => worker())) + running.delete(syncId) + emit(null, { type: 'done' }) + })() + + return { syncId, destinations: dests } +} + +export function cancelFleetSync(syncId: string): boolean { + const entry = running.get(syncId) + if (!entry) return false + for (const cancel of entry.cancels.values()) cancel() + return true +} diff --git a/apps/desktop/electron/main/services/transfers.ts b/apps/desktop/electron/main/services/transfers.ts index ed8dd5a..c18261d 100644 --- a/apps/desktop/electron/main/services/transfers.ts +++ b/apps/desktop/electron/main/services/transfers.ts @@ -135,7 +135,7 @@ async function writeSelectionList( type BuiltPlan = { plan: ExecutionPlan; cleanup: () => Promise } -async function buildPlan(request: TransferRequest, overrides: Partial = {}): Promise { +export async function buildPlan(request: TransferRequest, overrides: Partial = {}): Promise { const source = await resolveEndpoint(request.source) const destination = await resolveEndpoint(request.destination) const capabilities = await capabilitiesFor([source.connectionId, destination.connectionId]) diff --git a/apps/desktop/electron/preload/index.ts b/apps/desktop/electron/preload/index.ts index 3b2170b..316efd9 100644 --- a/apps/desktop/electron/preload/index.ts +++ b/apps/desktop/electron/preload/index.ts @@ -66,6 +66,8 @@ const api = { cancel: (runId: string) => call(IPC.fleetCancel, { runId }), check: (connectionIds: string[], concurrency = 4, timeoutSeconds = 180) => call(IPC.fleetCheck, { connectionIds, concurrency, timeoutSeconds }), + sync: (request: unknown) => call(IPC.fleetSync, request), + cancelSync: (syncId: string) => call(IPC.fleetSyncCancel, { syncId }), runs: (limit = 25) => call(IPC.fleetRuns, { limit }), saveCommand: (input: unknown) => call(IPC.fleetCommandSave, input), removeCommand: (name: string) => call(IPC.fleetCommandRemove, { name }), @@ -108,6 +110,12 @@ const api = { ipcRenderer.on(IPC.eventFleet, wrapped) return () => ipcRenderer.off(IPC.eventFleet, wrapped) }, + onFleetSync(listener: (payload: { syncId: string; connectionId: string | null; event: unknown }) => void): () => void { + const wrapped = (_event: unknown, payload: { syncId: string; connectionId: string | null; event: unknown }) => + listener(payload) + ipcRenderer.on(IPC.eventFleetSync, wrapped) + return () => ipcRenderer.off(IPC.eventFleetSync, wrapped) + }, onOpenSeries(listener: (payload: { seriesId: string; event: unknown }) => void): () => void { const wrapped = (_event: unknown, payload: { seriesId: string; event: unknown }) => listener(payload) ipcRenderer.on(IPC.eventOpenSeries, wrapped) diff --git a/apps/desktop/electron/shared/contract.ts b/apps/desktop/electron/shared/contract.ts index a263088..4f044ed 100644 --- a/apps/desktop/electron/shared/contract.ts +++ b/apps/desktop/electron/shared/contract.ts @@ -59,6 +59,10 @@ export const IPC = { fleetListRemove: 'fleet:list-remove', fleetRunDetail: 'fleet:run-detail', + /** Sync one path from a source host to many fleet hosts at once. */ + fleetSync: 'fleet:sync', + fleetSyncCancel: 'fleet:sync-cancel', + shellOpenExternal: 'shell:open-external', pluginsList: 'plugins:list', @@ -79,6 +83,8 @@ export const IPC = { eventOpenSeries: 'event:open-series', /** Main -> renderer, one channel carrying every fleet event. */ eventFleet: 'event:fleet', + /** Main -> renderer, per-destination progress for a fleet sync. */ + eventFleetSync: 'event:fleet-sync', /** * Main -> renderer, live progress for a dry run that has not finished yet. * @@ -384,6 +390,30 @@ export const FleetListRenameSchema = z.object({ to: FleetListNameSchema, }) +/** + * Sync one source path out to many fleet hosts at once. + * + * The renderer names the source (a saved connection + a path, or the local + * machine) and the destination connections by id; the main process resolves + * them, builds one server-to-server (or local-to-remote) transfer per + * destination, and runs them with bounded concurrency. As with a normal + * transfer, the renderer supplies a subset of options and never a raw rsync + * command, a remote shell, or an rsync binary path — those come from the saved + * connections. `delete` is intentionally NOT exposed here: a mirror that prunes + * the destination is a per-host, previewed decision, not a fan-out default. + */ +export const FleetSyncRequestSchema = z.object({ + source: EndpointRefSchema, + destinationConnectionIds: z.array(ConnectionIdSchema).min(1).max(500), + /** Where it lands on each destination; defaults to the source path. */ + destinationPath: PathSchema, + options: TransferOptionsSchema, + concurrency: z.number().int().min(1).max(64).default(4), +}) +export type FleetSyncRequest = z.infer + +export const FleetSyncIdSchema = z.string().min(1).max(128) + /** * Plugins. * diff --git a/apps/desktop/src/app/page.tsx b/apps/desktop/src/app/page.tsx index aed4f1a..6ac03b9 100644 --- a/apps/desktop/src/app/page.tsx +++ b/apps/desktop/src/app/page.tsx @@ -4,6 +4,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import Image from 'next/image' import { ArrowLeftRight, + HardDriveDownload, CircleAlert, CircleCheck, ExternalLink, @@ -18,6 +19,7 @@ import { } from 'lucide-react' import { ConnectionDialog } from '@/components/connection-dialog' import { FleetView } from '@/components/fleet-view' +import { FleetSyncPanel } from '@/components/fleet-sync-panel' import { PluginsDialog } from '@/components/plugins-dialog' import { ProfileBar } from '@/components/profile-bar' import { ServerManager } from '@/components/server-manager' @@ -134,7 +136,7 @@ export default function Workspace() { const [showPlugins, setShowPlugins] = useState(false) /** Which pane a plugin job is working in, so the pane is re-read when it changes files. */ const pluginSideRef = useRef<{ jobId: string; side: 'left' | 'right' } | null>(null) - const [tab, setTab] = useState<'transfer' | 'fleet'>('transfer') + const [tab, setTab] = useState<'transfer' | 'fleet' | 'sync'>('transfer') const [profiles, setProfiles] = useState([]) const [outsideShell, setOutsideShell] = useState(false) @@ -573,6 +575,10 @@ export default function Workspace() { Fleet + setTab('sync')}> + + Fleet Sync +
@@ -748,6 +754,10 @@ export default function Workspace() { setShowServers(true)} />
+
+ +
+ setShowServers(false)} diff --git a/apps/desktop/src/components/fleet-sync-panel.tsx b/apps/desktop/src/components/fleet-sync-panel.tsx new file mode 100644 index 0000000..a550cc9 --- /dev/null +++ b/apps/desktop/src/components/fleet-sync-panel.tsx @@ -0,0 +1,216 @@ +'use client' + +import { useCallback, useEffect, useMemo, useState } from 'react' +import { CircleAlert, CircleCheck, CircleDashed, HardDriveDownload, Loader2, Play, Square } from 'lucide-react' +import { Button } from '@/components/ui/button' +import { Checkbox } from '@/components/ui/checkbox' +import { Input } from '@/components/ui/input' +import { ScrollArea } from '@/components/ui/scroll-area' +import { api, unwrap, type Connection, type FleetSyncEvent } from '@/lib/api' + +/** + * Fleet Sync — push one source path out to many hosts at once. + * + * Pick a source (a server or this machine) + a path, tick the destination + * servers, and it runs one transfer per destination (server-to-server when the + * source is remote), streaming a progress row for each. The heavy lifting is the + * same transfer engine the two-pane view uses; this is just the fan-out. + */ + +type DestState = 'pending' | 'running' | 'done' | 'error' +type DestView = { id: string; name: string; host: string; state: DestState; percent: number; message: string } + +const STATE_ICON: Record = { + pending: , + running: , + done: , + error: , +} + +export function FleetSyncPanel() { + const [servers, setServers] = useState([]) + const [sourceId, setSourceId] = useState('') + const [sourcePath, setSourcePath] = useState('') + const [destPath, setDestPath] = useState('') + const [selected, setSelected] = useState>(new Set()) + const [syncId, setSyncId] = useState(null) + const [dests, setDests] = useState>({}) + + const active = useMemo( + () => Object.values(dests).some((d) => d.state === 'pending' || d.state === 'running'), + [dests], + ) + + useEffect(() => { + void (async () => { + const list = (await unwrap(api()?.fleet.servers())) ?? [] + setServers(list) + setSourceId((cur) => cur || list[0]?.id || '') + })() + }, []) + + useEffect(() => { + const off = api()?.events.onFleetSync(({ connectionId, event }) => { + if (!connectionId) return // the null-keyed 'done' marker + setDests((prev) => { + const cur = prev[connectionId] + if (!cur) return prev + const e = event as FleetSyncEvent + const next = { ...cur } + if (e.type === 'begin') next.state = 'running' + else if (e.type === 'progress') { + next.state = 'running' + next.percent = Math.max(0, Math.min(100, Math.round(e.progress.percent))) + } else if (e.type === 'stderr') next.message = e.line + else if (e.type === 'error') { + next.state = 'error' + next.message = e.message + } else if (e.type === 'exit') { + const ok = e.code === 0 || e.code === 24 + next.state = ok ? 'done' : 'error' + next.percent = ok ? 100 : next.percent + if (!ok) next.message = e.message + } + return { ...prev, [connectionId]: next } + }) + }) + return () => off?.() + }, []) + + const destServers = useMemo(() => servers.filter((s) => s.id !== sourceId), [servers, sourceId]) + const canRun = Boolean(sourceId) && sourcePath.trim().length > 0 && selected.size > 0 && !active + + const toggle = (id: string) => + setSelected((prev) => { + const next = new Set(prev) + next.has(id) ? next.delete(id) : next.add(id) + return next + }) + + const run = useCallback(async () => { + const ids = [...selected].filter((id) => id !== sourceId) + if (!sourceId || ids.length === 0 || sourcePath.trim().length === 0) return + + const initial: Record = {} + for (const id of ids) { + const s = servers.find((x) => x.id === id) + initial[id] = { id, name: s?.name ?? id, host: s?.host ?? '', state: 'pending', percent: 0, message: '' } + } + setDests(initial) + + const source = + sourceId === 'local' + ? { type: 'local' as const, path: sourcePath.trim() } + : { type: 'ssh' as const, connectionId: sourceId, path: sourcePath.trim() } + + const res = await unwrap( + api()?.fleet.sync({ + source, + destinationConnectionIds: ids, + destinationPath: destPath.trim() || sourcePath.trim(), + options: {}, + concurrency: 4, + }), + ) + setSyncId(res?.syncId ?? null) + }, [selected, sourceId, sourcePath, destPath, servers]) + + const cancel = useCallback(async () => { + if (syncId) await api()?.fleet.cancelSync(syncId) + }, [syncId]) + + return ( +
+
+ + Sync a path to the fleet +
+ + {/* Source */} +
+ + + setSourcePath(e.target.value)} + placeholder="/path/on/source" + className="h-[var(--control)] min-w-[240px] flex-1 text-[12px]" + /> +
+ + {/* Destination path */} +
+ + setDestPath(e.target.value)} + placeholder="defaults to the source path" + className="h-[var(--control)] min-w-[240px] flex-1 text-[12px]" + /> +
+ + {/* Destination servers */} +
+ Destinations {selected.size > 0 ? `· ${[...selected].filter((id) => id !== sourceId).length}` : ''} +
+ +
+ {destServers.map((s) => ( + + ))} +
+ + + {/* Actions */} +
+ + {active ? ( + + ) : null} +
+ + {/* Per-destination progress */} + +
+ {Object.values(dests).map((d) => ( +
+ {STATE_ICON[d.state]} + {d.name} +
+
+
+ {d.percent}% + {d.message ? {d.message} : null} +
+ ))} + {Object.keys(dests).length === 0 ? ( +
Pick a source, a path, and destinations, then Sync.
+ ) : null} +
+ +
+ ) +} diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts index b0dc831..d2750d5 100644 --- a/apps/desktop/src/lib/api.ts +++ b/apps/desktop/src/lib/api.ts @@ -150,6 +150,13 @@ export type TransferEvent = | { type: 'stderr' | 'stdout'; line: string } | { type: 'exit'; code: number; resumable: boolean; message: string } +/** Per-destination events for a fleet sync: rsync events plus the orchestrator's own. */ +export type FleetSyncEvent = + | TransferEvent + | { type: 'begin' } + | { type: 'error'; message: string } + | { type: 'done' } + export type FleetHostState = | 'pending' | 'connecting' @@ -368,6 +375,8 @@ type Api = { start(request: FleetRequest): Promise> cancel(runId: string): Promise> check(connectionIds: string[], concurrency?: number, timeoutSeconds?: number): Promise> + sync(request: unknown): Promise> + cancelSync(syncId: string): Promise> runs(limit?: number): Promise> runDetail(runId: string): Promise> saveCommand(input: FleetCommandSave): Promise> @@ -391,6 +400,7 @@ type Api = { events: { onTransfer(listener: (payload: { jobId: string; event: TransferEvent }) => void): () => void onFleet(listener: (payload: { runId: string; event: FleetEvent }) => void): () => void + onFleetSync(listener: (payload: { syncId: string; connectionId: string | null; event: FleetSyncEvent }) => void): () => void onPreview(listener: (payload: { previewId: string; progress: PreviewProgress }) => void): () => void onOpenSeries(listener: (payload: { seriesId: string; event: SeriesEvent }) => void): () => void onPluginProgress(listener: (payload: { jobId: string; event: PluginEvent }) => void): () => void