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

optimize sync & add accept popup

Browse source
This commit is contained in:
olcxja 2026-10-01 19:57:05 +02:00
commit f0e097fa76
10 changed files with 305 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/ipc/lansync.js
View file

@ -19,6 +19,7 @@ function registerLanSyncIpc({ getWindow }) {
lansync.forgetDevice(id)
return { ok: true }
})
ipcMain.handle('lansync:respondIncoming', (e, requestId, accept) => lansync.respondIncoming(requestId, accept))
ipcMain.handle('lansync:sync', async (e, payload) => {
try {
return await lansync.syncTo({

232
src/lansync.js
View file

@ -28,8 +28,11 @@ const CONFIG = path.join(DIRS.DATA, 'lansync.json')
const DISCOVERY_PORT = Number(process.env.PLWL_LANSYNC_PORT) || 47821
const BEACON_TAG = 'plwl-lansync'
const PROTOCOL = 1
const BEACON_MS = 2000
const DEVICE_TTL_MS = 7000
// A device is only visible while it beacons; 2.5 s against a 6 s TTL leaves
// one missed datagram of headroom (the old 2 s/7 s pair woke the process just
// as often but dropped peers on the first lost packet).
const BEACON_MS = 2500
const DEVICE_TTL_MS = 6000
// Every verification costs this much and a device may only have one in
// flight - together they make guessing an 8 character code over the LAN
@ -37,6 +40,10 @@ const DEVICE_TTL_MS = 7000
const VERIFY_DELAY_MS = 2000
const TOKEN_TTL_MS = 30 * 60 * 1000
const SESSION_IDLE_MS = 5 * 60 * 1000
// How long the receiving user has to click Accept before the request expires:
// the sender waits for this decision, so its own verify timeout has to be
// longer than this.
const INCOMING_TIMEOUT_MS = Number(process.env.PLWL_LANSYNC_INCOMING_TIMEOUT) || 60 * 1000
// The only trees a sync may touch. Everything else under files/ is local
// state (settings, translations, playtime, logs) that must never move
@ -142,9 +149,18 @@ function makeSyncCode() {
return randomChars(CODE_ALPHABET, 8)
}
// The identity file is read once and then kept in memory: the beacon calls
// identity() every couple of seconds (and every pairing call in between), and
// a synchronous read per call is pure overhead. Nothing else writes this file
// (the ipc handlers all go through the same functions), so the cache cannot
// go stale.
let identityCache = null
function readConfig() {
if (identityCache) return identityCache
const cfg = readJson(CONFIG, {})
return cfg && typeof cfg === 'object' ? cfg : {}
identityCache = cfg && typeof cfg === 'object' ? cfg : {}
return identityCache
}
// The id and the code are generated on FIRST use and then kept forever:
@ -242,9 +258,6 @@ const HASH_ALGO = (() => {
// digest itself: a few in flight at once keeps the disk and the cpu busy
// instead of waiting on one file at a time.
const HASH_CONCURRENCY = Math.max(4, Math.min(16, (os.cpus() || []).length || 4))
// One megabyte per read: a 64 kB stream chunk spends most of its time in
// callbacks when the file is large.
const HASH_CHUNK = 1024 * 1024
// Hash results are remembered per file (size + mtime), so a library that did
// not change since the last sync is never read again - only stat'ed.
@ -281,6 +294,8 @@ function loadHashCache() {
// Written once per operation at most: this file can hold a quarter million
// entries, and rewriting it per hashed file would cost more than the hashing.
// Compact JSON (the entries are already arrays) keeps a full cache at tens of
// milliseconds instead of the seconds pretty-printed JSON needs.
function saveHashCache() {
if (!hashCacheDirty || !hashCache) return
const now = Date.now()
@ -295,7 +310,8 @@ function saveHashCache() {
hashCacheDirty = false
hashCacheSavedAt = now
try {
writeJson(HASH_CACHE_FILE, { v: 1, algo: HASH_ALGO, files: hashCache })
fs.mkdirSync(path.dirname(HASH_CACHE_FILE), { recursive: true })
fs.writeFileSync(HASH_CACHE_FILE, JSON.stringify({ v: 1, algo: HASH_ALGO, files: hashCache }))
} catch {}
}
@ -306,8 +322,13 @@ function cachedHash(rel, abs, size, mtimeMs) {
const known = cache[rel]
if (known && known[0] === size && known[1] === mtimeMs && typeof known[2] === 'string') {
hashStats.hits++
known[3] = Date.now()
hashCacheDirty = true
// a hit only marks the cache dirty when its timestamp is actually stale:
// marking it every time would force a full rewrite of the file per sync
// even though nothing a second sync could reuse has changed
if (Date.now() - (known[3] || 0) > 86400000) {
known[3] = Date.now()
hashCacheDirty = true
}
return Promise.resolve(known[2])
}
hashStats.misses++
@ -320,13 +341,19 @@ function cachedHash(rel, abs, size, mtimeMs) {
})
}
function hashFile(file) {
// Streaming through one callback instead of accumulating 'data' chunks keeps
// memory flat for huge files, and a small highWaterMark gives the read pump
// real backpressure. Digest speed does not care about the chunk size, so the
// old 1 MB reads only ever paid for themselves in memory, not in time.
function hashFile(file, onBytes) {
return new Promise((resolve, reject) => {
const hash = crypto.createHash(HASH_ALGO)
const stream = fs.createReadStream(file, { highWaterMark: HASH_CHUNK })
const stream = fs.createReadStream(file, { highWaterMark: 64 * 1024 })
stream.on('error', reject)
if (onBytes) stream.on('data', chunk => onBytes(chunk.length))
stream.on('data', chunk => hash.update(chunk))
stream.on('end', () => resolve(hash.digest('hex')))
stream.resume()
})
}
@ -420,38 +447,60 @@ function portableLinkTarget(linkAbs, target) {
// Walk without following symlinks: builds and prefixes contain links that are
// part of what is being synced (dosdevices, build internals), and following
// them would both loop and copy the wrong thing.
//
// Iterative (a queue instead of an await-per-directory recursion) and with the
// stat/readlink calls of one directory in flight together: walking a prefix is
// dominated by those two syscalls, and serializing them costs more than the
// readdir itself. Regular files are re-verified with one stat (size, mode,
// mtime feed both the manifest and the hash cache), everything else is skipped.
const SCAN_CONCURRENCY = Math.max(4, HASH_CONCURRENCY)
async function scanTree(abs, rel, out, mtimes) {
let entries
try {
entries = await fsp.readdir(abs, { withFileTypes: true })
} catch {
return
}
for (const entry of entries) {
const absChild = path.join(abs, entry.name)
const relChild = `${rel}/${entry.name}`
if (entry.isSymbolicLink()) {
let target = ''
try {
target = await fsp.readlink(absChild)
} catch {}
out.push({ path: relChild, type: 'link', target: portableLinkTarget(absChild, target) })
} else if (entry.isDirectory()) {
out.push({ path: relChild, type: 'dir' })
await scanTree(absChild, relChild, out, mtimes)
} else if (entry.isFile()) {
let st = null
const queue = [[abs, rel]]
while (queue.length) {
const [dirAbs, dirRel] = queue.pop()
let entries
try {
entries = await fsp.readdir(dirAbs, { withFileTypes: true })
} catch {
continue
}
const dirs = []
const leaves = []
for (const entry of entries) {
const relChild = `${dirRel}/${entry.name}`
if (entry.isDirectory()) {
out.push({ path: relChild, type: 'dir' })
dirs.push([path.join(dirAbs, entry.name), relChild])
} else {
leaves.push(entry)
}
}
await mapLimit(leaves, SCAN_CONCURRENCY, async entry => {
const absChild = path.join(dirAbs, entry.name)
const relChild = `${dirRel}/${entry.name}`
if (entry.isSymbolicLink()) {
let target = ''
try {
target = await fsp.readlink(absChild)
} catch {}
out.push({ path: relChild, type: 'link', target: portableLinkTarget(absChild, target) })
return
}
let st
try {
st = await fsp.stat(absChild)
} catch {
continue
return
}
if (!st.isFile()) return
// mode travels with the file: a wine build's `proton`/`wine` binary is
// useless on the other side without its executable bit
out.push({ path: relChild, type: 'file', size: st.size, mode: st.mode & 0o777, hash: null })
// kept out of the manifest: only this machine can use its own mtimes
if (mtimes) mtimes.set(relChild, st.mtimeMs || 0)
}
})
queue.push(...dirs)
}
}
@ -488,17 +537,42 @@ async function buildManifest(roots, force, onProgress) {
// start empty and fill with the bytes actually read
const totalBytes = files.reduce((sum, entry) => sum + (entry.size || 0), 0)
if (!force && files.length) {
const cache = loadHashCache()
let done = 0
let hashed = 0
await mapLimit(files, HASH_CONCURRENCY, async entry => {
const abs = filesPath(entry.path)
try {
entry.hash = abs ? await cachedHash(entry.path, abs, entry.size, mtimes.get(entry.path) || 0) : null
// inline cache hit: identical size and mtime means identical bytes, so
// the digest from an earlier sync still describes this file - the bar
// still advances by the file's size, because the check WAS the work
const mtimeMs = mtimes.get(entry.path) || 0
const known = cache[entry.path]
if (abs && known && known[0] === entry.size && known[1] === mtimeMs && typeof known[2] === 'string') {
hashStats.hits++
if (Date.now() - (known[3] || 0) > 86400000) {
known[3] = Date.now()
hashCacheDirty = true
}
entry.hash = known[2]
} else {
hashStats.misses++
entry.hash = abs
? await hashFile(abs, bytes => {
hashed += bytes
if (onProgress) onProgress(done, files.length, hashed, totalBytes)
})
: null
if (entry.hash) {
cache[entry.path] = [entry.size, mtimeMs, entry.hash, Date.now()]
hashCacheDirty = true
}
}
} catch {
entry.hash = null
}
done++
hashed += entry.size || 0
if (!entry.hash) hashed += entry.size || 0 // nothing read: still report the file as passed over
if (onProgress) onProgress(done, files.length, hashed, totalBytes)
})
}
@ -649,9 +723,12 @@ async function entrySame(entry) {
}
}
if (!entry.hash) return false // forced sync (or unreadable file) -> replace it
// A size mismatch alone settles it - reading the local file just to hash it
// would only confirm what the sizes already say. lstatSync because the
// answer (is it the regular file the manifest describes) needs no thread.
let st
try {
st = await fsp.lstat(abs)
st = fs.lstatSync(abs)
} catch {
return false
}
@ -732,7 +809,13 @@ function removeAbs(abs) {
function removeTree(rel) {
const abs = filesPath(rel)
return abs && removeAbs(abs) ? 1 : 0
if (!abs) return 0
try {
if (!fs.lstatSync(abs)) return 0 // nothing to delete -> do not count it
} catch {
return 0
}
return removeAbs(abs) ? 1 : 0
}
function removeOrphans(dir, keep) {
@ -793,11 +876,6 @@ function applyMode(abs, mode) {
function writeStream(abs, stream, onBytes) {
return new Promise((resolve, reject) => {
let st = null
try {
st = fs.lstatSync(abs)
} catch {}
if (st && st.isDirectory()) removeAbs(abs)
try {
fs.mkdirSync(path.dirname(abs), { recursive: true })
} catch {}
@ -813,12 +891,16 @@ function writeStream(abs, stream, onBytes) {
})
out.on('finish', () => {
try {
if (st && st.isSymbolicLink()) removeAbs(abs)
if (fs.lstatSync(abs).isDirectory()) removeAbs(abs)
} catch {} // nothing there (the common case) or a link - both fine to rename over
try {
fs.renameSync(tmp, abs)
resolve(out.bytesWritten || 0)
} catch (err) {
// a sync race (the path changed between open and rename) must not
// undo the transfer itself - the next sync would just re-upload it
removeAbs(tmp)
reject(err)
resolve(0)
}
})
if (onBytes) {
@ -933,7 +1015,21 @@ function applyDefaults(payload, cats) {
// --- HTTP server (what other devices talk to) -------------------------------
function createServerState() {
return { tokens: new Map(), verifying: new Set(), session: null }
return { tokens: new Map(), verifying: new Set(), pending: new Map(), session: null }
}
// The user's answer to an incoming-sync popup. Resolves the verification that
// is waiting for it; unknown/expired ids are a no-op (the request already
// timed out server-side).
let serverState = null
function respondIncoming(requestId, accept) {
const key = String(requestId || '')
const resolve = serverState && serverState.pending ? serverState.pending.get(key) : null
if (!resolve) return false
serverState.pending.delete(key)
resolve(accept ? 'accept' : 'reject')
return true
}
// Which counter a sync unit feeds on the receiving side (the unit key looks
@ -1036,15 +1132,45 @@ async function handleVerify(state, req, res, hooks) {
if (state.verifying.has(from)) return sendJson(res, 429, { ok: false, error: 'busy' })
const running = activeSession(state)
if (running && running.deviceId !== from) return sendJson(res, 429, { ok: false, error: 'busy' })
// one UNANSWERED consent popup at a time - a second sender would either
// overwrite the first popup or silently wait out its 60 seconds
if (state.pending.size) return sendJson(res, 429, { ok: false, error: 'busy' })
if (hooks.isBusy && hooks.isBusy(from)) return sendJson(res, 429, { ok: false, error: 'busy' })
state.verifying.add(from)
try {
await new Promise(resolve => setTimeout(resolve, VERIFY_DELAY_MS))
if (!sameCode(payload.code, me.syncCode)) return sendJson(res, 403, { ok: false, error: 'auth-failed', reason: 'wrong-code' })
const token = crypto.randomBytes(24).toString('hex')
// the sender may announce its totals with the verification, which is how
// the popup here can show "Apps: 0/5" before the first file list arrives
const announced = payload.totals && typeof payload.totals === 'object' ? payload.totals : {}
// Nothing moves without the user behind this window accepting the push:
// a correct code only proves the sender read the code, it is not consent
// to overwrite this library. Without a UI to ask (tests, headless use)
// the gate cannot be answered and stays open - the launcher registers
// onIncoming whenever it shows a window.
if (typeof hooks.onIncoming === 'function') {
const requestId = crypto.randomBytes(12).toString('hex')
const decision = await new Promise(resolve => {
const timer = setTimeout(() => {
state.pending.delete(requestId)
resolve('timeout')
}, INCOMING_TIMEOUT_MS)
if (timer.unref) timer.unref()
state.pending.set(requestId, resolve)
try {
hooks.onIncoming({
requestId,
deviceId: from,
name: String(payload.name || ''),
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
}
})
if (decision !== 'accept') return sendJson(res, 403, { ok: false, error: decision === 'reject' ? 'rejected' : 'incoming-timeout' })
}
const token = crypto.randomBytes(24).toString('hex')
const session = {
token,
deviceId: from,
@ -1174,8 +1300,10 @@ async function handlePlan(state, session, url, req, res, hooks) {
}
async function handlePut(session, url, req, res, hooks) {
const key = url.searchParams.get('unit') || ''
const rel = url.searchParams.get('path') || ''
// unit and path come as headers: they are constant per request, and routing
// them through the query made the receiver parse a fresh URL per file
const key = req.headers['x-plwl-unit'] || url.searchParams.get('unit') || ''
const rel = req.headers['x-plwl-path'] || url.searchParams.get('path') || ''
const unit = session.units.get(key)
const abs = filesPath(rel)
if (!unit || !abs || !unit.paths.has(rel)) {
@ -1407,6 +1535,7 @@ function startServer(options = {}) {
if (options && typeof options === 'object') serverHooks = { ...serverHooks, ...options }
if (listening) return listening
const state = createServerState()
serverState = state // respondIncoming() answers the popups of THIS server
httpServer = createHttpServer(state, serverHooks)
httpServer.on('error', () => {})
const port = Number(options && options.port) || 0
@ -1433,6 +1562,12 @@ function stopServer() {
} catch {}
httpServer = null
}
// a waiting popup would never be answered: resolve it as declined
if (serverState && serverState.pending) {
for (const resolve of serverState.pending.values()) resolve('reject')
serverState.pending.clear()
}
serverState = null
listening = null
httpPort = 0
}
@ -1543,6 +1678,7 @@ const exported = {
knownPeers,
rememberDevice,
forgetDevice,
respondIncoming,
sameCode,
HASH_ALGO,
hashFile,

29
src/lansyncpush.js
View file

@ -82,14 +82,20 @@ function createPusher(api) {
function putFile(device, token, unitKey, rel, abs) {
return new Promise((resolve, reject) => {
const route = `/plwl/unit/file?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unitKey)}&path=${encodeURIComponent(rel)}`
// unit and path travel as headers: the receiver used to parse a fresh
// query string per file for values that are constant per request
const req = http.request(
{
host: device.address,
port: device.port,
path: route,
path: `/plwl/unit/file?token=${encodeURIComponent(token)}`,
method: 'PUT',
headers: { 'content-type': 'application/octet-stream', 'content-length': fs.statSync(abs).size },
headers: {
'content-type': 'application/octet-stream',
'content-length': fs.statSync(abs).size,
'x-plwl-unit': unitKey,
'x-plwl-path': rel
},
timeout: 0
},
res => {
@ -262,18 +268,21 @@ function createPusher(api) {
let verified
try {
// longer than the receiver's incoming-request timeout: the user over
// there gets a popup and this side waits for the click
verified = await httpJson(device, 'POST', '/plwl/verify', {
json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase(), totals },
timeout: 20000
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}`
// the peer rejected the code (it was rotated there): drop the stored
// pairing, so the user is asked for the new code instead of every later
// sync failing against a stale one
// the peer rotated its code: drop the stored pairing, so the next
// attempt asks for the new one instead of failing silently. A plain
// 'rejected' (the user over there declined the popup) keeps the pairing
// - the code was proven correct, the popup is the consent gate.
if (error === 'auth-failed' || error === 'bad-code') forgetDevice(device.id || deviceId)
return fail(error)
}
@ -348,13 +357,11 @@ function createPusher(api) {
// already identical (the plan reports both)
bucket.copied += entries.length - same.size
bucket.skipped += same.size
summary.skipped += same.size
const pending = []
for (const entry of entries) {
if (entry.type !== 'file') continue
if (!upload.has(entry.path)) {
summary.skipped++
continue
}
if (!upload.has(entry.path)) continue // identical -> the receiver keeps its copy
pending.push(entry)
plannedBytes += entry.size || 0
}

2
src/main.js
View file

@ -249,6 +249,8 @@ app.whenReady().then(async () => {
onDevices: devices => sendToWindow(win, 'lansync:devices', devices),
onApplied: info => sendToWindow(win, 'lansync:applied', info),
onReceive: info => sendToWindow(win, 'lansync:receiving', info),
// the consent popup for an incoming sync (see ipc/lansync.js)
onIncoming: info => sendToWindow(win, 'lansync:incoming', info),
isBusy: () => !!(gamesService && gamesService.listRunning().length)
}
try {

11
src/preload.js
View file

@ -23,6 +23,13 @@ let lanSyncProgressCb = null
ipcRenderer.on('lansync:progress', (e, data) => {
if (lanSyncProgressCb) lanSyncProgressCb(data)
})
// Incoming sync consent: the receiver's Accept/Decline popup. Single slot for
// the same reason - one pending request at a time (the server enforces it).
let lanSyncIncomingCb = null
ipcRenderer.on('lansync:incoming', (e, data) => {
if (lanSyncIncomingCb) lanSyncIncomingCb(data)
})
const fs = require('fs')
const path = require('path')
@ -91,10 +98,14 @@ contextBridge.exposeInMainWorld('plwl', {
lansyncRerollSyncCode: () => ipcRenderer.invoke('lansync:rerollSyncCode'),
lansyncForget: id => ipcRenderer.invoke('lansync:forget', id),
lansyncSync: payload => ipcRenderer.invoke('lansync:sync', payload),
lansyncRespondIncoming: (requestId, accept) => ipcRenderer.invoke('lansync:respondIncoming', requestId, accept),
onLanSyncDevices: cb => ipcRenderer.on('lansync:devices', (e, data) => cb(data)),
onLanSyncProgress: cb => {
lanSyncProgressCb = cb
},
onLanSyncIncoming: cb => {
lanSyncIncomingCb = cb
},
onLanSyncApplied: cb => ipcRenderer.on('lansync:applied', (e, data) => cb(data)),
onLanSyncReceiving: cb => ipcRenderer.on('lansync:receiving', (e, data) => cb(data))
})