olcxjas-softworks/PLWL
2
0
Fork 0
Code Issues Pull requests Projects Releases 16 Packages Wiki Activity Actions

[TEST] try to improve sync

Browse source
This commit is contained in:
olcxja 2026-10-06 07:28:47 +02:00
commit 0f3fbaa418
8 changed files with 374 additions and 145 deletions
Download patch file Download diff file Expand all files Collapse all files

92
src/lansync-manifest.js
View file

@ -22,7 +22,9 @@ 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.
async function scanTree(abs, rel, out, mtimes) {
// `emit` (optional) streams dirs and links out while the walk continues;
// 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()
@ -37,7 +39,9 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
for (const entry of entries) {
const relChild = `${dirRel}/${entry.name}`
if (entry.isDirectory()) {
out.push({ path: relChild, type: 'dir' })
const record = { path: relChild, type: 'dir' }
out.push(record)
if (emit) await emit(record)
dirs.push([path.join(dirAbs, entry.name), relChild])
} else {
leaves.push(entry)
@ -51,7 +55,9 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
try {
target = await fsp.readlink(absChild)
} catch {}
out.push({ path: relChild, type: 'link', target: portableLinkTarget(absChild, target) })
const record = { path: relChild, type: 'link', target: portableLinkTarget(absChild, target) }
out.push(record)
if (emit) await emit(record)
return
}
let st
@ -70,8 +76,14 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
}
}
// Manifest of one unit; `force` skips hashing (receiver replaces everything).
async function buildManifest(roots, force, onProgress) {
// Manifest of one unit, streamed: `handlers` may carry onEntry (a Promise
// per entry, awaited - backpressure), onTotal and onProgress. File entries
// 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) {
@ -88,45 +100,69 @@ function createManifestBuilder({ filesPath, filesRoot, hash, mapLimit, concurren
try {
target = await fsp.readlink(abs)
} catch {}
entries.push({ path: rel, type: 'link', target: portableLinkTarget(abs, target) })
const record = { path: rel, type: 'link', target: portableLinkTarget(abs, target) }
entries.push(record)
if (onEntry) await onEntry(record)
} else if (st.isDirectory()) {
entries.push({ path: rel, type: 'dir' })
await scanTree(abs, rel, entries, mtimes)
const record = { path: rel, type: 'dir' }
entries.push(record)
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, so the bar starts empty
// total size is known before hashing: the receiver sizes its compare bar
// from this, the sender's popup from onProgress
const totalBytes = files.reduce((sum, entry) => sum + (entry.size || 0), 0)
if (!force && files.length) {
let done = 0
let hashed = 0
await mapLimit(files, concurrency, async entry => {
const abs = filesPath(entry.path)
try {
// unchanged size + mtime: the cached digest still holds
const mtimeMs = mtimes.get(entry.path) || 0
entry.hash = abs
? await hash.cachedHash(entry.path, abs, entry.size, mtimeMs, bytes => {
hashed += bytes
if (onProgress) onProgress(done, files.length, hashed, totalBytes)
})
: null
} catch {
entry.hash = null
}
if (onTotal) await onTotal(totalBytes)
let done = 0
let hashed = 0
// one batch hashes with the whole disk queue behind it; its entries go
// 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
const mtimeMs = mtimes.get(entry.path) || 0
entry.hash = abs
? await hash.cachedHash(entry.path, abs, entry.size, mtimeMs, bytes => {
hashed += bytes
if (onProgress) onProgress(done, files.length, hashed, totalBytes)
})
: null
} 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
}
return { buildManifest, scanTree, portableLinkTarget }
// Buffering variant for callers that want the whole manifest first.
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 }

128
src/lansync.js
View file

@ -240,14 +240,23 @@ 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 {
onEntry(JSON.parse(line))
} catch {}
parsed = JSON.parse(line)
} catch {
return
}
chain = chain.then(() => onEntry(parsed)).catch(() => {})
})
rl.on('close', resolve)
rl.on('error', resolve)
const done = () => chain.then(resolve, () => resolve())
rl.on('close', done)
rl.on('error', done)
})
}
@ -801,7 +810,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 }
totals: { apps: Number(announced.apps) || 0, wines: Number(announced.wines) || 0, prefixes: Number(announced.prefixes) || 0, settings: Number(announced.settings) || 0 }
})
} catch {
resolve('reject') // no window to answer with -> never auto-accept silently
@ -857,13 +866,14 @@ 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' })
const entries = []
await readNdjson(req, entry => {
if (inlineSame(entry)) entries.push(entry)
})
// the mode each file has to end up with on this side
const modes = new Map()
for (const entry of entries) if (entry.type === 'file' && typeof entry.mode === 'number') modes.set(entry.path, entry.mode)
// comparing is its own phase: the bar restarts for the copy phase
const announce = progressGate((data, force) => emitReceive(hooks, session, 'compare', data))
// the unit counts from the start of its comparison, not from its commit
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
}
// 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
@ -877,24 +887,64 @@ async function handlePlan(state, session, url, req, res, hooks) {
}
}
} catch {}
const allowed = entries.filter(entry => withinRoots(entry.path, owned))
// comparing is its own phase: the bar restarts for the copy phase
const unitTotalBytes = allowed.reduce((sum, entry) => sum + (entry.size || 0), 0)
const announce = progressGate((data, force) => emitReceive(hooks, session, 'plan', data))
// The manifest is compared as it streams in (both sides hash at the same
// time). The sender sends its unit total first, so the bar starts with its
// full width; files missing from the sender's total compare as the usual
// entries - the bar just grows when that happens.
const entries = []
const results = new Map()
let unitTotalBytes = 0
let compared = 0
let comparedBytes = 0
session.barTotal = unitTotalBytes
session.barBytes = 0
const identical = await mapLimit(allowed, HASH_CONCURRENCY, async entry => {
const compareOne = 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 => {
if (!inlineSame(entry)) return
if (entry.unitBytes != null) {
// the sender announces its unit total before hashing starts; it can
// arrive after early compares (dirs/links), so it just overwrites
unitTotalBytes = Number(entry.unitBytes) || 0
session.barTotal = unitTotalBytes || 1
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 = []
@ -922,17 +972,19 @@ async function handlePlan(state, session, url, req, res, hooks) {
session.unitWritten = 0
session.barTotal = uploadBytes
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
const key = req.headers['x-plwl-unit'] || url.searchParams.get('unit') || ''
@ -945,8 +997,22 @@ 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
@ -959,6 +1025,9 @@ 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--
}
}
@ -1030,10 +1099,13 @@ function handleEnd(state, session, res, hooks) {
emitReceive(hooks, session, 'done', { deleted: session.deleted || 0 })
}
// Liveness ping: hashing can take minutes; re-arms the popup watchdog.
// Liveness ping: hashing can take minutes; re-arms the popup watchdog. The
// 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()
emitReceive(hooks, session, 'wait')
if (!session.writing) emitReceive(hooks, session, 'wait')
sendJson(res, 200, { ok: true })
}
@ -1140,6 +1212,7 @@ function getPusher() {
filesPath,
ownedRoot,
buildManifest,
buildManifestStream: manifest.buildManifestStream,
listDirs,
listDevices,
mapLimit,
@ -1219,6 +1292,7 @@ const exported = {
filesPath,
ownedRoot,
buildManifest,
buildManifestStream: manifest.buildManifestStream,
mapLimit,
listDirs,
syncTo

256
src/lansyncpush.js
View file

@ -1,8 +1,9 @@
'use strict'
// The pushing side of a LAN sync: per unit, send a manifest, learn what the
// receiver has, upload the rest, delete the surplus. Exported as a factory;
// src/lansync.js hands in its own helpers and state.
// The pushing side of a LAN sync: per unit, stream a manifest while it is
// still being hashed, learn what the receiver has, upload the rest, delete
// the surplus. Exported as a factory; src/lansync.js hands in its own
// helpers and state.
const http = require('http')
const os = require('os')
@ -12,10 +13,22 @@ 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,
@ -26,7 +39,7 @@ function createPusher(api) {
httpJson,
filesPath,
ownedRoot,
buildManifest,
buildManifestStream,
listDirs,
DIRS,
toPortablePath,
@ -35,8 +48,10 @@ function createPusher(api) {
prefixes
} = api
// Manifest in, upload list out; NDJSON avoids one giant string.
function planUnit(device, token, unit, entries, force, meta) {
// Manifest in, upload list out; NDJSON avoids one giant string. The body
// is written by `produce` WHILE the sender hashes, so the receiver can
// 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) => {
@ -62,23 +77,47 @@ function createPusher(api) {
})
}
)
req.on('error', reject)
let index = 0
const pump = () => {
while (index < entries.length) {
if (!req.write(`${JSON.stringify(entries[index++])}\n`)) {
req.once('drain', pump)
return
}
let dead = null
req.on('error', err => {
dead = err
reject(err)
})
// ordered writes with backpressure: the receiver reads continuously,
// but a slow socket must not buffer the whole manifest in memory
const write = obj => new Promise((resolveWrite, rejectWrite) => {
if (dead) return rejectWrite(dead)
const onDrain = () => {
cleanup()
resolveWrite()
}
req.end()
}
pump()
const onErr = () => {
cleanup()
rejectWrite(dead || new Error('plan-write-failed'))
}
const cleanup = () => {
req.off('drain', onDrain)
req.off('error', onErr)
}
if (req.write(`${JSON.stringify(obj)}\n`)) {
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) {
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(
{
@ -96,12 +135,26 @@ function createPusher(api) {
},
res => {
res.resume()
res.on('end', () => resolve({ status: res.statusCode }))
res.on('end', () => settle(resolve, { status: res.statusCode }))
res.on('aborted', () => settle(reject, new Error('response-aborted')))
}
)
req.on('error', reject)
// The timer spans the WHOLE transfer, response included: a stalled
// 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', reject)
stream.on('error', err => settle(reject, err))
// 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)
})
}
@ -265,8 +318,9 @@ 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 }
const totals = { apps: 0, wines: 0, prefixes: 0, settings: 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 => {
@ -274,32 +328,6 @@ 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"
@ -341,13 +369,58 @@ function createPusher(api) {
// snapshot per file, forward a few times per second (phase changes always)
const gate = typeof api.progressGate === 'function' ? api.progressGate(progress) : data => progress(data)
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 {}
}
try {
const runUnits = async () => {
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)
@ -357,14 +430,36 @@ function createPusher(api) {
barBytes = 0
barTotal = 0
emit('scan', unit, null, null, true)
const entries = await buildManifest(roots, !!force, (done, total, hashedBytes, totalBytes) => {
barBytes = hashedBytes
barTotal = totalBytes
// no file count until the plan returns
emit('hash', unit, null, null)
// The manifest streams WHILE this side hashes its tail: the receiver
// 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
emit('hash', unit, null, null)
}
})
await writes
// this side is done hashing; the receiver may still be comparing
barTotal = barTotal || 1
barBytes = barTotal
emit('wait', unit, null, null, true)
})
emit('plan', unit, null, null, true)
const plan = await planUnit(device, token, { ...unit, roots }, entries, !!force, { totals })
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 || [])
@ -383,32 +478,32 @@ function createPusher(api) {
pending.push(entry)
plannedBytes += entry.size || 0
}
// a few files in flight at once: small files cost round trips
// the bar counts only bytes that move, and starts empty for the copy
// a few files in flight at once: small files cost round trips; 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) return
const res = await putFile(device, token, unit.key, entry.path, filesPath(entry.path))
if (res.status !== 200) {
failed = new Error(`upload-failed-${res.status}`)
return
let failed = null
let done = 0
const send = async entry => {
if (failed || abort.dead) return
const res = await putFile(device, token, unit.key, entry.path, filesPath(entry.path))
if (res.status !== 200) {
failed = new Error(`upload-failed-${res.status}`)
return
}
done++
summary.uploaded++
summary.bytes += entry.size || 0
barBytes += entry.size || 0
emit('copy', unit, allDone + done, allTotal + pending.length)
}
done++
summary.uploaded++
summary.bytes += entry.size || 0
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
}
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
@ -431,9 +526,16 @@ 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 {
@ -442,7 +544,9 @@ function createPusher(api) {
return syncResult(true, null, summary, events)
}
return { syncToDevice, buildUnits, buildRegistryPayload, CATEGORIES }
// putFile is exposed for the stall tests (it is the piece that must never
// leave a request half-open)
return { syncToDevice, buildUnits, buildRegistryPayload, putFile, CATEGORIES }
}
module.exports = createPusher