feat(desktop): Fleet Sync — push a path from one host to many at once

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),
   )

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)
+  })
+})

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<string, () => void> }
+const running = new Map<string, RunningSync>()
+
+export type StartedFleetSync = { syncId: string; destinations: string[] }
+
+export async function startFleetSync(request: FleetSyncRequest, sender: WebContents): Promise<StartedFleetSync> {
+  const syncId = randomUUID()
+  const cancels = new Map<string, () => 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<void> => {
+    const req: TransferRequest = {
+      source: request.source,
+      destination: { type: 'ssh', connectionId: destId, path: request.destinationPath },
+      options: request.options,
+      deletesConfirmed: false,
+      selection: [],
+    }
+    let cleanup: () => Promise<void> = 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<void> => {
+      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
+}

apps/desktop/electron/main/services/transfers.ts

@@ -135,7 +135,7 @@ async function writeSelectionList(
 
 type BuiltPlan = { plan: ExecutionPlan; cleanup: () => Promise<void> }
 
-async function buildPlan(request: TransferRequest, overrides: Partial<RsyncOptions> = {}): Promise<BuiltPlan> {
+export async function buildPlan(request: TransferRequest, overrides: Partial<RsyncOptions> = {}): Promise<BuiltPlan> {
   const source = await resolveEndpoint(request.source)
   const destination = await resolveEndpoint(request.destination)
   const capabilities = await capabilitiesFor([source.connectionId, destination.connectionId])

apps/desktop/electron/preload/index.ts

@@ -66,6 +66,8 @@ const api = {
     cancel: (runId: string) => call<boolean>(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<boolean>(IPC.fleetSyncCancel, { syncId }),
     runs: (limit = 25) => call(IPC.fleetRuns, { limit }),
     saveCommand: (input: unknown) => call(IPC.fleetCommandSave, input),
     removeCommand: (name: string) => call<boolean>(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)

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<typeof FleetSyncRequestSchema>
+
+export const FleetSyncIdSchema = z.string().min(1).max(128)
+
 /**
  * Plugins.
  *

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<SyncProfile[]>([])
   const [outsideShell, setOutsideShell] = useState(false)
 
@@ -573,6 +575,10 @@ export default function Workspace() {
             <Server className="size-3.5" />
             Fleet
           </TabButton>
+          <TabButton active={tab === 'sync'} onClick={() => setTab('sync')}>
+            <HardDriveDownload className="size-3.5" />
+            Fleet Sync
+          </TabButton>
         </nav>
 
         <div className="ml-auto flex items-center gap-2">
@@ -748,6 +754,10 @@ export default function Workspace() {
         <FleetView onAddServer={() => setShowServers(true)} />
       </div>
 
+      <div className={`flex min-h-0 flex-1 flex-col ${tab === 'sync' ? '' : 'hidden'}`}>
+        <FleetSyncPanel />
+      </div>
+
       <ServerManager
         open={showServers}
         onClose={() => setShowServers(false)}