a LOT sync fixes...
This commit is contained in:
parent
c0f6e802ed
commit
806e173e43
14 changed files with 1073 additions and 240 deletions
|
|
@ -13,6 +13,10 @@ const os = require('os')
|
|||
const fs = require('fs')
|
||||
const path = require('path')
|
||||
|
||||
// Files sent at once: one at a time wastes most of a LAN round trip on the
|
||||
// small files a game or prefix is mostly made of.
|
||||
const UPLOAD_CONCURRENCY = 4
|
||||
|
||||
function createPusher(api) {
|
||||
const {
|
||||
CATEGORIES,
|
||||
|
|
@ -34,9 +38,12 @@ function createPusher(api) {
|
|||
|
||||
// Manifest in, list of files to upload out. NDJSON keeps a prefix with tens
|
||||
// of thousands of entries from being one giant string in memory either side.
|
||||
function planUnit(device, token, unit, entries, force) {
|
||||
function planUnit(device, token, unit, entries, force, meta) {
|
||||
// the receiver is told how many apps/prefixes/wines the whole sync carries
|
||||
// up front, which is what lets its popup say "Apps: 2/5"
|
||||
const totals = meta && meta.totals ? `&totals=${encodeURIComponent(JSON.stringify(meta.totals))}` : ''
|
||||
return new Promise((resolve, reject) => {
|
||||
const route = `/plwl/unit/plan?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}&label=${encodeURIComponent(unit.label || '')}&roots=${encodeURIComponent(JSON.stringify(unit.roots))}${force ? '&force=1' : ''}`
|
||||
const route = `/plwl/unit/plan?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}&label=${encodeURIComponent(unit.label || '')}&roots=${encodeURIComponent(JSON.stringify(unit.roots))}${force ? '&force=1' : ''}${totals}`
|
||||
const req = http.request(
|
||||
{
|
||||
host: device.address,
|
||||
|
|
@ -142,12 +149,11 @@ function createPusher(api) {
|
|||
|
||||
// --- registry payload -----------------------------------------------------
|
||||
|
||||
// Everything the app config holds travels: paths (portableGame rewrites them
|
||||
// to the FILES/prefix relative form both machines understand), the wine
|
||||
// build and prefix it runs in, args/env/dllOverrides and the play counters.
|
||||
function portableApp(game) {
|
||||
const out = library.portableGame(game)
|
||||
// playtime/lastPlayed are per machine, the receiver keeps its own counters
|
||||
delete out.playSeconds
|
||||
delete out.lastPlayed
|
||||
return out
|
||||
return library.portableGame(game)
|
||||
}
|
||||
|
||||
function buildRegistryPayload(categories) {
|
||||
|
|
@ -166,6 +172,15 @@ function createPusher(api) {
|
|||
// the receiver prunes its own builds down to this list and rescans
|
||||
payload.wines = wineUnits().map(unit => ({ name: unit.id }))
|
||||
}
|
||||
if (cats.has('defaults')) {
|
||||
// "Default settings": which wine everything falls back to and where a new
|
||||
// import lands. Opt-in on purpose - a sync of apps must not silently
|
||||
// change the other machine's defaults.
|
||||
payload.defaults = {
|
||||
defaultWine: (library.loadSettings().defaultWine) || 'system',
|
||||
defaultPrefix: prefixes.listPrefixes().default || ''
|
||||
}
|
||||
}
|
||||
if (cats.has('prefixes')) {
|
||||
const defs = prefixes.listPrefixes()
|
||||
payload.prefixes = {
|
||||
|
|
@ -230,6 +245,13 @@ function createPusher(api) {
|
|||
if (!device || !device.address || !device.port) return fail('offline')
|
||||
const cats = new Set(normalizeCategories(categories))
|
||||
if (!cats.size) return fail('nothing-selected')
|
||||
// The units are known before anything is read, so how many apps / wine
|
||||
// builds / prefixes this sync carries is announced with the verification:
|
||||
// the other side can show "Apps: 0/5" from its first event.
|
||||
const units = buildUnits(cats)
|
||||
const bucketOf = kind => (kind === 'wine' ? 'wines' : kind === 'prefix' ? 'prefixes' : 'apps')
|
||||
const totals = { apps: 0, wines: 0, prefixes: 0 }
|
||||
for (const unit of units) totals[bucketOf(unit.kind)] += 1
|
||||
// every progress event is kept (and handed to onProgress when given) so
|
||||
// callers can report on the transfer without subscribing first
|
||||
const events = []
|
||||
|
|
@ -241,7 +263,7 @@ function createPusher(api) {
|
|||
let verified
|
||||
try {
|
||||
verified = await httpJson(device, 'POST', '/plwl/verify', {
|
||||
json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase() },
|
||||
json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase(), totals },
|
||||
timeout: 20000
|
||||
})
|
||||
} catch (err) {
|
||||
|
|
@ -260,31 +282,42 @@ function createPusher(api) {
|
|||
// typing (a rejected code is never stored)
|
||||
rememberDevice(device.id || deviceId, code, device.name)
|
||||
|
||||
const units = buildUnits(cats)
|
||||
const bucketOf = kind => (kind === 'wine' ? 'wines' : kind === 'prefix' ? 'prefixes' : 'apps')
|
||||
const summary = newSummary(units.length)
|
||||
const finished = { apps: 0, wines: 0, prefixes: 0 }
|
||||
// bytes the units planned so far will put on the wire: the popup can then
|
||||
// show "1.2 GB / 3.4 GB" instead of only what has already arrived
|
||||
let plannedBytes = 0
|
||||
let itemsSeen = 0
|
||||
const emit = (phase, unit, done, total) =>
|
||||
progress({
|
||||
stage: phase,
|
||||
phase,
|
||||
kind: (unit && unit.kind) || '',
|
||||
id: (unit && unit.id) || '',
|
||||
key: (unit && unit.key) || '',
|
||||
unit: (unit && unit.label) || '',
|
||||
label: (unit && unit.label) || '',
|
||||
name: (unit && unit.label) || '',
|
||||
done,
|
||||
total,
|
||||
bytes: summary.bytes,
|
||||
totalBytes: plannedBytes,
|
||||
units: summary.units,
|
||||
unitsDone: itemsSeen,
|
||||
items: { ...summary.items }
|
||||
})
|
||||
// the bar tracks the PHASE being worked on (its own bytes), never a
|
||||
// leftover from the phase before: hashing starts from an empty bar
|
||||
let barBytes = 0
|
||||
let barTotal = 0
|
||||
const snapshot = (phase, unit, done, total) => ({
|
||||
stage: phase,
|
||||
phase,
|
||||
kind: (unit && unit.kind) || '',
|
||||
id: (unit && unit.id) || '',
|
||||
key: (unit && unit.key) || '',
|
||||
unit: (unit && unit.label) || '',
|
||||
label: (unit && unit.label) || '',
|
||||
name: (unit && unit.label) || '',
|
||||
done,
|
||||
total,
|
||||
bytes: summary.bytes,
|
||||
totalBytes: plannedBytes,
|
||||
units: summary.units,
|
||||
unitsDone: itemsSeen,
|
||||
items: { ...summary.items },
|
||||
itemsDone: { ...finished },
|
||||
totals: { ...totals },
|
||||
barBytes,
|
||||
barTotalBytes: barTotal
|
||||
})
|
||||
// hashing and copying walk thousands of files: the state is snapshotted on
|
||||
// every file but forwarded a few times per second (phase changes always
|
||||
// go through, so nothing on screen ever jumps backwards)
|
||||
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)
|
||||
const end = async () => {
|
||||
try {
|
||||
await httpJson(device, 'POST', `/plwl/end?token=${encodeURIComponent(token)}`, { json: {} })
|
||||
|
|
@ -294,13 +327,20 @@ function createPusher(api) {
|
|||
for (const unit of units) {
|
||||
const roots = unit.roots.filter(ownedRoot)
|
||||
if (!roots.length) continue
|
||||
const bucket = summary[bucketOf(unit.kind)]
|
||||
summary.items[bucketOf(unit.kind)] += 1
|
||||
const kind = bucketOf(unit.kind)
|
||||
const bucket = summary[kind]
|
||||
summary.items[kind] += 1
|
||||
itemsSeen += 1
|
||||
emit('scan', unit, 0, 0)
|
||||
const entries = await buildManifest(roots, !!force, (done, total) => emit('hash', unit, done, total))
|
||||
emit('plan', unit, 0, 0)
|
||||
const plan = await planUnit(device, token, { ...unit, roots }, entries, !!force)
|
||||
barBytes = 0
|
||||
barTotal = 0
|
||||
emit('scan', unit, 0, 0, true)
|
||||
const entries = await buildManifest(roots, !!force, (done, total, hashedBytes, totalBytes) => {
|
||||
barBytes = hashedBytes
|
||||
barTotal = totalBytes
|
||||
emit('hash', unit, done, total)
|
||||
})
|
||||
emit('plan', unit, 0, 0, true)
|
||||
const plan = await planUnit(device, token, { ...unit, roots }, entries, !!force, { totals })
|
||||
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 || [])
|
||||
|
|
@ -308,31 +348,56 @@ function createPusher(api) {
|
|||
// already identical (the plan reports both)
|
||||
bucket.copied += entries.length - same.size
|
||||
bucket.skipped += same.size
|
||||
for (const entry of entries) {
|
||||
if (entry.type === 'file' && upload.has(entry.path)) plannedBytes += entry.size || 0
|
||||
}
|
||||
let done = 0
|
||||
const pending = []
|
||||
for (const entry of entries) {
|
||||
if (entry.type !== 'file') continue
|
||||
if (!upload.has(entry.path)) {
|
||||
summary.skipped++
|
||||
continue
|
||||
}
|
||||
emit('copy', unit, done, upload.size)
|
||||
pending.push(entry)
|
||||
plannedBytes += entry.size || 0
|
||||
}
|
||||
// a few files in flight at once: with thousands of small ones the
|
||||
// round trip to the other machine costs more than the bytes do
|
||||
// the files that were already identical never move, so the bar counts
|
||||
// only the bytes that do - and starts empty for the transfer phase
|
||||
if (pending.length) {
|
||||
barBytes = 0
|
||||
barTotal = pending.reduce((sum, entry) => sum + (entry.size || 0), 0)
|
||||
emit('copy', unit, 0, 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) throw new Error(`upload-failed-${res.status}`)
|
||||
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, done, pending.length)
|
||||
}
|
||||
emit('copy', unit, done, upload.size)
|
||||
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, done, pending.length, true)
|
||||
const commit = await httpJson(device, 'POST', `/plwl/unit/commit?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}`, { json: {} })
|
||||
if (commit.data && commit.data.deleted) {
|
||||
summary.deleted += commit.data.deleted
|
||||
bucket.deleted += commit.data.deleted
|
||||
}
|
||||
finished[kind] += 1
|
||||
// a finished unit leaves a finished bar behind, even one that sent none
|
||||
barTotal = barTotal || 1
|
||||
barBytes = barTotal
|
||||
emit('commit', unit, 0, 0, true)
|
||||
}
|
||||
emit('registry', null, 0, 0)
|
||||
emit('registry', null, 0, 0, true)
|
||||
const applied = await httpJson(device, 'POST', `/plwl/apply?token=${encodeURIComponent(token)}`, { json: buildRegistryPayload([...cats]), timeout: 60000 })
|
||||
if (!applied.data || !applied.data.ok) throw new Error((applied.data && applied.data.error) || `http-${applied.status}`)
|
||||
for (const key of ['apps', 'wines', 'prefixes']) {
|
||||
|
|
@ -340,7 +405,7 @@ function createPusher(api) {
|
|||
if (deleted) summary[key].deleted += deleted
|
||||
}
|
||||
if (applied.data.deleted) summary.deleted += applied.data.deleted
|
||||
emit('done', null, summary.uploaded, summary.uploaded)
|
||||
emit('done', null, summary.uploaded, summary.uploaded, true)
|
||||
await end()
|
||||
} catch (err) {
|
||||
await end()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue