development.md

@@ -71,13 +71,11 @@ Renderer never touches Node/Electron directly — everything goes through the `w
 
 - Two sides: the receiver lives in lansync.js, the sender in lansyncpush.js, built by createPusher(helpers) — several launcher states share the module, so nothing may sit in its module scope. Both sides report their own progress (receiver through the onReceive hook).
 - An incoming sync needs CONSENT: a correct code opens the gate, but the receiver's onIncoming hook (registered in main.js) shows Accept/Decline, answered through lansync:respondIncoming. Unanswered requests expire (INCOMING_TIMEOUT_MS); a decline returns 'rejected' and KEEPS the pairing (the code was proven). Without an onIncoming hook (tests, headless) the gate stays open.
-- Progress is bytes plus per-category counters (items = announced units, itemsDone = finished, totals = everything picked) — "Apps: 2/5". The bar follows the CURRENT PHASE's bytes (barBytes/barTotalBytes) so a phase starts from an empty bar. Hashing and copying walk tens of thousands of files: every per-file update goes through progressGate (~10/s, phase changes forced) — never one IPC message per file. The gate holds a snapshot FUNCTION on the sender side: the snapshot is built when it flushes, so a throttled update never shows a byte count one gate window old.
+- Progress is bytes plus per-category counters (items = announced units, itemsDone = finished, totals = everything picked) — "Apps: 2/5". The bar follows the CURRENT PHASE's bytes (barBytes/barTotalBytes) so a phase starts from an empty bar. Hashing and copying walk tens of thousands of files: every per-file update goes through progressGate (~10/s, phase changes forced) — never one IPC message per file.
-- The sender's transferred counter counts bytes AS THEY ARE READ (putFile's onBytes callback), not on upload completion: a multi-GB file has to move it while it flies or the sender shows hundreds of MB behind the receiver. A unit's byte total travels as the `{ unitBytes }` control record BEFORE its files; the receiver has to handle it before the entry filter (it has no path/type, so a filter-first order dropped it and the comparing bar measured against a placeholder total of 1 — read as 100% for the whole phase). The comparing phase ends on a full bar, then the copy bar restarts empty (its first PUT byte emits it).
-- While the receiver answers the consent popup (and whenever its side waits on the other one), both popups show "Waiting for the other device…" (lansync.waiting): the sender emits a 'wait' phase before verify, the receiver's ping handler emits stage 'wait'. A unit counts toward "Apps: 1/5" as soon as its comparison starts (not when it commits), and phases without a file count (hash/compare/commit) clear the counter so a stale "3 / 10 files" never lingers. On the wire the receiver's comparing phase is called 'compare' (sender-side 'plan' is unchanged).
 - Registry-level settings are opt-in (the "settings" category, old name "defaults" still accepted as an alias on both sides). The payload carries defaults, wineSettings (displayName + disabled per build), prefixSettings, appSettings (name, wine, prefix, args, env, dllOverrides, disableUmu per app) and appOrder. The receiver only patches what already exists (unknown ids skipped, nothing created or deleted, unknown local entries kept at the end of the order) through the normal helpers (setWineDisplayName, library.updateGame, reorderGames).
 - File hashes are remembered in files/data/lansync-hashes.json (path -> size + mtime + hash; machine-local, never synced): equal size and mtime means equal bytes, so a repeat sync only stats the tree. Hashing and the receiver's comparison both run HASH_CONCURRENCY files at a time.
 - File permissions travel with the manifest (`mode`) and are applied after the bytes; a file with the right hash but wrong mode is repaired without re-upload.
-- A sender hashing a big unit is minutes of silence: it pings `/plwl/ping` every HEARTBEAT_MS, the receiver re-arms its popup watchdog (stage 'wait' shows "Waiting for the other device…" but never while a file is being written). A receiving popup its watchdog closed reopens on the next real event — only 'done' ends a sync, and reopening never asks for consent again (consent is the onIncoming popup at verify time; the progress popup is driven by the same token's events).
+- A sender hashing a big unit is minutes of silence: it pings `/plwl/ping` every HEARTBEAT_MS, the receiver re-arms its popup watchdog (stage 'wait' changes nothing on screen). A receiving popup its watchdog closed reopens on the next real event — only 'done' ends a sync, and reopening never asks for consent again (consent is the onIncoming popup at verify time; the progress popup is driven by the same token's events).
 - A sync must never walk through a symlinked ROOT: pruneTree/healUnitRoots (sync), removeTree/treeSize (fsutil) and healBuild (wines) lstat the root first. `listDirs()` counts symlinked folders as units, so a build on another disk still travels — as the link it is (`wineUnits` maps nested builds to their TOP folder under wines/).
 - Wine/Proton rewrite prefix links to absolute targets whenever they update a prefix, so on-disk links are not always relative after a run: the sync relativizes them on the wire (portableLinkTarget) and refreshWines()/healPrefix() repair what arrived from an older PLWL. A build whose binaries lost the executable bit is detected in refreshWines() and healed once (spawn() on a non-executable `proton` fails with EACCES while the game looks started).
 
@@ -101,7 +99,6 @@ node --check <changed js files>          # syntax
 node test/smoke.js                       # quick sanity
 node test/lansync.test.js                # LAN sync contract (identity, pairing, mirror)
 node test/lansync-receiving.test.js      # receive-side progress events (the popup)
-node test/lansync-progress.test.js       # live sender bytes, comparing bar widths
 node test/lansync-settings.test.js       # playtime, app settings, wine/prefix + defaults
 node test/lansync-links.test.js          # wire: exec bits + relative links
 node test/lansync-pending-links.test.js  # links to files created later, startup heal

files/translations/en-us.json

@@ -205,10 +205,8 @@
     "lansync.incomingMirror": "The other device is the source of truth: files and items on this device that do not exist there will be deleted.",
     "lansync.start": "Start sync",
     "lansync.progress": "Syncing to {0}…",
-    "lansync.waiting": "Waiting for the other device…",
     "lansync.phase.scan": "Preparing {0}",
     "lansync.phase.hash": "Hashing {0}",
-    "lansync.phase.compare": "Comparing {0}",
     "lansync.phase.plan": "Comparing {0}",
     "lansync.phase.copy": "Sending {0}",
     "lansync.recvPhase.copy": "Receiving {0}",
@@ -225,13 +223,14 @@
     "lansync.receiving": "Receiving from {0}…",
     "lansync.incomingTitle": "Incoming sync from {0}",
     "lansync.incomingText": "{0} wants to sync its apps, Wine versions and prefixes onto this device.",
-    "lansync.incomingSettings": "launcher settings only (no files)",
     "lansync.incomingNothing": "Nothing was selected, or the totals have not arrived yet.",
     "lansync.incomingTimedOut": "The incoming sync request timed out.",
     "lansync.accept": "Accept",
     "lansync.decline": "Decline",
     "lansync.rejected": "{0} declined the sync.",
     "lansync.unknownDevice": "Unknown device",
+    "lansync.recvWaiting": "Waiting for the file list…",
+    "lansync.recvStalled": "Waiting for the other device…",
     "lansync.recvIdle": "The other device stopped responding, the sync was interrupted.",
     "lansync.done": "Synced to {0}: {1} sent ({4}), {2} already up to date, {3} removed",
     "lansync.failed": "Sync failed: {0}",

files/translations/pl.json

@@ -205,10 +205,8 @@
     "lansync.incomingMirror": "Drugie urządzenie jest źródłem prawdy: pliki i elementy na tym urządzeniu, których tam nie ma, zostaną usunięte.",
     "lansync.start": "Rozpocznij synchronizację",
     "lansync.progress": "Synchronizowanie na {0}…",
-    "lansync.waiting": "Oczekiwanie na drugie urządzenie…",
     "lansync.phase.scan": "Przygotowywanie {0}",
     "lansync.phase.hash": "Haszowanie {0}",
-    "lansync.phase.compare": "Porównywanie {0}",
     "lansync.phase.plan": "Porównywanie {0}",
     "lansync.phase.copy": "Wysyłanie {0}",
     "lansync.recvPhase.copy": "Odbieranie {0}",
@@ -225,13 +223,14 @@
     "lansync.receiving": "Odbieranie od {0}…",
     "lansync.incomingTitle": "Przychodząca synchronizacja od {0}",
     "lansync.incomingText": "{0} chce przysłać swoje aplikacje, wersje Wine i prefiksy na to urządzenie.",
-    "lansync.incomingSettings": "tylko ustawienia launchera (bez plików)",
     "lansync.incomingNothing": "Nic nie zaznaczono albo podsumowanie jeszcze nie dotarło.",
     "lansync.incomingTimedOut": "Upłynął czas na odpowiedź na przychodzącą synchronizację.",
     "lansync.accept": "Akceptuj",
     "lansync.decline": "Odrzuć",
     "lansync.rejected": "{0} odrzucił synchronizację.",
     "lansync.unknownDevice": "Nieznane urządzenie",
+    "lansync.recvWaiting": "Oczekiwanie na listę plików…",
+    "lansync.recvStalled": "Oczekiwanie na drugie urządzenie…",
     "lansync.recvIdle": "Drugie urządzenie przestało odpowiadać, synchronizacja została przerwana.",
     "lansync.done": "Zsynchronizowano na {0}: wysłano {1} ({4}), aktualnych {2}, usunięto {3}",
     "lansync.failed": "Synchronizacja nie powiodła się: {0}",

renderer/app.js

@@ -1405,11 +1405,8 @@ function renderProgressBar(bar, bytes, totalBytes) {
 
 // Older peers only send the overall byte totals - then those drive the bar.
 function barValues(info) {
-  let bytes = Math.max(0, Number(info.barBytes == null ? info.bytes : info.barBytes) || 0)
+  const bytes = info.barBytes == null ? info.bytes : info.barBytes
-  let total = Math.max(0, Number(info.barTotalBytes == null ? info.totalBytes : info.barTotalBytes) || 0)
+  const total = info.barTotalBytes == null ? info.totalBytes : info.barTotalBytes
-  // a provisional total can be smaller than what already moved (it grows as
-  // entries arrive): the bar reads full, never over 100%
-  if (total > 0 && bytes > total) bytes = total
   return [bytes, total]
 }
 
@@ -1572,13 +1569,10 @@ async function runLanSync(device, code, categories, force) {
   state.lanPushing = true
   window.plwl.onLanSyncProgress(data => {
     if (!active) return
-    // waiting phases have no translation under lansync.phase.*
+    // the file counter is cumulative across units; only the bar follows phases
-    const phaseKey = data.phase === 'wait' ? 'lansync.waiting' : 'lansync.phase.' + data.phase
+    phase.textContent = t('lansync.phase.' + data.phase, data.label || '').trim()
-    phase.textContent = t(phaseKey, data.label || '').trim()
     renderProgressBar(bar, ...barValues(data))
-    // phases without a file count clear it: a stale "5 / 10" must not linger
     if (data.total > 0) count.textContent = t('lansync.progressFiles', data.done, data.total)
-    else count.textContent = ''
     stats.textContent = lanStatsLine(data.items, data.itemsDone, data.totals)
     bytes.textContent = lanBytesLine(data.bytes, data.totalBytes)
   })
@@ -1641,8 +1635,6 @@ function onLanIncoming(info) {
   if (totals.apps > 0) parts.push(t('lansync.statsApps', totals.apps))
   if (totals.wines > 0) parts.push(t('lansync.statsWines', totals.wines))
   if (totals.prefixes > 0) parts.push(t('lansync.statsPrefixes', totals.prefixes))
-  // a settings-only sync moves no units, but it is still a real request
-  if (!parts.length && totals.settings > 0) parts.push(t('lansync.incomingSettings'))
   const modal = showModal(`
     <h2 id="ls-incoming-title"></h2>
     <p id="ls-incoming-text"></p>
@@ -1732,9 +1724,7 @@ function armReceiveWatchdog() {
   receiveStall = setTimeout(() => {
     receiveStall = null
     const parts = receiveModalParts()
-    if (!parts) return
+    if (parts) parts.phase.textContent = t('lansync.recvStalled')
-    // silence also means no bytes: freeze the bar and ask for patience
-    parts.phase.textContent = t('lansync.waiting')
   }, RECEIVE_STALL_MS)
 }
 
@@ -1769,18 +1759,16 @@ function onLanReceiving(info) {
 function renderReceiving(info) {
   const parts = receiveModalParts()
   if (!parts) return
-  // a ping keeps the bar and counters as they are; only the phase line moves
+  // a ping keeps the screen as it is; only a reopened popup needs a phase line
   if (info.stage === 'wait') {
-    parts.phase.textContent = t('lansync.waiting')
+    if (!parts.phase.textContent) parts.phase.textContent = t('lansync.recvWaiting')
     return
   }
   // the same phases, worded for receiving
   const phaseKey = RECV_PHASE_KEYS[info.stage] || 'lansync.phase.' + info.stage
-  parts.phase.textContent = (info.stage === 'start' ? t('lansync.waiting') : t(phaseKey, info.label || '')).trim()
+  parts.phase.textContent = (info.stage === 'start' ? t('lansync.recvWaiting') : t(phaseKey, info.label || '')).trim()
   renderProgressBar(parts.bar, ...barValues(info))
-  // the file counter belongs to the copy/done phases; receive events for other
+  if (info.total > 0) parts.count.textContent = t('lansync.progressFiles', info.done, info.total)
-  // phases carry cumulative numbers that would look stale mid-phase
-  if ((info.stage === 'copy' || info.stage === 'done') && info.total > 0) parts.count.textContent = t('lansync.progressFiles', info.done, info.total)
   else parts.count.textContent = ''
   parts.stats.textContent = lanStatsLine(info.items, info.itemsDone, info.totals)
   parts.bytes.textContent = lanBytesLine(info.bytes, info.totalBytes)

src/lansync-manifest.js

@@ -22,9 +22,7 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
 
   // Walk without following symlinks (dosdevices etc. are part of the sync).
   // Iterative, with one directory's stat/readlink calls in flight together.
-  // `emit` (optional) streams dirs and links out while the walk continues;
+  async function scanTree(abs, rel, out, mtimes) {
-  // file entries only land in `out` (their digest comes later).
-  async function scanTree(abs, rel, out, mtimes, emit) {
     const queue = [[abs, rel]]
     while (queue.length) {
       const [dirAbs, dirRel] = queue.pop()
@@ -39,9 +37,7 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
       for (const entry of entries) {
         const relChild = `${dirRel}/${entry.name}`
         if (entry.isDirectory()) {
-          const record = { path: relChild, type: 'dir' }
+          out.push({ path: relChild, type: 'dir' })
-          out.push(record)
-          if (emit) await emit(record)
           dirs.push([path.join(dirAbs, entry.name), relChild])
         } else {
           leaves.push(entry)
@@ -55,9 +51,7 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
           try {
             target = await fsp.readlink(absChild)
           } catch {}
-          const record = { path: relChild, type: 'link', target: portableLinkTarget(absChild, target) }
+          out.push({ path: relChild, type: 'link', target: portableLinkTarget(absChild, target) })
-          out.push(record)
-          if (emit) await emit(record)
           return
         }
         let st
@@ -76,14 +70,8 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
     }
   }
 
-  // Manifest of one unit, streamed: `handlers` may carry onEntry (a Promise
+  // Manifest of one unit; `force` skips hashing (receiver replaces everything).
-  // per entry, awaited - backpressure), onTotal and onProgress. File entries
+  async function buildManifest(roots, force, onProgress) {
-  // go out as their digests finish, so the receiver compares while this side
-  // is still hashing; hashing itself stays parallel, a batch at a time.
-  async function buildManifestStream(roots, force, handlers) {
-    const onEntry = handlers && handlers.onEntry
-    const onTotal = handlers && handlers.onTotal
-    const onProgress = handlers && handlers.onProgress
     const entries = []
     const mtimes = new Map()
     for (const rel of roots) {
@@ -100,32 +88,22 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
         try {
           target = await fsp.readlink(abs)
         } catch {}
-        const record = { path: rel, type: 'link', target: portableLinkTarget(abs, target) }
+        entries.push({ path: rel, type: 'link', target: portableLinkTarget(abs, target) })
-        entries.push(record)
-        if (onEntry) await onEntry(record)
       } else if (st.isDirectory()) {
-        const record = { path: rel, type: 'dir' }
+        entries.push({ path: rel, type: 'dir' })
-        entries.push(record)
+        await scanTree(abs, rel, entries, mtimes)
-        if (onEntry) await onEntry(record)
-        await scanTree(abs, rel, entries, mtimes, onEntry)
       } else if (st.isFile()) {
         entries.push({ path: rel, type: 'file', size: st.size, mode: st.mode & 0o777, hash: null })
         mtimes.set(rel, st.mtimeMs || 0)
       }
     }
     const files = entries.filter(e => e.type === 'file')
-    // total size is known before hashing: the receiver sizes its compare bar
+    // total size is known before hashing, so the bar starts empty
-    // from this, the sender's popup from onProgress
     const totalBytes = files.reduce((sum, entry) => sum + (entry.size || 0), 0)
-    if (onTotal) await onTotal(totalBytes)
+    if (!force && files.length) {
       let done = 0
       let hashed = 0
-    // one batch hashes with the whole disk queue behind it; its entries go
+      await mapLimit(files, concurrency, async entry => {
-    // on the wire in order before the next batch starts
-    for (let i = 0; i < files.length; i += concurrency) {
-      const batch = files.slice(i, i + concurrency)
-      if (!force) {
-        await Promise.all(batch.map(async entry => {
         const abs = filesPath(entry.path)
         try {
           // unchanged size + mtime: the cached digest still holds
@@ -139,30 +117,16 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
         } catch {
           entry.hash = null
         }
-        }))
-      }
-      for (const entry of batch) {
         done++
         if (!entry.hash) hashed += entry.size || 0 // nothing read: still report the file as passed over
         if (onProgress) onProgress(done, files.length, hashed, totalBytes)
-        if (onEntry) await onEntry(entry)
+      })
-      }
     }
     hash.save()
     return entries
   }
 
-  // Buffering variant for callers that want the whole manifest first.
+  return { buildManifest, scanTree, portableLinkTarget }
-  async function buildManifest(roots, force, onProgress) {
-    const entries = []
-    await buildManifestStream(roots, force, {
-      onProgress,
-      onEntry: entry => { entries.push(entry) }
-    })
-    return entries
-  }
-
-  return { buildManifest, buildManifestStream, scanTree, portableLinkTarget }
 }
 
 module.exports = { createManifestBuilder }

src/lansync.js

@@ -240,23 +240,14 @@ function readJsonBody(req, limit = 16 * 1024 * 1024) {
 function readNdjson(req, onEntry) {
   return new Promise(resolve => {
     const rl = readline.createInterface({ input: req, crlfDelay: Infinity })
-    let chain = Promise.resolve()
     rl.on('line', line => {
       if (!line.trim()) return
-      // entries are handled strictly in order: a handler may be async (the
-      // streamed plan compares/hashes while the body keeps arriving), but
-      // the response must never reorder the manifest
-      let parsed = null
       try {
-        parsed = JSON.parse(line)
+        onEntry(JSON.parse(line))
-      } catch {
+      } catch {}
-        return
-      }
-      chain = chain.then(() => onEntry(parsed)).catch(() => {})
     })
-    const done = () => chain.then(resolve, () => resolve())
+    rl.on('close', resolve)
-    rl.on('close', done)
+    rl.on('error', resolve)
-    rl.on('error', done)
   })
 }
 
@@ -810,7 +801,7 @@ async function handleVerify(state, req, res, hooks) {
             requestId,
             deviceId: from,
             name: String(payload.name || ''),
-            totals: { apps: Number(announced.apps) || 0, wines: Number(announced.wines) || 0, prefixes: Number(announced.prefixes) || 0, settings: Number(announced.settings) || 0 }
+            totals: { apps: Number(announced.apps) || 0, wines: Number(announced.wines) || 0, prefixes: Number(announced.prefixes) || 0 }
           })
         } catch {
           resolve('reject') // no window to answer with -> never auto-accept silently
@@ -866,14 +857,13 @@ async function handlePlan(state, session, url, req, res, hooks) {
   } catch {}
   const owned = (Array.isArray(roots) ? roots : []).map(ownedRoot).filter(Boolean)
   if (!key || !owned.length) return sendJson(res, 400, { ok: false, error: 'bad-unit' })
-  // comparing is its own phase: the bar restarts for the copy phase
+  const entries = []
-  const announce = progressGate((data, force) => emitReceive(hooks, session, 'compare', data))
+  await readNdjson(req, entry => {
-  // the unit counts from the start of its comparison, not from its commit
+    if (inlineSame(entry)) entries.push(entry)
-  const kind = UNIT_KINDS[String(key).split(':')[0]]
+  })
-  if (kind && !session.counted.has(key)) {
+  // the mode each file has to end up with on this side
-    session.counted.add(key)
+  const modes = new Map()
-    session.items[kind] = (session.items[kind] || 0) + 1
+  for (const entry of entries) if (entry.type === 'file' && typeof entry.mode === 'number') modes.set(entry.path, entry.mode)
-  }
   // a force sync compares nothing: every entry counts as different
   const forced = url.searchParams.get('force') === '1'
   // the sender's totals, so the popup can say "2/5" from the start
@@ -887,65 +877,24 @@ async function handlePlan(state, session, url, req, res, hooks) {
       }
     }
   } catch {}
-  // The manifest is compared as it streams in (both sides hash at the same
+  const allowed = entries.filter(entry => withinRoots(entry.path, owned))
-  // time). The sender sends its unit total first, so the bar starts with its
+  // comparing is its own phase: the bar restarts for the copy phase
-  // full width; files missing from the sender's total compare as the usual
+  const unitTotalBytes = allowed.reduce((sum, entry) => sum + (entry.size || 0), 0)
-  // entries - the bar just grows when that happens.
+  const announce = progressGate((data, force) => emitReceive(hooks, session, 'plan', data))
-  const entries = []
-  const results = new Map()
-  let unitTotalBytes = 0
   let compared = 0
   let comparedBytes = 0
-  const compareOne = async entry => {
+  session.barTotal = unitTotalBytes
+  session.barBytes = 0
+  const identical = await mapLimit(allowed, HASH_CONCURRENCY, async entry => {
     const same = forced ? false : await entrySame(entry)
-    results.set(entry.path, same === true)
     compared++
     comparedBytes += entry.size || 0
     session.barBytes = comparedBytes
-    // the sender's announced total (or a provisional full bar until it
-    // arrives) - clamped at the top by the renderer, never over 100%
-    session.barTotal = unitTotalBytes || 1
     // no file counting yet: dirs/links are compared too
     announce({ unit: label })
-  }
+    return same === true
-  // files compared concurrently: the sender streams its manifest, this side
-  // keeps at most HASH_CONCURRENCY digests in flight (the sender's hashing
-  // queue is the wider throttle)
-  const jobs = []
-  let inFlight = 0
-  const settleOldest = () => Promise.race(jobs.splice(0, 1))
-  await readNdjson(req, async entry => {
-    // the sender announces its unit total before hashing starts (a control
-    // record, no path): it carries the compare bar's width, so it must be
-    // read before the entry filter drops it
-    if (entry.unitBytes != null) {
-      unitTotalBytes = Number(entry.unitBytes) || 0
-      session.barTotal = unitTotalBytes || 1
-      return
-    }
-    if (!inlineSame(entry)) return
-    entries.push(entry)
-    if (!withinRoots(entry.path, owned)) return
-    if (forced) {
-      await compareOne(entry)
-      return
-    }
-    // the sender's hashing queue is the throttle on its side; this side's
-    // own concurrency limit keeps a saturated disk from stacking up here
-    if (inFlight >= HASH_CONCURRENCY) await settleOldest()
-    inFlight++
-    const job = compareOne(entry).finally(() => {
-      inFlight--
   })
-    jobs.push(job)
-  })
-  await Promise.all(jobs)
   hashStore.save()
-  const allowed = entries.filter(entry => withinRoots(entry.path, owned))
-  const identical = allowed.map(entry => results.get(entry.path) === true)
-  // the mode each file has to end up with on this side
-  const modes = new Map()
-  for (const entry of allowed) if (entry.type === 'file' && typeof entry.mode === 'number') modes.set(entry.path, entry.mode)
   const paths = new Set()
   const upload = []
   const same = []
@@ -969,24 +918,20 @@ async function handlePlan(state, session, url, req, res, hooks) {
   // this unit's files/bytes, reported as they arrive
   session.planned += upload.length
   session.plannedBytes += uploadBytes
-  // the compare is finished: report it full, then reset for the copy phase
+  // the bar restarts for the copy phase
-  // (the first byte that arrives emits the new bar)
-  session.barTotal = unitTotalBytes || session.barTotal || 1
-  session.barBytes = session.barTotal
-  announce({ unit: label }, true)
   session.unitWritten = 0
-  session.unitLabel = label // named up front, so file events can label it
-  session.barBytes = 0
   session.barTotal = uploadBytes
-  sendJson(res, 200, { ok: true, upload, same, entries: paths.size })
+  session.barBytes = 0
+  const kind = UNIT_KINDS[String(key).split(':')[0]]
+  if (kind && !session.counted.has(key)) {
+    session.counted.add(key)
+    session.items[kind] = (session.items[kind] || 0) + 1
+  }
+  // the unit's label is announced up front, so file events can name it
+  session.unitLabel = label
+  sendJson(res, 200, { ok: true, upload, same, entries: paths.size })
+  announce({ unit: label }, true)
 }
-
-// One PUT, one file. The sender declares a content-length up front; if the
-// body stops short of it (sender crashed, lost link, file changed size after
-// the manifest), the request would otherwise stay open forever and wedge
-// this file, the session and the sync. Any arriving byte re-arms the timer,
-// so a slow honest transfer never trips it. Env override for tests.
-const PUT_IDLE_MS = Number(process.env.PLWL_LANSYNC_PUT_IDLE) || 60000
 
 async function handlePut(session, url, req, res, hooks) {
   // unit and path travel as headers: constant per request
@@ -1000,22 +945,8 @@ async function handlePut(session, url, req, res, hooks) {
   }
   // report bytes while the file is still arriving
   const announce = progressGate((data, force) => emitReceive(hooks, session, 'copy', data))
-  // a ping must not flash "waiting" while transfers are running
-  session.writing = (session.writing || 0) + 1
-  let idleTimer = null
-  const armIdle = () => {
-    if (idleTimer) clearTimeout(idleTimer)
-    idleTimer = setTimeout(() => {
-      // the write's stream 'error' handler removes the tmp file; the sender's
-      // request errors out and its own timer/heartbeat handles the rest
-      req.destroy(new Error('body-stalled'))
-    }, PUT_IDLE_MS)
-    if (idleTimer.unref) idleTimer.unref()
-  }
-  armIdle()
   try {
     await writeStream(abs, req, delta => {
-      armIdle()
       session.bytes = (session.bytes || 0) + delta
       session.unitWritten = (session.unitWritten || 0) + delta
       session.barBytes = session.unitWritten
@@ -1028,9 +959,6 @@ async function handlePut(session, url, req, res, hooks) {
     announce({ unit: session.unitLabel || '' }, true)
   } catch (err) {
     sendJson(res, 500, { ok: false, error: String((err && err.message) || err) })
-  } finally {
-    if (idleTimer) clearTimeout(idleTimer)
-    session.writing--
   }
 }
 
@@ -1098,20 +1026,14 @@ async function handleApply(session, req, res, hooks) {
 function handleEnd(state, session, res, hooks) {
   state.tokens.delete(session.token)
   if (state.session && state.session.token === session.token) state.session = null
-  // a finished sync leaves a finished bar, even one that moved no files
-  session.barTotal = session.barTotal || 1
-  session.barBytes = session.barTotal
   sendJson(res, 200, { ok: true })
   emitReceive(hooks, session, 'done', { deleted: session.deleted || 0 })
 }
 
-// Liveness ping: hashing can take minutes; re-arms the popup watchdog. The
+// Liveness ping: hashing can take minutes; re-arms the popup watchdog.
-// event must repeat on every ping (it is what keeps the popup's watchdog
-// armed), but never while a file is being written: a copy in flight reports
-// itself and must not flash the waiting text over the receiving phase.
 function handlePing(session, res, hooks) {
   session.lastAt = Date.now()
-  if (!session.writing) emitReceive(hooks, session, 'wait')
+  emitReceive(hooks, session, 'wait')
   sendJson(res, 200, { ok: true })
 }
 
@@ -1218,7 +1140,6 @@ function getPusher() {
       filesPath,
       ownedRoot,
       buildManifest,
-      buildManifestStream: manifest.buildManifestStream,
       listDirs,
       listDevices,
       mapLimit,
@@ -1298,7 +1219,6 @@ const exported = {
   filesPath,
   ownedRoot,
   buildManifest,
-  buildManifestStream: manifest.buildManifestStream,
   mapLimit,
   listDirs,
   syncTo

src/lansyncpush.js

@@ -1,9 +1,8 @@
 'use strict'
 
-// The pushing side of a LAN sync: per unit, stream a manifest while it is
+// The pushing side of a LAN sync: per unit, send a manifest, learn what the
-// still being hashed, learn what the receiver has, upload the rest, delete
+// receiver has, upload the rest, delete the surplus. Exported as a factory;
-// the surplus. Exported as a factory; src/lansync.js hands in its own
+// src/lansync.js hands in its own helpers and state.
-// helpers and state.
 
 const http = require('http')
 const os = require('os')
@@ -13,22 +12,10 @@ const path = require('path')
 // Files sent at once: one at a time wastes LAN round trips on small files.
 const UPLOAD_CONCURRENCY = 4
 
-// Two lost heartbeats in a row: the receiver is gone (40 s of silence).
-const MAX_MISSED_PINGS = 2
-
-// One file, one PUT. A file that has not finished on its own after this long
-// (stalled receiver, lost socket while the OS still reports it open, a file
-// that changed size after the manifest) fails the whole sync instead of
-// hanging it forever. Env override for tests.
-const UPLOAD_TIMEOUT_MS = Number(process.env.PLWL_LANSYNC_UPLOAD_TIMEOUT) || 10 * 60 * 1000
-
 // Keeps the receiver alive while this side hashes (minutes of silence); env
 // override for tests.
 const HEARTBEAT_MS = Number(process.env.PLWL_LANSYNC_HEARTBEAT) || 20000
 
-// One ping round trip; env override for tests.
-const PING_TIMEOUT_MS = Number(process.env.PLWL_LANSYNC_PING_TIMEOUT) || 10000
-
 function createPusher(api) {
   const {
     CATEGORIES,
@@ -39,7 +26,7 @@ function createPusher(api) {
     httpJson,
     filesPath,
     ownedRoot,
-    buildManifestStream,
+    buildManifest,
     listDirs,
     DIRS,
     toPortablePath,
@@ -48,10 +35,8 @@ function createPusher(api) {
     prefixes
   } = api
 
-  // Manifest in, upload list out; NDJSON avoids one giant string. The body
+  // Manifest in, upload list out; NDJSON avoids one giant string.
-  // is written by `produce` WHILE the sender hashes, so the receiver can
+  function planUnit(device, token, unit, entries, force, meta) {
-  // compare entries as they arrive (both sides hash at the same time).
-  function planUnit(device, token, unit, force, meta, produce) {
     // announced up front, so the receiver's popup can say "Apps: 2/5"
     const totals = meta && meta.totals ? `&totals=${encodeURIComponent(JSON.stringify(meta.totals))}` : ''
     return new Promise((resolve, reject) => {
@@ -77,47 +62,23 @@ function createPusher(api) {
           })
         }
       )
-      let dead = null
+      req.on('error', reject)
-      req.on('error', err => {
+      let index = 0
-        dead = err
+      const pump = () => {
-        reject(err)
+        while (index < entries.length) {
-      })
+          if (!req.write(`${JSON.stringify(entries[index++])}\n`)) {
-      // ordered writes with backpressure: the receiver reads continuously,
+            req.once('drain', pump)
-      // but a slow socket must not buffer the whole manifest in memory
+            return
-      const write = obj => new Promise((resolveWrite, rejectWrite) => {
-        if (dead) return rejectWrite(dead)
-        const onDrain = () => {
-          cleanup()
-          resolveWrite()
           }
-        const onErr = () => {
-          cleanup()
-          rejectWrite(dead || new Error('plan-write-failed'))
         }
-        const cleanup = () => {
+        req.end()
-          req.off('drain', onDrain)
-          req.off('error', onErr)
       }
-        if (req.write(`${JSON.stringify(obj)}\n`)) {
+      pump()
-          cleanup()
-          return resolveWrite()
-        }
-        req.once('drain', onDrain)
-        req.once('error', onErr)
-      })
-      produce(write).then(() => req.end(), err => {
-        try { req.destroy() } catch {}
-        reject(err)
-      })
     })
   }
 
-  function putFile(device, token, unitKey, rel, abs, onBytes) {
+  function putFile(device, token, unitKey, rel, abs) {
     return new Promise((resolve, reject) => {
-      const settle = (fn, value) => {
-        clearTimeout(timer)
-        fn(value)
-      }
       // unit and path travel as headers: constant per request
       const req = http.request(
         {
@@ -135,29 +96,12 @@ function createPusher(api) {
         },
         res => {
           res.resume()
-          res.on('end', () => settle(resolve, { status: res.statusCode }))
+          res.on('end', () => resolve({ status: res.statusCode }))
-          res.on('aborted', () => settle(reject, new Error('response-aborted')))
         }
       )
-      // The timer spans the WHOLE transfer, response included: a stalled
+      req.on('error', reject)
-      // receiver or a silently lost socket must fail the sync instead of
-      // hanging the popup on the last file. Generous - a file would have to
-      // take over ten minutes at 20 MB/s to hit it.
-      const timer = setTimeout(() => {
-        req.destroy(new Error('upload-timeout'))
-        settle(reject, new Error('upload-timeout'))
-      }, UPLOAD_TIMEOUT_MS)
-      if (timer.unref) timer.unref()
-      req.on('error', err => settle(reject, err))
       const stream = fs.createReadStream(abs)
-      stream.on('error', err => settle(reject, err))
+      stream.on('error', reject)
-      // live bytes: a multi-GB file has to move the counter while it flies,
-      // not only when its response lands (read = what this side hands over)
-      if (onBytes) stream.on('data', chunk => onBytes(chunk.length))
-      // pipe ends the request when the file ends; a body shorter than the
-      // declared content-length (file changed mid-sync) wedges the receiver
-      // - its own idle watchdog breaks that, this side's timer breaks every
-      // other silent case
       stream.pipe(req)
     })
   }
@@ -321,9 +265,8 @@ function createPusher(api) {
     // units are known before anything is read: announce them with the verify
     const units = buildUnits(cats)
     const bucketOf = kind => (kind === 'wine' ? 'wines' : kind === 'prefix' ? 'prefixes' : 'apps')
-    const totals = { apps: 0, wines: 0, prefixes: 0, settings: 0 }
+    const totals = { apps: 0, wines: 0, prefixes: 0 }
     for (const unit of units) totals[bucketOf(unit.kind)] += 1
-    if (cats.has('settings')) totals.settings = 1
     // every progress event is kept, so callers can report without subscribing
     const events = []
     const progress = data => {
@@ -331,6 +274,32 @@ function createPusher(api) {
       if (onProgress) onProgress(data)
     }
 
+    let verified
+    try {
+      // longer than the receiver's incoming-request timeout (popup wait)
+      verified = await httpJson(device, 'POST', '/plwl/verify', {
+        json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase(), totals },
+        timeout: 90000
+      })
+    } catch (err) {
+      return fail(String((err && err.message) || err))
+    }
+    if (!verified.data || !verified.data.ok) {
+      const error = (verified.data && verified.data.error) || `http-${verified.status}`
+      // a rotated code drops the pairing; a plain 'rejected' keeps it
+      if (error === 'auth-failed' || error === 'bad-code') forgetDevice(device.id || deviceId)
+      return fail(error)
+    }
+    const token = verified.data.token
+    // paired now: the next sync needs no typing
+    rememberDevice(device.id || deviceId, code, device.name)
+
+    // hashing precedes any byte: keep the receiver's popup alive meanwhile
+    const heartbeat = setInterval(() => {
+      httpJson(device, 'POST', `/plwl/ping?token=${encodeURIComponent(token)}`, { json: {}, timeout: 10000 }).catch(() => {})
+    }, HEARTBEAT_MS)
+    if (heartbeat.unref) heartbeat.unref()
+
     const summary = newSummary(units.length)
     const finished = { apps: 0, wines: 0, prefixes: 0 }
     // planned bytes so far, so the popup can show "1.2 GB / 3.4 GB"
@@ -369,64 +338,16 @@ function createPusher(api) {
       }
       return data
     }
-    // Snapshot per file, forward a few times per second (phase changes
+    // snapshot per file, forward a few times per second (phase changes always)
-    // always). The snapshot is built when the gate flushes, not when it is
+    const gate = typeof api.progressGate === 'function' ? api.progressGate(progress) : data => progress(data)
-    // pushed: a throttled byte count must not be one gate window old.
+    const emit = (phase, unit, done, total, force) => gate(snapshot(phase, unit, done, total), !!force)
-    const emitNow = data => progress(typeof data === 'function' ? data() : data)
-    const gate = typeof api.progressGate === 'function' ? api.progressGate(emitNow) : emitNow
-    const emit = (phase, unit, done, total, force) => gate(() => snapshot(phase, unit, done, total), !!force)
-    // consent may wait on a human: say so before anything else happens
-    emit('wait', null, null, null, true)
-
-    let verified
-    try {
-      // longer than the receiver's incoming-request timeout (popup wait)
-      verified = await httpJson(device, 'POST', '/plwl/verify', {
-        json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase(), totals },
-        timeout: 90000
-      })
-    } catch (err) {
-      return fail(String((err && err.message) || err))
-    }
-    if (!verified.data || !verified.data.ok) {
-      const error = (verified.data && verified.data.error) || `http-${verified.status}`
-      // a rotated code drops the pairing; a plain 'rejected' keeps it
-      if (error === 'auth-failed' || error === 'bad-code') forgetDevice(device.id || deviceId)
-      return fail(error)
-    }
-    const token = verified.data.token
-    // paired now: the next sync needs no typing
-    rememberDevice(device.id || deviceId, code, device.name)
-
-    // The heartbeat keeps the receiver's popup alive while this side hashes
-    // (minutes of silence). Consecutive ping failures mean the peer dropped
-    // (crash, sleep, lost network): without this a dead receiver hangs the
-    // sync until the next request notices, which may be never.
-    let missedPings = 0
-    const abort = { dead: false, reject: null }
-    const abortPromise = new Promise((resolve, reject) => { abort.reject = reject })
-    abortPromise.catch(() => {}) // only awaited via Promise.race below
-    const heartbeat = setInterval(() => {
-      httpJson(device, 'POST', `/plwl/ping?token=${encodeURIComponent(token)}`, { json: {}, timeout: PING_TIMEOUT_MS })
-        .then(() => { missedPings = 0 })
-        .catch(() => {
-          missedPings++
-          if (missedPings >= MAX_MISSED_PINGS && !abort.dead) {
-            abort.dead = true
-            abort.reject(new Error('receiver-lost'))
-          }
-        })
-    }, HEARTBEAT_MS)
-    if (heartbeat.unref) heartbeat.unref()
-
     const end = async () => {
       try {
         await httpJson(device, 'POST', `/plwl/end?token=${encodeURIComponent(token)}`, { json: {} })
       } catch {}
     }
-    const runUnits = async () => {
+    try {
       for (const unit of units) {
-        if (abort.dead) throw new Error('receiver-lost')
         const roots = unit.roots.filter(ownedRoot)
         if (!roots.length) continue
         const kind = bucketOf(unit.kind)
@@ -436,36 +357,14 @@ function createPusher(api) {
         barBytes = 0
         barTotal = 0
         emit('scan', unit, null, null, true)
-        // The manifest streams WHILE this side hashes its tail: the receiver
+        const entries = await buildManifest(roots, !!force, (done, total, hashedBytes, totalBytes) => {
-        // compares entries as they arrive, so both devices hash at the same
-        // time. Once this side is done hashing, the popup keeps a full bar
-        // and the waiting text until the plan answers (a cached sender waits
-        // for a receiver that still has to hash).
-        const entries = []
-        let writes = Promise.resolve()
-        const plan = await planUnit(device, token, { ...unit, roots }, !!force, { totals }, async write => {
-          await buildManifestStream(roots, !!force, {
-            // the receiver sizes its compare bar from this, before any hashing
-            onTotal: totalBytes => {
-              writes = writes.then(() => write({ unitBytes: totalBytes }))
-            },
-            onEntry: entry => {
-              entries.push(entry)
-              writes = writes.then(() => write(entry))
-            },
-            onProgress: (hashedDone, hashedTotal, hashedBytes, totalBytes) => {
           barBytes = hashedBytes
           barTotal = totalBytes
+          // no file count until the plan returns
           emit('hash', unit, null, null)
-            }
         })
-          await writes
+        emit('plan', unit, null, null, true)
-          // this side is done hashing; the receiver may still be comparing
+        const plan = await planUnit(device, token, { ...unit, roots }, entries, !!force, { totals })
-          barTotal = barTotal || 1
-          barBytes = barTotal
-          emit('wait', unit, null, null, true)
-        })
-        if (abort.dead) throw new Error('receiver-lost')
         if (!plan.data || !plan.data.ok) throw new Error((plan.data && plan.data.error) || `http-${plan.status}`)
         const upload = new Set(plan.data.upload || [])
         const same = new Set(plan.data.same || [])
@@ -484,36 +383,32 @@ function createPusher(api) {
           pending.push(entry)
           plannedBytes += entry.size || 0
         }
-        // a few files in flight at once: small files cost round trips; the
+        // a few files in flight at once: small files cost round trips
-        // bar counts only bytes that move and starts empty for the copy
+        // the bar counts only bytes that move, and starts empty for the copy
         if (pending.length) {
           barBytes = 0
           barTotal = pending.reduce((sum, entry) => sum + (entry.size || 0), 0)
           emit('copy', unit, allDone, allTotal + pending.length, true)
+        }
         let failed = null
         let done = 0
         const send = async entry => {
-            if (failed || abort.dead) return
+          if (failed) return
-            const res = await putFile(device, token, unit.key, entry.path, filesPath(entry.path), delta => {
+          const res = await putFile(device, token, unit.key, entry.path, filesPath(entry.path))
-              summary.bytes += delta
-              barBytes += delta
-              emit('copy', unit, allDone + done, allTotal + pending.length)
-            })
           if (res.status !== 200) {
             failed = new Error(`upload-failed-${res.status}`)
             return
           }
           done++
           summary.uploaded++
-            // the byte counters already moved while the file was read; the
+          summary.bytes += entry.size || 0
-            // completion still has to push the file count next
+          barBytes += entry.size || 0
           emit('copy', unit, allDone + done, allTotal + pending.length)
         }
         if (typeof api.mapLimit === 'function') await api.mapLimit(pending, UPLOAD_CONCURRENCY, send)
         else for (const entry of pending) await send(entry)
         if (failed) throw failed
-        }
+        emit('copy', unit, allDone + done, allTotal + pending.length, true)
-        emit('copy', unit, allDone + pending.length, allTotal + pending.length, true)
         // the totals count a unit's files only once they are on the wire
         allDone += pending.length
         allTotal += pending.length
@@ -536,16 +431,9 @@ function createPusher(api) {
         if (deleted) summary[key].deleted += deleted
       }
       if (applied.data.deleted) summary.deleted += applied.data.deleted
-      // a sync with nothing to send still ends with a full bar
-      barTotal = barTotal || 1
-      barBytes = barTotal
       emit('done', null, allDone, allTotal, true)
       await end()
-    }
-    try {
-      await Promise.race([runUnits(), abortPromise])
     } catch (err) {
-      abort.dead = true
       await end()
       return fail(String((err && err.message) || err), summary, events)
     } finally {
@@ -554,9 +442,7 @@ function createPusher(api) {
     return syncResult(true, null, summary, events)
   }
 
-  // putFile is exposed for the stall tests (it is the piece that must never
+  return { syncToDevice, buildUnits, buildRegistryPayload, CATEGORIES }
-  // leave a request half-open)
-  return { syncToDevice, buildUnits, buildRegistryPayload, putFile, CATEGORIES }
 }
 
 module.exports = createPusher