From 0f3fbaa418de71551554dbe6989cd2446acd2fe0 Mon Sep 17 00:00:00 2001 From: olcxja Date: Tue, 6 Oct 2026 07:28:47 +0200 Subject: [PATCH] [TEST] try to improve sync --- development.md | 3 +- files/translations/en-us.json | 5 +- files/translations/pl.json | 5 +- renderer/app.js | 30 ++-- renderer/plwl.png | Bin 9458 -> 9536 bytes src/lansync-manifest.js | 92 ++++++++---- src/lansync.js | 128 +++++++++++++---- src/lansyncpush.js | 256 ++++++++++++++++++++++++---------- 8 files changed, 374 insertions(+), 145 deletions(-) diff --git a/development.md b/development.md index c4cea56..8234e8f 100644 --- a/development.md +++ b/development.md @@ -72,10 +72,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. +- 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' 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 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 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). diff --git a/files/translations/en-us.json b/files/translations/en-us.json index d16e330..3ade575 100644 --- a/files/translations/en-us.json +++ b/files/translations/en-us.json @@ -205,8 +205,10 @@ "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}", @@ -223,14 +225,13 @@ "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}", diff --git a/files/translations/pl.json b/files/translations/pl.json index 58806a7..71122ba 100644 --- a/files/translations/pl.json +++ b/files/translations/pl.json @@ -205,8 +205,10 @@ "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}", @@ -223,14 +225,13 @@ "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}", diff --git a/renderer/app.js b/renderer/app.js index f7fa768..4bf1313 100644 --- a/renderer/app.js +++ b/renderer/app.js @@ -1405,8 +1405,11 @@ function renderProgressBar(bar, bytes, totalBytes) { // Older peers only send the overall byte totals - then those drive the bar. function barValues(info) { - const bytes = info.barBytes == null ? info.bytes : info.barBytes - const total = info.barTotalBytes == null ? info.totalBytes : info.barTotalBytes + let bytes = Math.max(0, Number(info.barBytes == null ? info.bytes : info.barBytes) || 0) + let total = Math.max(0, Number(info.barTotalBytes == null ? info.totalBytes : info.barTotalBytes) || 0) + // 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] } @@ -1569,10 +1572,13 @@ async function runLanSync(device, code, categories, force) { state.lanPushing = true window.plwl.onLanSyncProgress(data => { if (!active) return - // the file counter is cumulative across units; only the bar follows phases - phase.textContent = t('lansync.phase.' + data.phase, data.label || '').trim() + // waiting phases have no translation under lansync.phase.* + const phaseKey = data.phase === 'wait' ? 'lansync.waiting' : 'lansync.phase.' + data.phase + 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) }) @@ -1635,6 +1641,8 @@ 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(`

@@ -1724,7 +1732,9 @@ function armReceiveWatchdog() { receiveStall = setTimeout(() => { receiveStall = null const parts = receiveModalParts() - if (parts) parts.phase.textContent = t('lansync.recvStalled') + if (!parts) return + // silence also means no bytes: freeze the bar and ask for patience + parts.phase.textContent = t('lansync.waiting') }, RECEIVE_STALL_MS) } @@ -1759,16 +1769,18 @@ function onLanReceiving(info) { function renderReceiving(info) { const parts = receiveModalParts() if (!parts) return - // a ping keeps the screen as it is; only a reopened popup needs a phase line + // a ping keeps the bar and counters as they are; only the phase line moves if (info.stage === 'wait') { - if (!parts.phase.textContent) parts.phase.textContent = t('lansync.recvWaiting') + parts.phase.textContent = t('lansync.waiting') 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.recvWaiting') : t(phaseKey, info.label || '')).trim() + parts.phase.textContent = (info.stage === 'start' ? t('lansync.waiting') : t(phaseKey, info.label || '')).trim() renderProgressBar(parts.bar, ...barValues(info)) - if (info.total > 0) parts.count.textContent = t('lansync.progressFiles', info.done, info.total) + // the file counter belongs to the copy/done phases; receive events for other + // 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) diff --git a/renderer/plwl.png b/renderer/plwl.png index ee0c558dbb65ef4e1acac8fbd50ec8f3a2b4171e..cfab4e6a14d524a387219b57ed55b480d3ffc7b1 100644 GIT binary patch literal 9536 zcmb7qc|4Te`~N*=H1;)njK{t<$P!|-C{faaY$0nTvae$r;mMjxvL&g6kUi^IN<_9o z_7M?^i7>+$zccFj{QmyFUc82T&N6SQ|v6lED!{-pEf#i0fJ!QPZ-3s z4}94AzH|j2%sxg|eh|dbM*oGq_kQUN9t!yzUiQD}?dl)o=<5On1qB_vcEi)p$=i=3^v48s^v49(c1_@uP>y|2j{c5k zwPV}9yGGBp9(>(^jHK>`J%tP;2Fv82yG8*@ERfU%(Rc>v&ZYPNx5umXdQiQ(j?U=E zkeP2;Yh5r~4&w2IU(!$pY?AG@3XUIwSt5;nZB9eX&`kQS6$3E}!o08bO;X*^bn^2K z3J6GibHvo<(jl0GzrVly{H*|JTM-w@!t{#al8QxPVCS(55rvM^_NQF$qxT2u?zdDV(hcXYlj_^qzoB3xjmDViCap=X7 z(UFnW7h_Rng@vq8x@1I~D{hK`n0F&N_5=e-r&(-$`_p#YH0jA$lxkifGj#lT8K+#_ zszy60#MyL&pKsnkM~~C|tYoUOWVR{zQT{PwX3m!dPrs;trMxj9)%*I#8ehq^rgjm* zHO<{9$u0kg@P-yDYLpUAO5`wSgP6e?+ZhqMi^-L+cqJl%K?+IXq3{Bz-&uJ<(( zN7ow6A^33M^QUr~(3jgRe*6WOSk~4~OWi;H?7|u`hUUwRy}7a&J@t~|);@Jj&D^pw zZg%#>R0cc?QK+3%TvAd!#S(D6;)n=zwd}G~>eVU6FPchfo@r-fRCRUZDHJPtaW1NB zqbCa6sO4m!B6v)0|1l>gpE6ekTz03_c6+%?ab9cG>!VGIC2v}Ps_y^Zky4bpyzE!w z*OLVO(0PEvWScsMh6?{8OU=!>`sQ!Uogl@B4Wp=51PUeO%(-)tDUa}lFSwZy{0Ajj z-a@uBYmPP&DF?!qYwernG$xM(>{C(SU*9^vrpG`!_h2-~__@lVLy67kK$7r>9pS*bSCL6om z+wN}vf{7YL)SVe8r%#GXLE=YC7;*jmWDmpX2&m-qL%*IBqce^k9t;bW3I%^?*F5#t zi7qRXYi~c9TWE+yoge(&m6!J`nJLKmdQ?*DZrYfS*M;My5jP?wQ?4-Bi!^wI)B zc%#*VUewjoi%v<2H+nv|9+u|Y6Um!sT6}*fmswj3I(brNSF!mqlB@rr?EQ*Ij|7h^ zq)L#Z!+}eA*OENFML9Ur8l=8MpWO~!Kdg8P4r1%q&2|z3hwn@t84-FrbJM4X8T*et z(9oC|x#EI(Qhz`9+nM`Yi~qEfW;=|+8!KUgcb#Ak&dyian;*maUMc51HWg3#7cNeI zw?`#xdf4OY)dRe|su4Hus?sj0>r8(+ej^Eio1X48PbhPP`FnKbzxetn{WQeIc!};0 znM8rj;vWLZ=N!Vr&)AQ1yw1<-FGwBOSNZRCPKa)fr^IVentNA%KJTNahnO}qA~Bl<^cwr)qz4N=;XlTzrLkeG;D!}GmX*3T)Q2O;R?IQTFhCYe&R z0(}YevI3$}!|xFVR_W|FGDlhz79KMqT}>7^miyx31>GoiH>m8gQmf#$ zte2IVc5JP7zZ}yHr!{~9b;ZlC*O%C>t{tvH9h%dmzLC*Mgz>adSyG;*4 z_!kXg%1)Hz_Wp>z{)@<9xvy)|y9JjNxlgl0PEHpsVXqv9Feygj*`_9ow}5hF$nPS4 zMx#287z!pY4I^Lco;)>E=ayz z8wj@ApzEPXl$M}&+K07=A-u4VcFs!jE)R6_)Ty+C4P~;j+WE9U0kXWrLA81}!@E^o z(|)x~TJ7XN($Gr}2UhjcC1YT@%uGIi`eAeh1x+K>kQXOdTH5_HyzZ&;$OEF|uO=R+ zr*r#D{av#;-J1}Np6%{1F&|M@J1KbpiivfyJhu68()q&d_j4dO@>0JxP2w;`CnV1I zJyZ=>@*A@|(+V><|6r_gtkapN&_g4WuH>_%`0Y?uEMzw5e=zc7)0qj0!*cRzkVutC zMTJH3|CrY!SA5VvS=#S*LNA)ZK#4yLrCz%FHOmP*X$*lV+v9tsmD8DL;rpce@1y+! z@l1@kOU4w|pBq$a!+tI%Y{j1DYk+KNu&pYMLY-X!mqL+y-=z(X&c8$Jy_EVJb?f@n z5!Id^{vNnbPXjL$SCOtH^It%UkzxCbOC8xF<-cCnh7yvJ0^ye*qo-vUpudt>2vD=! zu4EKtA}?#JaIGkKxSBnLL-UDgo;4Q>)Dt3d{~E(4g5*X+>-TN#knawPAs8Su6SUy} z!KjEw-{O?n=s6H^X)`9IE>c&-0A+E0G`#5;|3@Qd4gL3-z}z*CDcC^-%&t1T5W z^BFA*{!bB8i?Jx^Ugy(V&3QQPDM{>SAO!*mV$UK~FbwqUlOgO${v|z0o=5PGd1~+V zG-1`h{10(Z({~K{tW5?FNu-$ISYWun&g}_IBrR4wbA=W9PlPZMyf2>>*=)-UDUT+G z|Bw3UZIoQVK4Oow!SQJsWDB}&OqVpaaGGQjWj_^oG|}SJKffC3ht$*%(Lc}b$yo06 zM#gz;sfGEEmC!BT7S#v{KlKV^_GJ^2P0DW|K;}9<%9yc5_k^Lp=JeKUMSJ@U0QGwh z(KDtr+0bQ7>d8$IXH5u#)O;Nt<)a62{r6uKs;}YWa0va|o@T<0iO}n#l|n_9r7$Q< z+fNwD@#B4eUc|oeM$X;3fTgmKwU3OLDcVkQZt03)e9|D$+-r6b&C4EwiKhGs9?s4k zCp>{;3=5H0>XvSd2aXNh6RQux;B$7V`SIjX6z%8OO8Zc;y0od|`kZmPYiEE!OatvzSXfFZimO%aiuP3pSB zi+(X!G<->#nkBMJNg#jIH|M6}oc635fUPFv<~sK8fU9BO^csbYlAba_+78#heXJ#} zN<>640y)TvAsYQT40zN9dC7uF7-7o0qxWUjn?V641Y6Z*s*BbjJH{bwZ`H&ty5Bv^ z4P|F~Lbq_Lj}$sP4fRjHx82fx$xeoi z?M2bjL8?I~bxm4&hOK#l4?Bd0c_6XP5C3fD#1O}Ya3Wn{#dZvz`KdU%TY&Td$2OCf zCJ)`m4b2XXV)NXTDHTsfqfoE4WgTvCws`jq>yqH@varN)Q$=Yl)Q@W(EJD5?!jV+E zFRz@6Ree)eer+wubmd~VNH6{rcSO7>0T zW)suX+^l++kW`!}aFMhbFBA}1UESGT!{OpZ-nx28>F`Eizu&+8^9l+It~P?omyMl0 z!@m^dMay!AFeGiM#3nZL#^~+AB1<^blcgOMh2$hYPcwF_*H~0k)1R5#XypWnP8be{ zU#{{nH!S_uWjUMl@Jb8{vUqj*+RMqXmBx5~9k15fIwJDO16~!&wtvVg4QmI-BX>7_ zK57h$3$jTFvVG0)ENR)OYo zgb7=3<$RMx=X;u&@Ixa~dcvn|o>UXLU1jL{rfb9Fy3aT3Fa&TM=yKy$+Z^)J`HhB> zuJ0?mtS^ql6w)UsuW@unD%Gx~p;L>88Qk__H!;f2$ zo%f!9|M5fZ^VdznZiFELe@`I_Bq*AA)xG13 z=TySCJ}rsuE1$3VCqzXF2w1V$iICt2XX!G6Wru@FxRIWlK*%=Brt&XY20vSJ21hE~ z*ai=UdNjWI5n-~Dq0!FJyznVsaqC*dyIBq@u1y=}r(t^=1zia(;N0F^Ii$-p9_&M_w&RC` z&rISL*cpTBXgE0bZ%N&vg1Wl8^^bR0vWT$o%|UZZA&-cy-zy)s79xtf?8vZKU!baB z1no>j5)u{Oe1YrB8&zFZ_quD-?u8H*3SKI%Ff*^Z+W6vyO!)6SOPlX4STc<0gxY?T zE8U#s*RZo;BfVoxvA(|z_&JDJ<)Kmn)$q6~IWuJvJbfGoTn~9XptN!0zRW~6$Zqui%D`(yY>u5RiOwA)+C9?WZBW*-eS@mIofQ6!!IPSIi>zE3Yd9&JP%` zYWRv<$G5k4oD8w?l#Kg9MC5^B!HzD{e1zIjsKa(9v%ftt2L=HzZ;TldxFku1-rwPU z|JtSNVch!EE{&6zYI!Afh#;a9_(O0JhHdtx$j*M}S(>e&P-;kqopWIfL)Te#4-TZ+Vv2mV6<==>d(-@mkcphP@sQb~?-L!625;ShHT-H{ z8rWLPxDvS&z{YU}VTK~XOI#T@4pOZ!{o+2odRtw$7NapmZaF)Ifvw*eSV~tU(hNg_ zi-LM20#0|u&^w|09IWxEV8he= z$6H!QV;7@Pq{|ojY|kdQo11Z!Wp9dGTH=GJqwOtAE?W$OOCT$;2=SMOuVyE-PDW*`GC*nSUL$BF3uh34HV& zokw$)NSdD}(*gCLd7KP`2n@u=!9lzu`}pqi(gO}ro!c!(orIFt3Ps7Vw0l5B7TQ#b z8Xtx_%}FB94B?xTMD0s-$YpoCoi>;4_|N4|41WH+Uo6Bqt<$245kou%^z~8r7G|ka z`#qfVGAn~k^>{+S(vSeT^DdAHRPAk(jme{Zv5lh@Zdx)j-AdE~JrKimYYYieDLO+@ z%`eVo##2wsBB)lPhhK9Fpe=gqaPb02A0jTjl0?6-TQ3l;qcQfw)?9bjE+qaCgz|wi z0?;CSdu-QCBbjA)21E1i?(Pt!cM}=_30b)SF-JnP=8rHX@>*&+=2PB6hboot&ND#?ZdG z`>V&HsKl^+p2PZ7T>b!l7)^!FT)EHR4 zD^nztez+ip_AB{JP1gq#l3SJMu<7shM!BHF42w%)-1UJ!)y*oJh)COWtfr)?ps66G zb5JNZR@|>{r`UT(uO|nC_BP@Vpix7@*D@pBh~>{{VDVY7cvyno3EsRug_Wg|-v_<@ zs8mQGpn|3%(rzu$wr-<*l$8&+*mr>*RbM|&RJXdm+{oR>hiR;4>W+KmfZd&~Xq1mM zeF-P%j}NP>->zTw-8e(c4x-_7m%q4*lRj%vaWmIim}#MSR7V1(adYxrwO7sbZ0CJ7 zUHO9hvi6`XuDR+!Y-G`3#y01H3q+ZTM(rrv4-f`f*AIE&mnO zZTdI($c^BHclr-QOKsbtK!9{JKk__Enm=XnjpUYU{M=_MV)JW~Zgnn!U~gS2faY-6 z-lzu;rwkQR)sD13)eG2%A(nv>fFBWwb=F`YSb(H1Tz$QkIc(zyeqrHSm!1sIr!S3R z0yi3dRU8gkb-C!IGHqHB;*La`oPEPZMeKg{@aA;-^x`b}huf~M zj*bp{m5^xS*G}1-w=GfIs;a7vp$Udzg@vSFbYxRM~U%ZI`Lcn)A9OrCqpL^JkP2A4cz zOAbn?1$`bt;I6&Lt1^ry_9iC~)8s%gYa6T=d=f!$bw&_Qa8q4p=?$`mq$F87j;N)W z>_V?HUvBL+OT`_>t!^=6wdu&n637vU{`X%I3>!jlCaC8d&T4>4>U|R02EW^v8x>5v zwZcnfz^(#V>wIk%-xpExaIsCYvds<62^8qnWP3s5`j0>bLp>aGH-a-q8}wRxi>-X+ z6<>#eTWWF8^UQh>ndFGK?>krGoj%Eg{p9)}EoQQ#hss1c5V z4(N@>PSm61>m#R{8vp;sR$m8ETnz|3H_W#DWcb=ydI#t^{A9?{hNR4tB~Xa>YV%TCCCDE?41zXfK5Or6i`lYDB7(q1 z7?;Ik?4yWidCupIvSb&~;&D?c!W!U4U+I$sF^CnScZEzCT-Rca^NzR5xaUHC3M3EbTk}x3|#-CP$Z0PY$8SL`s}Pn zy1u-j+`{?hA%7MzZ8~jxmZY)!P`Re#d*8)6%3QDItBaoC7Jv3vj9QiH8rP4hYC7u= z6n4&3F=2?o);<7awqL$)Qn6!k@p*_jHi7rOO5pwHs6Zrfn{Se=a)LJ zfB(q!54nP#q9Y>E4<^oag(Ys)X%#GXK6?VStOSs|LgGSrsis|ir;oxwK1iE3A&I2q z-@`K?FaRs5sh&=wY|f?A7oP(%Ns~Wz0G|OS1BeM77kSJC(HHKaB6Q~fKp`CySv8}# zhc9xD(9u^*P3kI*$&kUI+~xx`G{KE1 zu(pR@YOg99b64Li+W8j%-=;!`_`#RptQzbz63RK3tQ;f_?azX#Sd7H&(OeS2KKU#J z`AI_i1FgSSJ~Q~s6o&bpX=j+^5wH+2?X6*e*I$nl| zGVHXBbO}+pdjJv^si3#U3zlZaq|DzXpi+A9p^3V5VLicE4tlBZc=(2vy1MWcp6e$N zgpvnDtpz(9l42abz1`tx`c!G102CTJN<$FC12a`)98}&IvCc@41D+!Yr&mTnyPuJ1 z^BeiaZ`y}E;k%w_bQlFWku84)s(exMI89nI*<%Pa;FZH|TAIya}evq;d+Nk(z0+1-5radE^VVfz~#^bR?m zr#uw@kFF-5J2FzRIo{Zu_4=6$qlNi6t9a@J2ZF&J2vW)Z-#UHj$B)HzeAPyL`3gO%D^4d>uoXYf#j?1avQLm zb0K^NH~3XjGv4?08T!ck{{cYXlpZ1T8Megn75(Ooytq(VS6!n;efS{Bzx?OV=Elb9 zDZifOp@)62uzn>jO~DK^|TB z7B0XZ%yKVa1vXQeF~GU(z6uG`d@L=Oj>?OsrWAuW2nH5L(6Hc(HbVmgahnvyZU{oO z(BB29u72v4$^P$N{qZj1TkF)wNVLyo!nihiF1x9_z$}$?tku!HfFJ8H-&n4=_Ve-{4^AK4K)hc&@UymzVwF9eFKPqrV#W zolZUx%5~>^GW8Mc!MUJ-s-}OQAA=+mQzF<`tvI2@Y zkjnv&Q2>qQRLv}J`1=9Y0&r8GC(ju8PZ$mX~(;NtGd%G%)d$6m$% zsy=CK_!x*-2$xTWpQ=rJb-w@V?vjFeyEJ`7nnu);Y!|e)9xZe3yl}fs^xKuB zHJ)agjnu#f{#%k%*Psd1SXVzbwY(HY)9D%c**75#smZt3)zyIw(qPQ4xF$^>Qh|$+ zvV#Q&T&`VJQ?j~1Rq4f_gs$s`{-&&_7%gd=h>l2&i6Qw1`})r0_oi)eG<%;)H+ZN0 zetcYs=Z-w+HF0Z|t~m+kxd8kN<|Kyaxmc>678moiIILauyQJVmcI&yntH^UFA@e|? z(!y`C&btkTM8Ab;0D7dv~t6*~8{`jHp>gJ}ZrNvs-00&s^%3^GeYD`Sb$=k<1 zGVMDKeo|##m_z3j;#QSMw5iVEg$eN@cmG8-TqHZOhCwdm@pHcU0|lRt%+1fI7=M&Z zKO>Q93^&UyJW!YsQ|SAE28Pk!f-T`xwzyW{4!Lq{aZj7)Jq?_k;0(9A2{AGI@`9eu zK}{m03E4vdHQ9w3&|PySew%xy6=$;)bCBdg`l}*e59IHUURkG*VOHKhZW@~lK$XH` zNp=NtKlpK82-4A-_s0trv*A9E{AibDM6m&dg>alj+2!k}6`sFBezCvVv&-rfoHOPm zF?jEbAr8ezj(n*QR-K>!^W2SOqZkhv=(`3On|XMAO)FA3h$Q7kN1EKVa{tUA-av}% z929<5cb=ztKTc3B-k%5Bm!jR8bJ2XmLLu6(>tR(=-QkmCvs=an^_3)6-D*c+(DX#f IF$etr1LCoDGynhq literal 9458 zcmb7Kc{r47*nfv%ETQZ>Wi4b+BBMo=5M{|WgGwbkk!1`e$6m>jJ+fzOvBu2gWT{Zc zzRoz7DeE8%2H!Kz>3siu|9sbl!TUbX{oLE{-k#_yriL7B!fX%(aa=SyZvjCt@GA^r zVFn-eKE=1d2dkISbzcbLY@vU{+B}PIfPV`48QAz;^}OX5;N)`?3J3@|e#gVj*Tu=} z=5bFS*NjD7VF;3hE}lPgEiiK-E7(i4FKY=`1(@`l4QuM9=EBRlRQL%&!g6_+Oi*0o^Z9>RsAK z(iZ-gmV+6jxjT^N&=Vs|MK*}LB*p-yf5fs{9x{4$jGqaLzV;~(l6+cf!wX4U@WnD< z&zCy?-!EJBF0s}XR8&+v$Q*EN*Tx;hM?!k-f=C{HM!X2re*n#iO{?UQ*(~S9N@HKn z?(<3AP_PA{(NIvisEG7#R_(9rd_Za=I_0orAS$Xlo7EQ&Q) zBw>`S819kpJ*%U37&D{C$4)^F4HZ=5uFW;50`3(~Ts>@=yo(pCQ7_aT?s+?@iIbbW z#wP#`1#G1!LpV-+)J`^6G@QhTJuMtxnh~%xe)`A8jrn3{26V@>MWi+FbXnX)5TrT1 z<;z)oS2(@!NRs)>(qq>&)ptar3h%V<{s`GZP3(4A2D!R6PY1kNatlScjjxAQ4TWay zFh8jryLRWK%S zG{y&k<&*dQu9nmxnPqx7`O@K}i!H<57bvm8*2ig9e6fAGH8q^?-@h*|DNzz&5yrX- zwrB+h2kcngMyK)(LO&K4J-TlUIzV=exl?6XmkLz0wWIxlEZA7!IEN)li|6{S1Any5 znWIpN!^6~jjdvjNO9g`i8;fHPECRWY%X`7_gISkcU6paybYlmV(UYp*z2-bI ztFhjOGtZ{nAVr4;n!N#`2Rg2X)Y8<9TwXRWF75FRdCP+F_KYA_0Nu@WxnIzAM9u%S z=9Aef$<8KovU41fBr@cd&Ba7!x^*+m-*eF+1Bw?fnd z>-l1w^@P!eLM?JYTg}JZRYX=~0Hew{UbGqGZLONJ(wORMryFWQ5vcZ5lE6kvTVyjy zkw5ljxx0AY_Xsz}bJ&%i=ba66-knkUybAsJVY3?$H-UsriSZOzBngU$Fp)D!0xL4q zPfgP;mB!|Rwzk8uS=@2H3EdlO3tI*62ukfeP#T6zWXdHB>S=3--|#sA5tL_)fmj<6 zICEceO>T8{$EYqrd;2_!*!Ae$#ph35Vc))cZyss2QN~)iY*Pcdd3Y{Dd01hCeO>(B z{B~ROMAxU$`p%gdC#byV0! zM97~g6e)}wEAgt(V_>9Z`o<-2^3Xogj}$;5As~~S<5?-IcdrSm?cdDN-F~}c&pyqK-cA7=M$GHrRMOj)#5TEW# z`8YLv-dz7w)`=5+rN`_JyU1EFU{&qiMzhn>4*E$0Cjw_g?M-xdMM}N+y7%TuVfEaN zuNO`NZWfwZ3LAeFS{zy01?trTbVUwO{biCmK7x!zo|%t0mHK2iFZ~`w})a^E|Gpvz`WLVNW{b$66QsPAM_L0c7oy zy9c{l!O{VMhYLH&ha2}b50H2u4o5QWN_oVu~=4|QGlV!;2pepBcl zkgoHb#&hMOLpfaP?Ef6{hWz~fI~l9cg^?pH(7&cpu1MAH9@iM~6C-6BD>vnpqqSmv zL?ICFZ;y7}xMmZ4hVE$K<=Pce_b!ThK_vd~ho*)HS)gnIs5HI$jrAdkB6ayiIRHXm zmH|vtr^R1{bPfz3c^bmMU^FF+Jqq=zjs4!_%>;st5$j9kh8{$Ty+r`<+E=-i9u0bO ziO=3`RS)8gBwHVq7y3_7oIu`|>mwJh4g-bE zn=2cZ^pKP~Atp*4`z1}yB6yAm*uFP)J-pmZKTbduiGJKKlrN`!TpCVHeeOI1iy$`B z)nrT=NmhXAQ%2Tgsh@yz$UPGWD;(&8OC<#yEBWtH*&)n>q$K|TeJGIZCIBms1zsvc zmC?55qhXi8%dy+sv}c)RoEW^97HSDV^>z?rn4zJZ)oYvRNOl1j#3?$MsAP8{zf#$W zphcww`1v_ce|B5NGm&y8n$Sk*Pvj68B#9Wgk9ixsS#DXc6asq@|V-4 zl2lkX!~Q0_cf(9fI$w7)>g$RdN^@6Fxf*)3NJO8%f|5`kX{@Okb-PkeVa$fp?P=~b zu%!;0qidY!^UaMhR@0vhL#u7R{w4mFI9~|t&z_5JA_MfGlzi#>to2nF4q1hjSznBT zp%(sLZxuAfk6)Zb1VT&?(Ln;yl38U z(i$6*Tm(&oOL)o9?G4}#;sdZLmj{(^uVTFjTt@p;wFD`Q{hk}yO)NhyW6Bq5gL>6` zd3p7yuv9mOne4+1HXhz zK&Cl*?j<z|OB&)w|KOk)R{^5*T_2JJ4xD zZ7_uqU+vPSSkC;qtgPk3g9U>Sx*WddAi4bg{cmoApe(9$;&xAy4_H#dCMrrzJ-&d~ z9C(fY7|9NU5S*g;{&Kr-MAV1ja!b_H&FdTY1u)PROFqK{Q9?aA+pgb@5b@S`Ddo&# z38+hNThw@GJaj{>xb$vI{TmX3;Tn(1K#LUBvwqrYG+Y+_{{Wu&4eHEl&G1oeL!J^m9VDGO-$(Q z3E%?Hbu=qjSA0b7PFgEQ>Zbu$1M8}%>d}NDJ{S_c!$Ml9Esx9@pnM5Y3aQ**r@aUG zl5gy`v<`iFkL8gVFsnoOO)?7I(cL^ntR{b0?W*nV8SZujj-xjK|GGw@M&w|+P^Q)n zYfQ9NWSD=7!l;bl`xEsR!3a389!SxidU%`Xm_mTZz|FOdz4ei>H&dwt7H7cve){@} zySS{z!=W?k*8Gq3-{ZNUC{D3Ay_MzVq5W&n`ZOoaHTwa-t=)*?!tyq8ulY5~(KwR8 zMco%#QQ{V-xbAJY-tLf@&!a!}HGz@KnPmfQ_IqsKDu3L}S09jXTIMS%s(b2-#Sbq$ z7xfBV*m6`DdahsTKB5iq3EQJ^y>;MWk)Snz&Cn$|edp$erWYN81eD(JL)#lm%5$#2 z_>zzmM)7@y4`HAmOJ?ZR8I${R;boP_{hd@kP#E75o>WG%EBokPuv#SMZJ(0>YZ08h z>`lOn18wL;U8(-K?WMg#;$a~fV;!W~vQ7c!77pZ-CWjh6CC3kfnwsx94ZS6H)6;Dg z9X|~IHNL&ESMe%mXrP`JDM2ZgMOPT=le*W*R07wcFE4|&V+l&|U6b4Hy<-pt3fx(H z_)Vky9tY#I&U0Qyc0nUYFLfqc7`TSY1b))Z#uF7LK*1=byuH==aU%SpjA z3-k-{TsPbpo7eg!6xSTyOsQoojsq>EI@J96tmWmoD3w#EY)RIi2bv>bNRl}OKeYQ| z$KWjKZNk~yXQY8pRG?F3L9@h8aTY}3#6UK0Zj{KR3OEYNQ!+XSvLhG)0HuaU9N3F9 zI|0YeYd4EIJeYQcK^x=2wFPTT$t+0x@<(2TCq4M^ zl@0(b$222T9k2s)bi+#6=8?39N)bbbWKLX#fa9Q)dtC0_ z?rd6)>Q@S5cro3_U2CQz^=(K9Y<4zkXiMCnAlf$pwR5-D$!^d)rFv>hJGt93{2^lR zvQ8TphI0DlK`17R^BPEmy{6g^eb|~+_^XvY-SNRR-vLf>rYm9qz^s@!kJB1*ZnKPf zM{WA$`b-S0jrw+95>dasHLbiSqQ8dcg4fC4t|&62bpJ}?V@T#gLi2;m$P`Zz$BUnD zs13e%GqePz&kZFr?p+QWGGtiig9A$r3=|CBXqI%R<=*bjh8^Iz-pFTr$C96>55-2z z1Tn*b!p@%~xjjqa*hb(gigGZQhD9Utin6yNbI(l^xZtk=O;c(Vek4@{OT)X*^3tmirQGX(5dHFazf(2ME3eRJvd)!?Q{FdEj+9>b#j{+AP0}&L_*R@7j;*^_E>zN^>m*k=pMn)XHM*T0a540T)iBYjs{xyDn27$u zuXA&v1%&Mu3sTibWY55LmiGFR$9f370?@RxGXrs}KZ|eGaw2TIcJE7Hy=~vn>(x>)gopPd08^Lw{u$OZ555(Vr~x zPk(&cTI0I@V3|>QX&i&Y#N$uO)M+iKkHsy)V10ovYukPxYta6y?Ug$3dKHEY_)LsHI* z>ya&$3P~NL?BK=LPOVA;;H4lw=+}=!uN{IF7VF8!K?CWFuVP@~XF$s8H>oFh&)zS} zxH!wQ3W8LkZrt*+cbk-K(&S*3(`lf8-@#hX%5rUf-K&ulM#y(7YG^<2ClH(k>{&1o zVfQ<;M&f{}l6-Tvb!}~Jl@@$grdVk-GP13|JVRAi_g-)UQ};DV8aZb#{KLX|=hXZR z(>g&2jwE=AKZ&I}%=z4@CB^$1iR3r1n3Ko;%66%lsp794l}5Y1}3Wr6Va2$e%!iYgOBs+B^4 z^jhm{94S>ERKW~Zd}*ZA{Lc^fOQ}3)KQa9Slol17zz_Bl&HcaN)ajjP`F~~JGXDm! z-ha*`0KNxw;Y|!B%21hTQ=oU1w^Dzs3b|h_^_9B0CmMpfs$&%tt?g~4_+32LkBy19 za-cXc%LWYwV?9fzbuJJH>%5+pFV}byO)2bK9cxOe!URt{y=zSC0=uuH)^_~}W}{{m zp3&b~For92$Ex-~4oY~4P+S|SaaSNH|BxG~{qr^)=kmaY?y*a9(HLFNY4L(oN`Bi=olqQUlX_UEUu-mV-j>c#ig6HkSRI$Kd-~+R# zu;>=$FUrJJQYu;DDm?5cQ=u$+d>oEjyi8$_CFPie7qh0-Y7r`RKVaNGghs8mE)A%cLq20I3Pn_f#Fp}Gb=;>0|tbke6gvKq}N@z_C( z?YLW?HAukZ49@r;qG^+_;E6*+k-+?0@(cuCHtQXCK^t!^c%lxi3ra||?e1(Z54t4g z)z%{VSdb(qPy^AI>=h`Nx1?)obqIbF=o=WAd}e%{Ad_kz5hcDQgrQAJ*p6!{f^0C0 z;h)z51j(6Zy1;XS3Pb%q zaY!Ah_V#xJHD-?IpLZ@hJ(8e&_`R;EY{%5Nfa!X=>=vs52-AVK{F$K1kQzKSX&8#& zoKIB^%=M$7mvfGSlqEYEz!1HOvI)YnQo8Vo)9ZuzOzuvgB6{_ELsw9`?nCO{VV5J} z-_CLE+zV2|$Cbh``hTB-B9LU!yHx+Nu)W0P*|JJgWw`C5t*szd-&ND0NPV2TB0Sw3 zWao7a2mjbV1+OG%eQ`G#Ne5lhGy4);zO@x#uKC0Ov^J1GD5)#FtFqK4D^g4Z1Hc01 zmQ+%0o<1KF695&|W6!#Y)Xfe4w7gu1UOG&4-6 zf`LvzRjo`|x&|v{yG;4=D~SwRj@zFfqsAp(8bPmCv|f~Fv0^qR0szX^&dxE@m|mM` zi=C5(T36zD=YTB*Gvd4{ErxA@ns7YL|G|?sPHx-vc6InumfUpv=iMZyoOFul)mc(|+yT zpkXOWZ`=%Ilg9O_gQDmr>}+(tus$dsofh&yV@24|L278Fx3}~`j(+58PNAi+Yjll=j;EwBsN<0 zae6upYQI6+Ii&CDx#SHkrO92`=L@)MU*D%rH;?i0#sKogBHu$h$MGJM7fejbonlF; zKT`)Tfk6c)f(9K-6yQaTO$Vwy5ry8r4+1h*+gY(?#bk_w-h_Vj?3c#awYSG!Lf-HVhAyM2-mB_LrDa(YIM zj+&a$Nk}8!Q^jMTudlN~u`csB(M*crk?#@g*z`hopBuNHctO9`PU9rd$|55%_9gG6 zBZ$5Gyvfm^Z-ax+1E}5FjdV4bm3gzuT zE2S5}^)*U8B3>$*SGOb&ou_2{Ba29IFyZvCWqXB34mBIJ``KDBpuqs`A++LAmED9k z-<3M^o3IujJUo1NeN?40v+9*4Q@_S2hh)BIn==@I0ZzCGo_A%a?mIfdofI^n_wYEfVdO#AAa#pd{;cWLThBkw1c z-JM6hI)Hxi3lbCzT<0avyS3H?eh^<<^VO$_?8~Z;aSsfKZr@Hz8t4*Sq`&q2`I)$8 z`{SptUv0H%@T8wYT)X^9yVwBS6zmsY{z|@B4FU7t98A>e;Ir ze0;BTX?NdvHwTOR;U1?7y8KC0f2aSJ;PT{8&b>yew;&5khfO>HP6Y;#!o$aKK z_hK)&04AfxM56g)fM0`2SttNhl}zu}>oaA$=(sUmm8xz3ql|n{%W4rzb0hVvBrc(m zzn6nFv}#%E!0?JbjHtX!4S%pY!?(%F5mr?jK(;!{nEa#+TnQ*WwRsb&6dIZNdguJr ztAv0IC3<49Pvj$Uc=++mtyLe?hrH4x)Kj%@iaw_yW@cttX!@$kwS)I<(gGm}?n}Q5 z!2G5RE+RZ1;`Zc!SC5Xy)zmzgapf}1A1-<`U#DQL$0T1^QK7)^)5Qa=xTa7&TU$2HZ8>OqGWyR1y`vAWKPdv-iHUvLOHifBVLwE`cz7lSg-^ju z)Vrd7pApj=09tW;3hCe$fgW+7b}rR|`K^36Bh)U&s>>m{RA&0(vM^}LfPo9cejNBu z!KdcB+4N=^Rp9bD>nmBPRAm_SV*vin6?VU-I?Fd>4-iLK{w)Ue*}h${^C5?E_U{o8nNzsZ$j;p9J@gVk3$#7hM$;7ytO@ zqpU;&`uGU6_XUN8XL6fD#ht6uoy9J|MxPnYKeOn)dM%t40wdfy+F1i9F5i|~miS9W zJ%U2o@zK%CYKHLrNatx7r>dc$@$&uq+?Yi39mp=pXg;K8S^pVdF$BFq$v8InZ3?$i z9aKUu<${s`eeWI|44*+eF(_qUE#=3AIAT z6ShLeSZ_!;`bIMcCF6>7hZeVD8*)|nbRalp0i?7CA?}vd{}0a$!{9|L48iuIXR@%Ne5Pnx;)!CY3|58 z75SJ5qgTxD-)sKy>CN8VyLVsnzj?txWtQ|w{QB3JN2xP|zI8yr#*5O=R$dkMTol|& zu%0wGIq;%6wbkg6=TrYS@ycu7Fm&(!g$%YvPP8=x6}x|xWE@_V<-2QBBZ|3Mn#YJQ z1{V(0)rAUp(AHjE$S%RI)nL!>_*-hG%D~ zk?D`wVIf{y#24+yHPRV#A8hUAm1!DT=JR4zp`xb_uae*@f6VL9GSXWeyKhit#+!7^ zwoV!j@D77(H%8ObK^IlHqj~NB?>jgDuM0Z;dx(bAy`MF&xe4ID5_HkP^nArxbj1Gv Dt@3 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 } diff --git a/src/lansync.js b/src/lansync.js index fecc632..8621905 100644 --- a/src/lansync.js +++ b/src/lansync.js @@ -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 diff --git a/src/lansyncpush.js b/src/lansyncpush.js index 8b6ed16..01b412f 100644 --- a/src/lansyncpush.js +++ b/src/lansyncpush.js @@ -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