code refactor & sync bugfixes
This commit is contained in:
parent
2bcb47076a
commit
9edcc9841b
20 changed files with 820 additions and 812 deletions
574
src/lansync.js
574
src/lansync.js
|
|
@ -7,7 +7,6 @@
|
|||
// behind, and paths stored in the registries stay relative as they are.
|
||||
|
||||
const http = require('http')
|
||||
const dgram = require('dgram')
|
||||
const crypto = require('crypto')
|
||||
const os = require('os')
|
||||
const path = require('path')
|
||||
|
|
@ -17,10 +16,15 @@ const readline = require('readline')
|
|||
|
||||
const { DIRS, readJson, writeJson, toPortablePath } = require('./paths')
|
||||
const { rewireSymlinks, restoreExecutableBits } = require('./symlinks')
|
||||
const { progressGate } = require('./fsutil')
|
||||
const { progressGate, mapLimit } = require('./fsutil')
|
||||
const { createHashCache } = require('./hashcache')
|
||||
const { createIdentity, sameCode } = require('./lansync-identity')
|
||||
const { createManifestBuilder } = require('./lansync-manifest')
|
||||
const { createDiscovery } = require('./lansync-discovery')
|
||||
const library = require('./library')
|
||||
const wines = require('./wines')
|
||||
const prefixes = require('./prefixes')
|
||||
const { isPrefixRel } = prefixes
|
||||
|
||||
const CONFIG = path.join(DIRS.DATA, 'lansync.json')
|
||||
|
||||
|
|
@ -39,7 +43,9 @@ const DEVICE_TTL_MS = 6000
|
|||
// flight - together they make guessing an 8 character code over the LAN
|
||||
// pointless (and a failed guess cannot be retried in a tight loop).
|
||||
const VERIFY_DELAY_MS = 2000
|
||||
const TOKEN_TTL_MS = 30 * 60 * 1000
|
||||
// How long a session token stays valid without any request. Env override for
|
||||
// tests, which cannot wait half an hour to prove expiry.
|
||||
const TOKEN_TTL_MS = Number(process.env.PLWL_LANSYNC_TOKEN_TTL) || 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
|
||||
|
|
@ -100,7 +106,6 @@ function appCategory(game) {
|
|||
// inside its prefix ("drive_c/..." for plain wine, "pfx/drive_c/..." for
|
||||
// Proton-managed prefixes) and anything absolute or ".." points outside the
|
||||
// launcher data, so it is never part of a sync.
|
||||
const PREFIX_REL_RE = /^(pfx[/\\])?drive_c([/\\]|$)/i
|
||||
function classifyApp(app) {
|
||||
if (!app || (typeof app !== 'string' && typeof app !== 'object')) return null
|
||||
const raw = typeof app === 'string' ? app : app.path || app.exePath || app.workdir
|
||||
|
|
@ -109,7 +114,7 @@ function classifyApp(app) {
|
|||
if (!raw || typeof raw !== 'string') return 'external'
|
||||
const value = raw.replace(/\\/g, '/')
|
||||
const outside = value.startsWith('..') || path.isAbsolute(raw) || /^[a-z]:\//i.test(value) || /^[a-z][a-z0-9+.-]*:\/\//i.test(value)
|
||||
if (PREFIX_REL_RE.test(value)) return 'prefix'
|
||||
if (isPrefixRel(value)) return 'prefix'
|
||||
if (/^apps\//.test(value)) return 'portable'
|
||||
return outside ? 'external' : null
|
||||
}
|
||||
|
|
@ -125,248 +130,21 @@ function normalizeCategories(list) {
|
|||
|
||||
// --- identity ---------------------------------------------------------------
|
||||
|
||||
// The sync code is read out on one machine and typed on the other, so the
|
||||
// alphabet skips the characters people mix up (0/O, 1/I/L).
|
||||
const CODE_ALPHABET = 'ABCDEFGHJKMNPQRSTUVWXYZ23456789'
|
||||
const ID_ALPHABET = 'abcdefghijklmnopqrstuvwxyz0123456789'
|
||||
|
||||
function randomChars(alphabet, count) {
|
||||
let out = ''
|
||||
while (out.length < count) {
|
||||
for (const byte of crypto.randomBytes(count)) {
|
||||
if (out.length >= count) break
|
||||
out += alphabet[byte % alphabet.length]
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// 24 characters: the current timestamp in base36 (unique per millisecond,
|
||||
// 8 chars until 2089) plus random padding, so two machines generating an id
|
||||
// within the same millisecond still get different ids.
|
||||
function makeDeviceId() {
|
||||
const stamp = Date.now().toString(36)
|
||||
return (stamp + randomChars(ID_ALPHABET, Math.max(0, 24 - stamp.length))).slice(0, 24)
|
||||
}
|
||||
|
||||
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, {})
|
||||
identityCache = cfg && typeof cfg === 'object' ? cfg : {}
|
||||
return identityCache
|
||||
}
|
||||
|
||||
// The id and the code are generated on FIRST use and then kept forever:
|
||||
// other devices remember the id (they pair with it) and the code (so it only
|
||||
// has to be typed once).
|
||||
function identity() {
|
||||
const cfg = readConfig()
|
||||
let changed = false
|
||||
if (typeof cfg.deviceId !== 'string' || cfg.deviceId.length !== 24) {
|
||||
cfg.deviceId = makeDeviceId()
|
||||
changed = true
|
||||
}
|
||||
if (typeof cfg.syncCode !== 'string' || cfg.syncCode.length !== 8) {
|
||||
cfg.syncCode = makeSyncCode()
|
||||
changed = true
|
||||
}
|
||||
if (!cfg.known || typeof cfg.known !== 'object') {
|
||||
cfg.known = {}
|
||||
changed = true
|
||||
}
|
||||
if (changed) writeJson(CONFIG, cfg)
|
||||
return cfg
|
||||
}
|
||||
|
||||
// Both rotations return the whole identity ({ deviceId, syncCode, known }),
|
||||
// so callers get a consistent view of this device after the change.
|
||||
function rotateDeviceId() {
|
||||
const cfg = identity()
|
||||
const previous = cfg.deviceId
|
||||
cfg.deviceId = makeDeviceId()
|
||||
writeJson(CONFIG, cfg)
|
||||
if (cfg.deviceId === previous) return rotateDeviceId()
|
||||
return cfg
|
||||
}
|
||||
|
||||
function rotateSyncCode() {
|
||||
const cfg = identity()
|
||||
const previous = cfg.syncCode
|
||||
cfg.syncCode = makeSyncCode()
|
||||
writeJson(CONFIG, cfg)
|
||||
if (cfg.syncCode === previous) return rotateSyncCode() // never a no-op rotation
|
||||
return cfg
|
||||
}
|
||||
|
||||
// A device the user synced with successfully: its code is stored next to its
|
||||
// id, so the next sync needs no typing. A rejected code is forgotten again,
|
||||
// which is what drives the "type the code again" flow.
|
||||
function knownDevice(id) {
|
||||
return identity().known[id] || null
|
||||
}
|
||||
|
||||
// Devices this launcher paired with: id -> { code, name, at }. The code is the
|
||||
// one that was accepted, so later syncs need no typing.
|
||||
function knownPeers() {
|
||||
return identity().known || {}
|
||||
}
|
||||
|
||||
function rememberDevice(id, code, name) {
|
||||
const cfg = identity()
|
||||
const stored = String(code || '').trim().toUpperCase()
|
||||
cfg.known = { ...cfg.known, [id]: { code: stored, syncCode: stored, name: name || '', at: Date.now() } }
|
||||
writeJson(CONFIG, cfg)
|
||||
}
|
||||
|
||||
function forgetDevice(id) {
|
||||
const cfg = identity()
|
||||
const known = { ...cfg.known }
|
||||
delete known[id]
|
||||
cfg.known = known
|
||||
writeJson(CONFIG, cfg)
|
||||
}
|
||||
|
||||
function sameCode(a, b) {
|
||||
const x = Buffer.from(String(a == null ? '' : a).trim().toUpperCase())
|
||||
const y = Buffer.from(String(b == null ? '' : b).trim().toUpperCase())
|
||||
return x.length === y.length && x.length > 0 && crypto.timingSafeEqual(x, y)
|
||||
}
|
||||
const identityApi = createIdentity(CONFIG)
|
||||
const { identity, rotateDeviceId, rotateSyncCode, knownDevice, knownPeers, rememberDevice, forgetDevice, makeDeviceId, makeSyncCode } = identityApi
|
||||
|
||||
// --- files: hashing, paths, manifests ---------------------------------------
|
||||
|
||||
// BLAKE2b resists collisions like SHA-2 but is noticeably faster - syncing a
|
||||
// wine prefix means hashing tens of thousands of files, so this is the hot
|
||||
// path. Older runtimes fall back to SHA-2.
|
||||
const HASH_ALGO = (() => {
|
||||
for (const algo of ['blake2b512', 'sha512', 'sha256']) {
|
||||
try {
|
||||
crypto.createHash(algo)
|
||||
return algo
|
||||
} catch {}
|
||||
}
|
||||
return 'sha256'
|
||||
})()
|
||||
|
||||
// Hashing a tree is dominated by opening the (often tiny) files, not by the
|
||||
// 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))
|
||||
|
||||
// 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.
|
||||
const HASH_CACHE_FILE = path.join(DIRS.DATA, 'lansync-hashes.json')
|
||||
const HASH_CACHE_MAX = 250000
|
||||
let hashCache = null
|
||||
let hashCacheDirty = false
|
||||
let hashCacheSavedAt = 0
|
||||
const hashStats = { hits: 0, misses: 0 }
|
||||
|
||||
function hashCacheStats() {
|
||||
return { ...hashStats, size: hashCache ? Object.keys(hashCache).length : 0, algo: HASH_ALGO }
|
||||
}
|
||||
|
||||
function resetHashCacheStats() {
|
||||
hashStats.hits = 0
|
||||
hashStats.misses = 0
|
||||
}
|
||||
|
||||
// Dropping the in-memory copy makes the next lookup re-read the file from
|
||||
// disk - what tests use to prove a file was hashed instead of remembered.
|
||||
function forgetHashCache() {
|
||||
hashCache = null
|
||||
hashCacheDirty = false
|
||||
}
|
||||
|
||||
function loadHashCache() {
|
||||
if (!hashCache) {
|
||||
const stored = readJson(HASH_CACHE_FILE, null)
|
||||
hashCache = stored && typeof stored === 'object' && stored.files && typeof stored.files === 'object' ? stored.files : {}
|
||||
}
|
||||
return hashCache
|
||||
}
|
||||
|
||||
// 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()
|
||||
if (now - hashCacheSavedAt < 60000) return
|
||||
const names = Object.keys(hashCache)
|
||||
if (names.length > HASH_CACHE_MAX) {
|
||||
names.sort((a, b) => (hashCache[b][3] || 0) - (hashCache[a][3] || 0))
|
||||
const trimmed = {}
|
||||
for (const name of names.slice(0, HASH_CACHE_MAX)) trimmed[name] = hashCache[name]
|
||||
hashCache = trimmed
|
||||
}
|
||||
hashCacheDirty = false
|
||||
hashCacheSavedAt = now
|
||||
try {
|
||||
fs.mkdirSync(path.dirname(HASH_CACHE_FILE), { recursive: true })
|
||||
fs.writeFileSync(HASH_CACHE_FILE, JSON.stringify({ v: 1, algo: HASH_ALGO, files: hashCache }))
|
||||
} catch {}
|
||||
}
|
||||
|
||||
// The remembered digest of a file, or null when it has to be read again.
|
||||
// A hit only marks the cache dirty when its timestamp is actually stale:
|
||||
// marking it on every lookup would force a full rewrite of the file per sync
|
||||
// even though nothing a second sync could reuse has changed.
|
||||
function hashCacheHit(rel, size, mtimeMs) {
|
||||
const cache = loadHashCache()
|
||||
const known = cache[rel]
|
||||
if (!known || known[0] !== size || known[1] !== mtimeMs || typeof known[2] !== 'string') {
|
||||
hashStats.misses++
|
||||
return null
|
||||
}
|
||||
hashStats.hits++
|
||||
if (Date.now() - (known[3] || 0) > 86400000) {
|
||||
known[3] = Date.now()
|
||||
hashCacheDirty = true
|
||||
}
|
||||
return known[2]
|
||||
}
|
||||
|
||||
// Hash of a file we already stat'ed: size and mtime unchanged means the bytes
|
||||
// are the same, and the digest from an earlier sync still describes them.
|
||||
function cachedHash(rel, abs, size, mtimeMs) {
|
||||
const known = hashCacheHit(rel, size, mtimeMs)
|
||||
if (known) return Promise.resolve(known)
|
||||
return hashFile(abs).then(hash => {
|
||||
if (hash) {
|
||||
loadHashCache()[rel] = [size, mtimeMs, hash, Date.now()]
|
||||
hashCacheDirty = true
|
||||
}
|
||||
return hash
|
||||
})
|
||||
}
|
||||
|
||||
// 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: 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()
|
||||
})
|
||||
}
|
||||
// Digests are remembered per file (size + mtime), so a library that did not
|
||||
// change since the last sync is never read again - only stat'ed. One store per
|
||||
// launcher state (tests re-require this module per device).
|
||||
const hashStore = createHashCache(DIRS.DATA)
|
||||
const hashFile = hashStore.hashFile
|
||||
|
||||
// Protocol paths are FILES-relative with '/' separators. Anything escaping
|
||||
// files/ (a '..' segment, an absolute path) is rejected - manifests come from
|
||||
|
|
@ -386,163 +164,10 @@ function ownedRoot(rel) {
|
|||
return SYNC_ROOTS.some(prefix => clean.startsWith(prefix)) ? clean : null
|
||||
}
|
||||
|
||||
async function mapLimit(items, limit, fn) {
|
||||
const out = new Array(items.length)
|
||||
let next = 0
|
||||
const workers = Array.from({ length: Math.min(limit, items.length) }, async () => {
|
||||
while (next < items.length) {
|
||||
const i = next++
|
||||
out[i] = await fn(items[i], i)
|
||||
}
|
||||
})
|
||||
await Promise.all(workers)
|
||||
return out
|
||||
}
|
||||
// --- manifests (the sender's hashing) ---------------------------------------
|
||||
|
||||
// Symlink targets travel RELATIVE whenever they point inside the launcher data
|
||||
// (a prefix linking into its wine build, an app linking to its own assets): the
|
||||
// mirrored trees have the same internal layout on both machines, so a relative
|
||||
// target is correct over there while an absolute one would point at the
|
||||
// sender's own folders - which is exactly how a synced prefix ends up with
|
||||
// hundreds of dangling links. Targets outside files/ (a system font, a mounted
|
||||
// drive, /) are machine-specific and stay as they are.
|
||||
function portableLinkTarget(linkAbs, target) {
|
||||
if (typeof target !== 'string' || !target || !path.isAbsolute(target)) return target
|
||||
const rel = path.relative(path.dirname(linkAbs), target)
|
||||
if (!rel || path.isAbsolute(rel)) return target
|
||||
const root = path.resolve(DIRS.FILES)
|
||||
const resolved = path.resolve(path.dirname(linkAbs), rel)
|
||||
if (resolved !== root && !resolved.startsWith(root + path.sep)) return target
|
||||
return rel
|
||||
}
|
||||
|
||||
// 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) {
|
||||
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 {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// Manifest of one sync unit. `force` skips hashing entirely - the receiver
|
||||
// then treats every file as different ("force sync all files").
|
||||
async function buildManifest(roots, force, onProgress) {
|
||||
const entries = []
|
||||
const mtimes = new Map()
|
||||
for (const rel of roots) {
|
||||
const abs = filesPath(rel)
|
||||
if (!abs) continue
|
||||
let st
|
||||
try {
|
||||
st = await fsp.lstat(abs)
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
if (st.isSymbolicLink()) {
|
||||
let target = ''
|
||||
try {
|
||||
target = await fsp.readlink(abs)
|
||||
} catch {}
|
||||
entries.push({ path: rel, type: 'link', target: portableLinkTarget(abs, target) })
|
||||
} else if (st.isDirectory()) {
|
||||
entries.push({ path: rel, type: 'dir' })
|
||||
await scanTree(abs, rel, entries, mtimes)
|
||||
} 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')
|
||||
// the unit's total size is known before a byte is hashed, so the bar can
|
||||
// start empty and fill with the bytes actually read
|
||||
const totalBytes = files.reduce((sum, entry) => sum + (entry.size || 0), 0)
|
||||
if (!force && files.length) {
|
||||
let done = 0
|
||||
let hashed = 0
|
||||
await mapLimit(files, HASH_CONCURRENCY, async entry => {
|
||||
const abs = filesPath(entry.path)
|
||||
try {
|
||||
// identical size and mtime means identical bytes, so the digest from
|
||||
// an earlier sync still describes this file
|
||||
const mtimeMs = mtimes.get(entry.path) || 0
|
||||
const known = hashCacheHit(entry.path, entry.size, mtimeMs)
|
||||
if (known && abs) {
|
||||
entry.hash = known
|
||||
} else {
|
||||
entry.hash = abs
|
||||
? await hashFile(abs, bytes => {
|
||||
hashed += bytes
|
||||
if (onProgress) onProgress(done, files.length, hashed, totalBytes)
|
||||
})
|
||||
: null
|
||||
if (entry.hash) {
|
||||
loadHashCache()[entry.path] = [entry.size, mtimeMs, entry.hash, Date.now()]
|
||||
hashCacheDirty = true
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
entry.hash = null
|
||||
}
|
||||
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)
|
||||
})
|
||||
}
|
||||
saveHashCache()
|
||||
return entries
|
||||
}
|
||||
const manifest = createManifestBuilder({ filesPath, filesRoot: DIRS.FILES, hash: hashStore, mapLimit, concurrency: HASH_CONCURRENCY })
|
||||
const buildManifest = manifest.buildManifest
|
||||
|
||||
// --- HTTP helpers (shared with the pusher) ----------------------------------
|
||||
|
||||
|
|
@ -699,7 +324,7 @@ async function entrySame(entry) {
|
|||
if (!st.isFile() || st.size !== entry.size) return false
|
||||
let same = false
|
||||
try {
|
||||
same = (await cachedHash(entry.path, abs, st.size, st.mtimeMs || 0)) === entry.hash
|
||||
same = (await hashStore.cachedHash(entry.path, abs, st.size, st.mtimeMs || 0)) === entry.hash
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
|
|
@ -1194,6 +819,13 @@ function authorize(state, token) {
|
|||
return null
|
||||
}
|
||||
session.lastAt = Date.now()
|
||||
// The expiry SLIDES with every request: hashing a big unit takes longer than
|
||||
// any fixed TTL, and the request that finally moves files would arrive on a
|
||||
// token that died while the sender was still hashing - the sync then ends
|
||||
// right when it is ready to send. The heartbeat keeps the token alive through
|
||||
// such a phase; a sender that really vanished still ages out this long after
|
||||
// its last request.
|
||||
session.expires = Date.now() + TOKEN_TTL_MS
|
||||
if (state.session && state.session.token === token) state.session.lastAt = session.lastAt
|
||||
return session
|
||||
}
|
||||
|
|
@ -1371,7 +1003,7 @@ async function handlePlan(state, session, url, req, res, hooks) {
|
|||
announce({ unit: label })
|
||||
return same === true
|
||||
})
|
||||
saveHashCache()
|
||||
hashStore.save()
|
||||
const paths = new Set()
|
||||
const upload = []
|
||||
const same = []
|
||||
|
|
@ -1533,128 +1165,22 @@ function handlePing(session, res, hooks) {
|
|||
|
||||
// --- discovery --------------------------------------------------------------
|
||||
|
||||
let beaconSocket = null
|
||||
let beaconTimer = null
|
||||
let httpPort = 0
|
||||
let seen = new Map()
|
||||
let devicesCb = null
|
||||
|
||||
function ipToInt(ip) {
|
||||
const parts = String(ip).split('.')
|
||||
if (parts.length !== 4) return null
|
||||
let out = 0
|
||||
for (const part of parts) {
|
||||
const n = Number(part)
|
||||
if (!Number.isInteger(n) || n < 0 || n > 255) return null
|
||||
out = (out << 8) | n
|
||||
}
|
||||
return out >>> 0
|
||||
}
|
||||
|
||||
function intToIp(int) {
|
||||
return [24, 16, 8, 0].map(shift => (int >>> shift) & 255).join('.')
|
||||
}
|
||||
|
||||
// Broadcast to the global address AND to every interface's own subnet
|
||||
// broadcast - some networks drop one or the other.
|
||||
function broadcastAddresses() {
|
||||
const out = ['255.255.255.255']
|
||||
const interfaces = os.networkInterfaces()
|
||||
for (const list of Object.values(interfaces)) {
|
||||
for (const iface of list || []) {
|
||||
if (iface.family !== 'IPv4' || iface.internal || !iface.netmask) continue
|
||||
const addr = ipToInt(iface.address)
|
||||
const mask = ipToInt(iface.netmask)
|
||||
if (addr == null || mask == null) continue
|
||||
out.push(intToIp((addr | (~mask >>> 0)) >>> 0))
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
function listDevices() {
|
||||
const known = identity().known
|
||||
const now = Date.now()
|
||||
return [...seen.values()]
|
||||
.filter(d => now - d.at <= DEVICE_TTL_MS)
|
||||
.map(d => {
|
||||
const pair = known[d.id]
|
||||
return { id: d.id, name: d.name, address: d.address, port: d.port, paired: !!pair, code: pair ? pair.code : '' }
|
||||
})
|
||||
.sort((a, b) => String(a.name).localeCompare(String(b.name)))
|
||||
}
|
||||
|
||||
function emitDevices() {
|
||||
if (devicesCb) devicesCb(listDevices())
|
||||
}
|
||||
|
||||
function pruneDevices() {
|
||||
const now = Date.now()
|
||||
let changed = false
|
||||
for (const [id, device] of seen) {
|
||||
if (now - device.at > DEVICE_TTL_MS) {
|
||||
seen.delete(id)
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if (changed) emitDevices()
|
||||
}
|
||||
|
||||
function beacon() {
|
||||
if (!beaconSocket || !httpPort) return
|
||||
const me = identity()
|
||||
const payload = Buffer.from(JSON.stringify({ t: BEACON_TAG, v: PROTOCOL, id: me.deviceId, name: os.hostname(), port: httpPort }))
|
||||
for (const address of broadcastAddresses()) {
|
||||
try {
|
||||
beaconSocket.send(payload, 0, payload.length, DISCOVERY_PORT, address)
|
||||
} catch {}
|
||||
}
|
||||
pruneDevices()
|
||||
}
|
||||
|
||||
function startDiscovery(onDevices) {
|
||||
stopDiscovery()
|
||||
devicesCb = onDevices
|
||||
beaconSocket = dgram.createSocket({ type: 'udp4', reuseAddr: true })
|
||||
beaconSocket.on('error', () => stopDiscovery())
|
||||
beaconSocket.on('message', (buf, rinfo) => {
|
||||
let msg = null
|
||||
try {
|
||||
msg = JSON.parse(buf.toString('utf8'))
|
||||
} catch {}
|
||||
if (!msg || msg.t !== BEACON_TAG || msg.v !== PROTOCOL || !msg.id) return
|
||||
if (msg.id === identity().deviceId) return // our own broadcast
|
||||
const previous = seen.get(msg.id)
|
||||
const next = { id: msg.id, name: String(msg.name || rinfo.address), address: rinfo.address, port: Number(msg.port) || 0, at: Date.now() }
|
||||
const changed = !previous || previous.address !== next.address || previous.port !== next.port || previous.name !== next.name
|
||||
seen.set(msg.id, next)
|
||||
if (changed) emitDevices()
|
||||
})
|
||||
beaconSocket.bind(DISCOVERY_PORT, () => {
|
||||
try {
|
||||
beaconSocket.setBroadcast(true)
|
||||
} catch {}
|
||||
beacon()
|
||||
beaconTimer = setInterval(beacon, BEACON_MS)
|
||||
})
|
||||
}
|
||||
|
||||
function stopDiscovery() {
|
||||
if (beaconTimer) {
|
||||
clearInterval(beaconTimer)
|
||||
beaconTimer = null
|
||||
}
|
||||
if (beaconSocket) {
|
||||
const socket = beaconSocket
|
||||
beaconSocket = null
|
||||
try {
|
||||
socket.close()
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
const discovery = createDiscovery({
|
||||
identity,
|
||||
tag: BEACON_TAG,
|
||||
protocol: PROTOCOL,
|
||||
port: DISCOVERY_PORT,
|
||||
beaconMs: BEACON_MS,
|
||||
deviceTtlMs: DEVICE_TTL_MS,
|
||||
getPort: () => httpPort
|
||||
})
|
||||
const startDiscovery = discovery.start
|
||||
const stopDiscovery = discovery.stop
|
||||
const listDevices = discovery.listDevices
|
||||
|
||||
// --- service ----------------------------------------------------------------
|
||||
|
||||
let httpPort = 0
|
||||
let httpServer = null
|
||||
let serverHooks = {}
|
||||
let listening = null // promise resolved once the receiving server is up
|
||||
|
|
@ -1686,6 +1212,9 @@ function startServer(options = {}) {
|
|||
}
|
||||
|
||||
function stopServer() {
|
||||
// new digests must not die with the process: the throttled mid-sync saves
|
||||
// may have skipped the last few units
|
||||
hashStore.save(true)
|
||||
if (httpServer) {
|
||||
try {
|
||||
httpServer.close()
|
||||
|
|
@ -1705,7 +1234,7 @@ function stopServer() {
|
|||
function start(hooks = {}) {
|
||||
serverHooks = { ...serverHooks, ...hooks }
|
||||
return startServer().then(info => {
|
||||
if (!beaconSocket) startDiscovery(serverHooks.onDevices)
|
||||
startDiscovery(serverHooks.onDevices)
|
||||
return info
|
||||
})
|
||||
}
|
||||
|
|
@ -1721,7 +1250,7 @@ function serverPort() {
|
|||
|
||||
function getState() {
|
||||
const me = identity()
|
||||
return { deviceId: me.deviceId, syncCode: me.syncCode, hash: HASH_ALGO, port: httpPort, devices: listDevices() }
|
||||
return { deviceId: me.deviceId, syncCode: me.syncCode, hash: hashStore.HASH_ALGO, port: httpPort, devices: listDevices() }
|
||||
}
|
||||
|
||||
// The pusher lives in its own module and is built with THIS module's helpers
|
||||
|
|
@ -1810,11 +1339,12 @@ const exported = {
|
|||
forgetDevice,
|
||||
respondIncoming,
|
||||
sameCode,
|
||||
HASH_ALGO,
|
||||
HASH_ALGO: hashStore.HASH_ALGO,
|
||||
hashFile,
|
||||
hashCacheStats,
|
||||
resetHashCacheStats,
|
||||
forgetHashCache,
|
||||
hashCacheStats: hashStore.stats,
|
||||
resetHashCacheStats: hashStore.resetStats,
|
||||
forgetHashCache: hashStore.forget,
|
||||
clearHashCache: hashStore.clear,
|
||||
HASH_CONCURRENCY,
|
||||
progressGate,
|
||||
filesPath,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue