diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-08-23 21:55:20 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-08-23 21:55:20 +0200 |
| commit | b3709ac4d362987a9d025616c95065ceed0d216b (patch) | |
| tree | 32e0cc5cc2775eddf516d114fa9799347a214bda | |
| parent | 012ba5b0cb8c556ce773423ca38d5184b74659ac (diff) | |
| download | meshbay-b3709ac4d362987a9d025616c95065ceed0d216b.tar.gz | |
feat(node): persistent index cache, visible scan progress, adaptive reconcile, and delta sync
Indexer performance work, in four parts:
- Persistent (path, size, mtime) -> hash cache (indexer/cache.py) so a node
restart no longer re-hashes every file — measured at 23 minutes for a
114 GB library on a slow disk before this, near-instant after. Hashing
is deliberately kept sequential (max_workers=1): it was never actually
concurrent despite the pool size, and two interleaved reads seek-thrash
a spinning disk instead of going faster.
- Byte-based scan progress (IndexProgress), surfaced via the loopback
index-status route, the handshake ack, and a periodic INDEX_PROGRESS
push to connected peers — drives a progress bar in the Create Group
wizard and "add a directory" in Settings, and an animated presence dot.
Guaranteed to settle back to idle via try/finally and a final push on
the scanning->false transition.
- The reconcile backstop's directory walks now run in the executor
instead of blocking the daemon's event loop; its interval defaults to
10 min (was 60s) with adaptive backoff to 2h when nothing changes,
reset on a real change or a peer connecting, and is now a per-group
operator setting (signed op + group Settings UI).
- INDEX_DELTA wired up (protocol support existed, nothing called it):
_on_index_change now sends additions/deletions instead of rebuilding
the full entries list, coalesced over a short window so a burst of
file events produces one push, and the hub swarm registration for
public groups only (re-)registers newly added hashes.
Also fixes several bugs found while testing the above against real
libraries (a 114 GB and a 100+ GB group on a USB HDD):
- /api/reload blocked until the reload — including a brand-new group's
full initial scan — finished, which the Electron bridge's fixed 30s
call timeout turned into a hard failure on any real library. The route
now fires the reload without waiting (ops.start_reload), matching
add_root/remove_root's existing pattern; the wizard's own step order
was fixed to wait for the group to actually appear hosted before the
steps that need it (extra roots, GEK), with retries for the residual
race between that and the daemon's own bookkeeping.
- transport.js's hand-rolled msgpack codec had no case for uint64/int64
(0xcf/0xd3) and crashed decoding any message containing one — hit by
IndexProgress.scanned_bytes/total_bytes for any group over ~4.3 GB.
Verified against real msgpack-encoded bytes from the Python side.
- chat_hist_resp, and this change's own index_progress and
set_scan_settings_ack pushes, were not routed by message type and
could be handed to an unrelated pending request by the transport's
"oldest pending" fallback, stalling it until its own 30s timeout and
corrupting whatever received the wrong reply in its place.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016SF6RKNBKg9qejmoMJ9ybA
34 files changed, 2645 insertions, 88 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index ad109b2..c48f011 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -54,6 +54,13 @@ OP_MEMBER_UPLOAD = "member_upload" # anything about key material, but an unsigned toggle would let any member # turn a disabled one back on. OP_APPS_ENABLED = "apps_enabled" +# How often the indexer's reconciliation backstop runs, and how long it waits +# after a file's last write before hashing it. Signed like the rest for +# consistency with every other operator-only setting here, even though the +# worst a wrong value costs is index staleness or extra disk churn — not a +# security property in itself, but the pattern (every operator setting is +# signed) is what keeps the authorization model simple to reason about. +OP_SET_SCAN_SETTINGS = "set_scan_settings" OP_ROOT_ADD = "root_add" OP_ROOT_REMOVE = "root_remove" OP_GROUP_ATTACH = "group_attach" diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 53e5085..e1bfb8e 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -24,6 +24,12 @@ class MNP: HANDSHAKE_ACK = "handshake_ack" INDEX_SYNC = "index_sync" # full Mesh Group Index INDEX_DELTA = "index_delta" # incremental update + # Node -> already-connected members: "the operator's node is scanning + # right now, N/M bytes done". Never the entries themselves (that is + # INDEX_SYNC/INDEX_DELTA's job) — just enough for a presence dot to + # animate. Pushed periodically while scanning, and once more on the + # transition back to idle, so the indicator is guaranteed to turn off. + INDEX_PROGRESS = "index_progress" FILE_REQUEST = "file_req" # request chunk(s) FILE_CHUNK = "file_chunk" # encrypted chunk response STREAM_SEGMENT = "stream_seg" # HLS/DASH segment @@ -82,6 +88,8 @@ class MNP: MEMBER_UPLOAD_ACK = "member_upload_ack" APPS_ENABLED = "apps_enabled" # operator → node: which group apps to show APPS_ENABLED_ACK = "apps_enabled_ack" + SET_SCAN_SETTINGS = "set_scan_settings" # operator → node: reconcile/debounce timing + SET_SCAN_SETTINGS_ACK = "set_scan_settings_ack" # Device linking. A new device files a request bound to a code it displays; # an already-pinned device of the same account approves it. Neither the hub # nor the node can produce the countersignature. diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index d18d9a4..044457b 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -298,7 +298,7 @@ function Nav({ user, theme, onThemeChange, onLogout, onMenuToggle, unreadCount, // ── Sidebar ────────────────────────────────────────────────────────────────── -function Sidebar({ groups, presence, route, menuOpen, role, hasNodeKey }) { +function Sidebar({ groups, presence, indexProgressPct, route, menuOpen, role, hasNodeKey }) { const isStaff = role === 'moderator' || role === 'admin'; return html` <aside class="sidebar ${menuOpen ? 'open' : ''}"> @@ -339,13 +339,16 @@ function Sidebar({ groups, presence, route, menuOpen, role, hasNodeKey }) { // actually cares about. Anything else is "not known yet". const state = presence[g.id] ?? (g.node_online === true ? 'online' : g.node_online === false ? 'offline' : 'unknown'); + const label = state === 'indexing' + ? t('presence.indexing', { pct: indexProgressPct[g.id] ?? 0 }) + : t('presence.' + state); return html` <a key=${g.id} class="sidebar-item ${route === '/group/' + g.id ? 'active' : ''}" href="#/group/${g.id}"> <span class="presence presence-${state}" - title="${t('presence.' + state)}" - aria-label="${t('presence.' + state)}"></span> + title="${label}" + aria-label="${label}"></span> <span class="sidebar-item-name">${g.name}</span> </a> `; @@ -840,6 +843,9 @@ function CreateGroupWizard({ token, username, onCreated }) { const [setupSteps, setSetupSteps] = useState([]); const [setupError, setSetupError] = useState(''); const [groupId, setGroupId] = useState(''); + // Bytes-based, not file-count-based: one 20 GB file finishing last must + // not read as "8 of 9 done" while it is still the only thing running. + const [indexProgress, setIndexProgress] = useState(null); const linkNodeKey = useCallback(async (pk) => { if (!pk) return; @@ -908,12 +914,14 @@ function CreateGroupWizard({ token, username, onCreated }) { const steps = [ { label: t('wizard.step_create_hub'), status: 'pending' }, { label: t('wizard.step_attach'), status: 'pending' }, + { label: t('wizard.step_index'), status: 'pending' }, ]; if (roots.length > 1) steps.push({ label: t('wizard.step_add_roots'), status: 'pending' }); steps.push({ label: t('wizard.step_gek'), status: 'pending' }); steps.push({ label: t('wizard.step_pair'), status: 'pending' }); setSetupSteps([...steps]); + setIndexProgress(null); let si = 0; const update = (status) => { @@ -922,6 +930,27 @@ function CreateGroupWizard({ token, username, onCreated }) { }; const advance = () => { si++; }; + // Every step from here on is scoped to the group the node just attached. + // The node hot-loads a brand-new group synchronously — scan included — + // before it is added to groups_ctx or its own in-memory config + // (daemon.py _reload_config_inner: the config swap is the *last* thing + // that function does, a beat after the scan, not atomic with it) — so a + // call that lands in that beat gets refused even though the wait above + // already reported the scan as done. A handful of short retries absorbs + // that gap without a real cross-process synchronization primitive. + const withRetry = async (fn, attempts = 5, delayMs = 400) => { + for (let i = 0; i < attempts; i++) { + try { + return await fn(); + } catch (err) { + const msg = String((err && err.message) || ''); + const notHostedYet = /not configured on this node|not hosted on this node/i.test(msg); + if (!notHostedYet || i === attempts - 1) throw err; + await new Promise((r) => setTimeout(r, delayMs)); + } + } + }; + try { // 1. Create group on hub update('running'); @@ -942,32 +971,48 @@ function CreateGroupWizard({ token, username, onCreated }) { attachBody.upload_dir = mainRoot.path; } await platform.node.call('POST', '/api/groups/attach', attachBody); + // Fire-and-forget on the node's side (ui/app.py) — this call itself + // returns immediately, well before the scan below finishes. It used + // to be the thing the wizard waited on, which is exactly what made + // "Attaching to node" time out on a real library (see ops.start_reload). await platform.node.call('POST', '/api/reload'); update('done'); advance(); - // 3. Add extra roots (if >1) + // 3. Wait for the node's own initial scan of this group to finish — + // the group is not usable for anything below (extra roots, GEK) until + // this finishes, so nobody lands on a page that looks broken, or hits + // a "not configured" error from racing ahead of it. Can take tens of + // minutes on a slow disk (see the StarWars benchmark) — the node + // keeps scanning on its own either way (test_hot_reload_survives_ + // client_close.py); this step is only about not lying about it. + update('running'); + await platform.waitForGroupHosted(gid, setIndexProgress); + update('done'); + advance(); + + // 4. Add extra roots (if >1) if (roots.length > 1) { update('running'); for (let i = 0; i < roots.length; i++) { if (i === (uploadIdx < roots.length ? uploadIdx : 0)) continue; const r = roots[i]; - await platform.node.call('POST', `/api/groups/${gid}/roots`, { + await withRetry(() => platform.node.call('POST', `/api/groups/${gid}/roots`, { path: r.path, name: r.name, upload: i === uploadIdx, - }); + })); } update('done'); advance(); } - // 4. GEK init + // 5. GEK init update('running'); - await platform.node.call('POST', `/api/groups/${gid}/gek`); + await withRetry(() => platform.node.call('POST', `/api/groups/${gid}/gek`)); update('done'); advance(); - // 5. Generate pairing code + // 6. Generate pairing code update('running'); const pairResult = await platform.node.call('POST', '/api/operator/pair'); if (pairResult && pairResult.code) { @@ -976,7 +1021,7 @@ function CreateGroupWizard({ token, username, onCreated }) { } update('done'); - // Reload node config so it picks up the new group + // Reload once more so any roots added at step 4 are picked up. try { await platform.node.call('POST', '/api/reload'); } catch { /* best effort */ } setStep(3); @@ -1105,6 +1150,9 @@ function CreateGroupWizard({ token, username, onCreated }) { // Step 2: Automatic setup progress if (step === 2) { + const pct = indexProgress && indexProgress.total_bytes + ? Math.min(100, Math.round(100 * indexProgress.scanned_bytes / indexProgress.total_bytes)) + : 0; return html`<div class="page-content"> <h2>${t('wizard.title')}</h2> <p class="page-message">${t('wizard.setting_up')}</p> @@ -1120,6 +1168,19 @@ function CreateGroupWizard({ token, username, onCreated }) { </div> `)} </div> + ${indexProgress && indexProgress.scanning && html` + <div class="index-progress" style="margin-top:12px"> + <div class="index-progress-bar"> + <div class="index-progress-fill" style="width:${pct}%"></div> + </div> + <div class="index-progress-label">${t('wizard.indexing_progress', { pct })}</div> + ${indexProgress.current_dir && html` + <div class="index-progress-dir"> + ${t('wizard.indexing_current_dir', { dir: indexProgress.current_dir })} + </div> + `} + </div> + `} ${setupError && html` <div class="error-msg" style="margin-top:16px">${setupError}</div> <div style="display:flex;gap:8px;margin-top:8px"> @@ -2835,8 +2896,15 @@ function App() { // for the session only: it is a cache of observations, not a source of truth, // and a reload should go back to asking. const [presence, setPresence] = useState({}); - const notePresence = useCallback((gid, state) => { + // Percentage alongside 'indexing' presence — kept separate from `presence` + // itself so a changing % does not require treating every tick as a new + // presence state (see the Sidebar dot's title/aria-label). + const [indexProgressPct, setIndexProgressPct] = useState({}); + const notePresence = useCallback((gid, state, pct) => { setPresence(prev => (prev[gid] === state ? prev : { ...prev, [gid]: state })); + if (pct !== undefined) { + setIndexProgressPct(prev => (prev[gid] === pct ? prev : { ...prev, [gid]: pct })); + } }, []); const handleLeftGroup = useCallback((gid) => { @@ -3084,6 +3152,7 @@ function App() { ${user && html`<${Sidebar} groups=${groups} presence=${presence} + indexProgressPct=${indexProgressPct} route=${route} menuOpen=${menuOpen} role=${user.role} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/group-page.js b/packages/meshbay-hub/src/meshbay_hub/static/group-page.js index 1528213..29f3602 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/group-page.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/group-page.js @@ -76,6 +76,10 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, // every registered app when a node predates the setting (or hasn't answered // yet), so nothing disappears for an existing group. const [enabledApps, setEnabledApps] = useState(null); + // Reconcile interval / debounce currently in effect on the node — shown + // to the operator in Settings, not enforced from here (indexer.py owns + // that). Null until the handshake ack arrives. + const [scanSettings, setScanSettings] = useState(null); // Paired ≠ operator account. `is_node_admin` says the hub account owning this // node is the one connecting; this says the node pinned *this browser's* key // as an operator key. Only the second one lets you sign an invite, and only @@ -123,6 +127,25 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, cacheGroupIndex(groupId, group ? group.name : groupId, fresh); }, [groupId, group]); + // additions/deletions only (daemon.py _broadcast_index_change, once there + // is a previous snapshot to diff against) — applied on top of whatever + // applyIndex last put in `entries`, instead of replacing the whole table + // for one changed file. + const applyIndexDelta = useCallback((deltaMsg) => { + setEntries((prev) => { + const deletions = new Set(deltaMsg.deletions || []); + const kept = prev.filter((e) => !deletions.has(e.id)); + // The index is keyed by content hash: an addition whose id is already + // present is the same duplicate-content case indexer.py's own + // reconcile sweep leaves alone, not a second row for one file. + const keptIds = new Set(kept.map((e) => e.id)); + const additions = (deltaMsg.additions || []).filter((e) => !keptIds.has(e.id)); + const fresh = kept.concat(additions); + cacheGroupIndex(groupId, group ? group.name : groupId, fresh); + return fresh; + }); + }, [groupId, group]); + useEffect(() => { let cancelled = false; @@ -172,11 +195,24 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, setIsNodeAdmin(!!ack.is_node_admin); setMemberUpload(ack.member_upload !== false); setEnabledApps(ack.enabled_apps || null); + setScanSettings(ack.scan_settings || null); // Changed while we are connected, by an operator who may be someone // else entirely. Without this the button stays until a reconnection, // and a button that is still there is a button people press. transport.onUploadPolicy = (allowed) => setMemberUpload(allowed); transport.onAppsEnabled = (apps) => setEnabledApps(apps); + // The node's own scan (a root added while we were already connected, + // or reconcile catching one back up) — never the entries, just + // enough to animate the sidebar dot. Guaranteed a final push at the + // False transition (daemon.py _progress_pusher), so this always + // settles back to 'online' rather than getting stuck. + transport.onIndexProgress = (status) => { + if (cancelled || !onPresence) return; + const pct = status.total_bytes + ? Math.min(100, Math.round(100 * status.scanned_bytes / status.total_bytes)) + : 0; + onPresence(groupId, status.scanning ? 'indexing' : 'online', pct); + }; setOperatorPaired(transport.memberRole === 'operator'); // A first join to this node generated an identity for it; leave it with @@ -203,6 +239,10 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, if (cancelled) return; applyIndex(msg); }; + transport.onIndexDelta = (msg) => { + if (cancelled) return; + applyIndexDelta(msg); + }; // We are in: an invitation to this group has served its purpose. if (onJoined) onJoined(groupId); @@ -214,7 +254,20 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, touchActivity(); // First-hand evidence, and the strongest available: this browser spoke // to the node. It outranks whatever the hub said in the group list. - if (onPresence) onPresence(groupId, 'online'); + // A scan already under way at the moment of connecting (ack.indexing, + // webrtc_server.py _complete_handshake) shows as indexing right away + // rather than waiting for the next periodic push. + if (onPresence) { + const idx = ack.indexing; + if (idx && idx.scanning) { + const pct = idx.total_bytes + ? Math.min(100, Math.round(100 * idx.scanned_bytes / idx.total_bytes)) + : 0; + onPresence(groupId, 'indexing', pct); + } else { + onPresence(groupId, 'online'); + } + } } catch (err) { if (cancelled) return; @@ -448,6 +501,8 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, onMemberUpload=${(allowed) => setMemberUpload(allowed)} enabledApps=${enabledApps} onEnabledApps=${(keys) => setEnabledApps(keys)} + scanSettings=${scanSettings} + onScanSettings=${(s) => setScanSettings(s)} onLeft=${onLeft} onPaired=${() => setOperatorPaired(true)} /> `} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js index a14fd21..0b64a03 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js @@ -22,6 +22,7 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, isNodeAdmin, userId, operatorPaired, connected, memberUpload, onMemberUpload, enabledApps, onEnabledApps, + scanSettings, onScanSettings, onPaired, onLeft }) { const [members, setMembers] = useState([]); const [adminId, setAdminId] = useState(''); @@ -36,6 +37,10 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, const [nodeGroupName, setNodeGroupName] = useState(''); const [nodeBusy, setNodeBusy] = useState(false); const [nodeMsg, setNodeMsg] = useState(''); + // Bytes-based indexing progress while a newly added directory is being + // scanned — same source as the Create Group wizard's step, see + // platform.watchIndexProgress. + const [nodeIndexProgress, setNodeIndexProgress] = useState(null); const loadNodeInfo = useCallback(async () => { if (!platform.node.available) return; @@ -194,6 +199,51 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, } }, [transportRef, onEnabledApps, activeApps]); + const [scanBusy, setScanBusy] = useState(false); + const [scanMsg, setScanMsg] = useState(''); + const [reconcileMinutes, setReconcileMinutes] = useState( + scanSettings ? Math.round(scanSettings.reconcile_interval_secs / 60) : 10); + const [debounceSeconds, setDebounceSeconds] = useState( + scanSettings ? Math.round(scanSettings.debounce_secs) : 2); + // The node is the source of truth; once it has answered, the fields track + // it rather than whatever this browser guessed before connecting. + useEffect(() => { + if (!scanSettings) return; + setReconcileMinutes(Math.round(scanSettings.reconcile_interval_secs / 60)); + setDebounceSeconds(Math.round(scanSettings.debounce_secs)); + }, [scanSettings]); + + /** + * How often the reconciliation backstop runs, and how long a changed file + * is left alone before being hashed. Same shape as toggleApp: signed, and + * the fields do not claim success until the node has confirmed it. + */ + const saveScanSettings = useCallback(async () => { + const transport = transportRef && transportRef.current; + setScanMsg(''); + setScanBusy(true); + try { + if (!transport || !transport.connected) { + throw new Error('Not connected to the node'); + } + const sk = transport.sessionKeys && transport.sessionKeys.skEdB64; + const signFn = (sk && window.MeshBayKeys) + ? (transcript) => window.MeshBayKeys.signBytes(sk, transcript) + : null; + await transport.setScanSettings(reconcileMinutes * 60, debounceSeconds, signFn); + const applied = { + reconcile_interval_secs: reconcileMinutes * 60, + debounce_secs: debounceSeconds, + }; + if (onScanSettings) onScanSettings(applied); + setScanMsg(t('settings_node.scan_saved')); + } catch (err) { + setScanMsg(err.message); + } finally { + setScanBusy(false); + } + }, [transportRef, onScanSettings, reconcileMinutes, debounceSeconds]); + const [removing, setRemoving] = useState(''); /** @@ -382,6 +432,37 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, </div> `} + ${/* How hard the node works watching its own disk — indexer.py + DirectoryIndexer. A performance knob, not a permission: it + changes nothing about who can see or do what. */ + isNodeAdmin && connected && html` + <div class="settings-section"> + <h3 class="settings-heading">${t('settings_node.scan_title')}</h3> + <p class="settings-hint">${t('settings_node.scan_hint')}</p> + <div class="settings-row"> + <label class="settings-label"> + ${t('settings_node.scan_reconcile_label')} + <input type="number" min="1" max="1440" step="1" + value=${reconcileMinutes} disabled=${scanBusy} + onInput=${e => setReconcileMinutes(Number(e.target.value))} /> + </label> + </div> + <div class="settings-row"> + <label class="settings-label"> + ${t('settings_node.scan_debounce_label')} + <input type="number" min="0" max="300" step="1" + value=${debounceSeconds} disabled=${scanBusy} + onInput=${e => setDebounceSeconds(Number(e.target.value))} /> + </label> + </div> + <button class="btn btn-small btn-secondary" style="margin-top:8px" + disabled=${scanBusy} onClick=${saveScanSettings}> + ${scanBusy ? t('settings_node.scan_saving') : t('settings_node.scan_save')} + </button> + ${scanMsg && html`<p class="settings-hint">${scanMsg}</p>`} + </div> + `} + ${/* Roots management (Electron-only, when node is local) */ nodeDetected && nodeRoots.length > 0 && html` <div class="settings-section"> @@ -425,12 +506,18 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, onClick=${async () => { const chosen = await platform.rootPicker.choose(); if (!chosen) return; - setNodeBusy(true); setNodeMsg(''); + setNodeBusy(true); setNodeMsg(''); setNodeIndexProgress(null); try { await platform.node.call('POST', '/api/groups/' + groupId + '/roots', { path: chosen.path, name: chosen.name }); await platform.node.call('POST', '/api/reload'); + // The root is already scanning in the background on the + // node regardless of whether anyone watches this — see + // the "closing the client" test in test_hot_reload_*.py. + // This is only about not leaving the operator staring at + // an unchanged screen while it happens. + await platform.watchIndexProgress(groupId, setNodeIndexProgress); setNodeMsg(t('node.root_added')); await loadNodeInfo(); } catch (err) { setNodeMsg(platform.bridgeMessage(err)); } @@ -438,6 +525,28 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, }}> <${Icon} name="folder-plus" /> ${t('node.add_root')} </button> + ${nodeIndexProgress && nodeIndexProgress.scanning && html` + <div class="index-progress" style="margin-top:8px"> + <div class="index-progress-bar"> + <div class="index-progress-fill" style="width:${ + nodeIndexProgress.total_bytes + ? Math.min(100, Math.round( + 100 * nodeIndexProgress.scanned_bytes / nodeIndexProgress.total_bytes)) + : 0}%"></div> + </div> + <div class="index-progress-label">${t('wizard.indexing_progress', { + pct: nodeIndexProgress.total_bytes + ? Math.min(100, Math.round( + 100 * nodeIndexProgress.scanned_bytes / nodeIndexProgress.total_bytes)) + : 0, + })}</div> + ${nodeIndexProgress.current_dir && html` + <div class="index-progress-dir"> + ${t('wizard.indexing_current_dir', { dir: nodeIndexProgress.current_dir })} + </div> + `} + </div> + `} </div> </div> `} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js index 59127af..866212f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -457,6 +457,7 @@ export default { 'presence.online': 'Ein Node, der diese Gruppe hostet, ist online', 'presence.offline': 'Kein Node für diese Gruppe erreichbar', 'presence.unknown': 'Noch nicht geprüft', + 'presence.indexing': 'Indizierung läuft — {pct}%', 'chat.load_older': '{n} ältere Nachrichten laden', 'chat.start_of_history': 'Anfang des Gesprächs', 'chat.today': 'Heute', @@ -564,6 +565,13 @@ export default { // Settings (node) 'settings_node.detach_failed_continue': 'Die Gruppe konnte nicht vom Node getrennt werden. Trotzdem auf dem Hub löschen?', 'settings_node.roots': 'Freigegebene Verzeichnisse', + 'settings_node.scan_title': 'Scan', + 'settings_node.scan_hint': 'Wie oft der Node seine Verzeichnisse erneut auf verpasste Änderungen prüft und wie lange er nach einer Änderung wartet, bevor eine Datei indiziert wird.', + 'settings_node.scan_reconcile_label': 'Prüfintervall (Minuten)', + 'settings_node.scan_debounce_label': 'Wartezeit nach einer Änderung (Sekunden)', + 'settings_node.scan_save': 'Speichern', + 'settings_node.scan_saving': 'Wird gespeichert…', + 'settings_node.scan_saved': 'Gespeichert.', // Create-group wizard 'wizard.title': 'Gruppe erstellen', @@ -586,6 +594,9 @@ export default { 'wizard.step_add_roots': 'Verzeichnisse hinzufügen', 'wizard.step_gek': 'Verschlüsselungsschlüssel initialisieren', 'wizard.step_pair': 'Kopplung einrichten', + 'wizard.step_index': 'Dateien werden indiziert', + 'wizard.indexing_progress': 'Indizierung… {pct}%', + 'wizard.indexing_current_dir': 'Wird gescannt: {dir}', 'wizard.done_title': 'Gruppe erstellt', 'wizard.done_message': 'Ihre Gruppe ist bereit. Ihr Node hostet sie und die Verschlüsselung ist eingerichtet.', 'wizard.go_to_group': 'Zur Gruppe', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index 58590b1..a83b067 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -364,6 +364,9 @@ export default { 'wizard.step_add_roots': 'Adding directories', 'wizard.step_gek': 'Initializing encryption key', 'wizard.step_pair': 'Setting up pairing', + 'wizard.step_index': 'Indexing files', + 'wizard.indexing_progress': 'Indexing… {pct}%', + 'wizard.indexing_current_dir': 'Scanning: {dir}', 'wizard.finish_later': 'Finish setup later', 'wizard.done_title': 'Group created', 'wizard.done_message': 'Your group is ready. Your node is hosting it and encryption is set up.', @@ -386,6 +389,13 @@ export default { // Unified Group Settings (node sections) 'settings_node.roots': 'Shared directories', 'settings_node.detach_failed_continue': 'Could not detach the group from the node. Delete on hub anyway?', + 'settings_node.scan_title': 'Scanning', + 'settings_node.scan_hint': 'How often the node re-checks its directories for changes it may have missed, and how long it waits after a file changes before indexing it.', + 'settings_node.scan_reconcile_label': 'Re-check interval (minutes)', + 'settings_node.scan_debounce_label': 'Wait after a change (seconds)', + 'settings_node.scan_save': 'Save', + 'settings_node.scan_saving': 'Saving…', + 'settings_node.scan_saved': 'Saved.', // Members 'members.col_role': 'Role', @@ -486,6 +496,7 @@ export default { 'presence.online': 'A node serving this group is online', 'presence.offline': 'No node is reachable for this group', 'presence.unknown': 'Not checked yet', + 'presence.indexing': 'Indexing — {pct}% done', 'chat.load_older': 'Load {n} older messages', 'chat.start_of_history': 'Beginning of the conversation', 'chat.today': 'Today', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js index 75a67b6..1621182 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -452,6 +452,7 @@ export default { 'presence.online': 'Un node que sirve este grupo está en línea', 'presence.offline': 'Ningún node accesible para este grupo', 'presence.unknown': 'Aún sin comprobar', + 'presence.indexing': 'Indexando — {pct}%', 'chat.load_older': 'Cargar {n} mensajes anteriores', 'chat.start_of_history': 'Inicio de la conversación', 'chat.today': 'Hoy', @@ -559,6 +560,13 @@ export default { // Node settings 'settings_node.detach_failed_continue': 'No se pudo desconectar el grupo del node. ¿Eliminar del hub de todos modos?', 'settings_node.roots': 'Directorios compartidos', + 'settings_node.scan_title': 'Escaneo', + 'settings_node.scan_hint': 'Con qué frecuencia el node vuelve a comprobar sus directorios en busca de cambios que pudo haber pasado por alto, y cuánto tiempo espera tras un cambio antes de indexar un archivo.', + 'settings_node.scan_reconcile_label': 'Intervalo de comprobación (minutos)', + 'settings_node.scan_debounce_label': 'Espera tras un cambio (segundos)', + 'settings_node.scan_save': 'Guardar', + 'settings_node.scan_saving': 'Guardando…', + 'settings_node.scan_saved': 'Guardado.', // Create group wizard 'wizard.title': 'Crear grupo', @@ -581,6 +589,9 @@ export default { 'wizard.step_add_roots': 'Añadiendo directorios', 'wizard.step_gek': 'Inicializando clave de cifrado', 'wizard.step_pair': 'Configurando emparejamiento', + 'wizard.step_index': 'Indexando archivos', + 'wizard.indexing_progress': 'Indexando… {pct}%', + 'wizard.indexing_current_dir': 'Analizando: {dir}', 'wizard.done_title': 'Grupo creado', 'wizard.done_message': 'Su grupo está listo. Su node lo aloja y el cifrado está configurado.', 'wizard.go_to_group': 'Ir al grupo', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js index d60c5d0..6e21a06 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -456,6 +456,7 @@ export default { 'presence.online': 'Un node servant ce groupe est en ligne', 'presence.offline': 'Aucun node joignable pour ce groupe', 'presence.unknown': 'Pas encore vérifié', + 'presence.indexing': 'Indexation en cours — {pct} %', 'chat.load_older': 'Charger {n} messages plus anciens', 'chat.start_of_history': 'Début de la conversation', 'chat.today': 'Aujourd’hui', @@ -575,6 +576,13 @@ export default { // Node settings 'settings_node.detach_failed_continue': 'Impossible de détacher le groupe du node. Supprimer quand même sur le hub ?', 'settings_node.roots': 'Répertoires partagés', + 'settings_node.scan_title': 'Analyse', + 'settings_node.scan_hint': 'À quelle fréquence le node revérifie ses répertoires à la recherche de changements manqués, et combien de temps il attend après une modification avant d\'indexer un fichier.', + 'settings_node.scan_reconcile_label': 'Intervalle de revérification (minutes)', + 'settings_node.scan_debounce_label': 'Attente après un changement (secondes)', + 'settings_node.scan_save': 'Enregistrer', + 'settings_node.scan_saving': 'Enregistrement…', + 'settings_node.scan_saved': 'Enregistré.', // Create group wizard 'wizard.title': 'Créer un groupe', @@ -597,6 +605,9 @@ export default { 'wizard.step_add_roots': 'Ajout des répertoires', 'wizard.step_gek': 'Initialisation de la clé de chiffrement', 'wizard.step_pair': 'Mise en place de l\'appariement', + 'wizard.step_index': 'Indexation des fichiers', + 'wizard.indexing_progress': 'Indexation… {pct} %', + 'wizard.indexing_current_dir': 'Analyse : {dir}', 'wizard.done_title': 'Groupe créé', 'wizard.done_message': 'Votre groupe est prêt. Votre node l\'héberge et le chiffrement est configuré.', 'wizard.go_to_group': 'Aller au groupe', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js index b65e2ce..edd15ab 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -454,6 +454,7 @@ export default { 'presence.online': 'Un node che ospita questo gruppo è online', 'presence.offline': 'Nessun node raggiungibile per questo gruppo', 'presence.unknown': 'Non ancora verificato', + 'presence.indexing': 'Indicizzazione in corso — {pct}%', 'chat.load_older': 'Carica {n} messaggi precedenti', 'chat.start_of_history': 'Inizio della conversazione', 'chat.today': 'Oggi', @@ -573,6 +574,13 @@ export default { // Settings — node 'settings_node.detach_failed_continue': 'Impossibile scollegare il gruppo dal node. Eliminare comunque dal hub?', 'settings_node.roots': 'Directory condivise', + 'settings_node.scan_title': 'Scansione', + 'settings_node.scan_hint': 'Con quale frequenza il node ricontrolla le sue directory per cambiamenti che potrebbe aver perso, e quanto tempo attende dopo una modifica prima di indicizzare un file.', + 'settings_node.scan_reconcile_label': 'Intervallo di ricontrollo (minuti)', + 'settings_node.scan_debounce_label': 'Attesa dopo una modifica (secondi)', + 'settings_node.scan_save': 'Salva', + 'settings_node.scan_saving': 'Salvataggio…', + 'settings_node.scan_saved': 'Salvato.', // Create-group wizard 'wizard.title': 'Crea gruppo', @@ -595,6 +603,9 @@ export default { 'wizard.step_add_roots': 'Aggiunta delle directory', 'wizard.step_gek': 'Inizializzazione della chiave di cifratura', 'wizard.step_pair': 'Configurazione del pairing', + 'wizard.step_index': 'Indicizzazione dei file', + 'wizard.indexing_progress': 'Indicizzazione… {pct}%', + 'wizard.indexing_current_dir': 'Scansione: {dir}', 'wizard.done_title': 'Gruppo creato', 'wizard.done_message': 'Il suo gruppo è pronto. Il suo node lo ospita e la cifratura è configurata.', 'wizard.go_to_group': 'Vai al gruppo', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js index e965590..00fffed 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -442,6 +442,7 @@ export default { 'presence.online': 'このグループをホストする node が稼働中です', 'presence.offline': 'このグループに到達できる node がありません', 'presence.unknown': '未確認', + 'presence.indexing': 'インデックス中 — {pct}%', 'chat.load_older': '以前のメッセージを {n} 件読み込む', 'chat.start_of_history': '会話のはじまり', 'chat.today': '今日', @@ -557,6 +558,13 @@ export default { // Settings — node 'settings_node.detach_failed_continue': 'node からグループを切り離せませんでした。それでも hub 上で削除しますか？', 'settings_node.roots': '共有ディレクトリ', + 'settings_node.scan_title': 'スキャン', + 'settings_node.scan_hint': 'node が見逃した変更がないかディレクトリを再確認する頻度と、ファイルの変更後にインデックスするまで待つ時間。', + 'settings_node.scan_reconcile_label': '再確認の間隔（分）', + 'settings_node.scan_debounce_label': '変更後の待機時間（秒）', + 'settings_node.scan_save': '保存', + 'settings_node.scan_saving': '保存中…', + 'settings_node.scan_saved': '保存しました。', // Wizard 'wizard.title': 'グループを作成', @@ -579,6 +587,9 @@ export default { 'wizard.step_add_roots': 'ディレクトリを追加中', 'wizard.step_gek': '暗号化鍵を初期化中', 'wizard.step_pair': 'ペアリングをセットアップ中', + 'wizard.step_index': 'ファイルをインデックス中', + 'wizard.indexing_progress': 'インデックス中… {pct}%', + 'wizard.indexing_current_dir': 'スキャン中: {dir}', 'wizard.done_title': 'グループを作成しました', 'wizard.done_message': 'グループの準備ができました。node がホストし、暗号化が設定されています。', 'wizard.go_to_group': 'グループを開く', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js index b37be08..2837ebb 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -456,6 +456,7 @@ export default { 'presence.online': 'Een node die deze groep host is online', 'presence.offline': 'Geen node bereikbaar voor deze groep', 'presence.unknown': 'Nog niet gecontroleerd', + 'presence.indexing': 'Bezig met indexeren — {pct}%', 'chat.load_older': '{n} oudere berichten laden', 'chat.start_of_history': 'Begin van het gesprek', 'chat.today': 'Vandaag', @@ -575,6 +576,13 @@ export default { // Node settings 'settings_node.detach_failed_continue': 'Kon de groep niet van de node loskoppelen. Toch op de hub verwijderen?', 'settings_node.roots': 'Gedeelde mappen', + 'settings_node.scan_title': 'Scannen', + 'settings_node.scan_hint': 'Hoe vaak de node zijn mappen opnieuw controleert op wijzigingen die zijn gemist, en hoe lang hij na een wijziging wacht voordat een bestand wordt geïndexeerd.', + 'settings_node.scan_reconcile_label': 'Controle-interval (minuten)', + 'settings_node.scan_debounce_label': 'Wachttijd na een wijziging (seconden)', + 'settings_node.scan_save': 'Opslaan', + 'settings_node.scan_saving': 'Bezig met opslaan…', + 'settings_node.scan_saved': 'Opgeslagen.', // Create group wizard 'wizard.title': 'Groep aanmaken', @@ -597,6 +605,9 @@ export default { 'wizard.step_add_roots': 'Mappen toevoegen', 'wizard.step_gek': 'Versleutelingssleutel initialiseren', 'wizard.step_pair': 'Koppeling instellen', + 'wizard.step_index': 'Bestanden indexeren', + 'wizard.indexing_progress': 'Bezig met indexeren… {pct}%', + 'wizard.indexing_current_dir': 'Scannen: {dir}', 'wizard.done_title': 'Groep aangemaakt', 'wizard.done_message': 'Uw groep is klaar. Uw node host de groep en de versleuteling is ingesteld.', 'wizard.go_to_group': 'Naar groep', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js index b445f79..760f29c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -469,6 +469,7 @@ export default { 'presence.online': 'Node hostujący tę grupę jest dostępny', 'presence.offline': 'Brak osiągalnego node dla tej grupy', 'presence.unknown': 'Jeszcze nie sprawdzono', + 'presence.indexing': 'Indeksowanie — {pct}%', 'chat.load_older': 'Wczytaj {n} starszych wiadomości', 'chat.start_of_history': 'Początek rozmowy', 'chat.today': 'Dziś', @@ -596,6 +597,13 @@ export default { // Settings — node 'settings_node.detach_failed_continue': 'Nie udało się odłączyć grupy od node. Usunąć mimo to na hub?', 'settings_node.roots': 'Katalogi współdzielone', + 'settings_node.scan_title': 'Skanowanie', + 'settings_node.scan_hint': 'Jak często node ponownie sprawdza swoje katalogi w poszukiwaniu pominiętych zmian oraz jak długo czeka po zmianie pliku przed jego zindeksowaniem.', + 'settings_node.scan_reconcile_label': 'Interwał sprawdzania (minuty)', + 'settings_node.scan_debounce_label': 'Oczekiwanie po zmianie (sekundy)', + 'settings_node.scan_save': 'Zapisz', + 'settings_node.scan_saving': 'Zapisywanie…', + 'settings_node.scan_saved': 'Zapisano.', // Create-group wizard 'wizard.title': 'Utwórz grupę', @@ -618,6 +626,9 @@ export default { 'wizard.step_add_roots': 'Dodawanie katalogów', 'wizard.step_gek': 'Inicjalizacja klucza szyfrowania', 'wizard.step_pair': 'Konfigurowanie parowania', + 'wizard.step_index': 'Indeksowanie plików', + 'wizard.indexing_progress': 'Indeksowanie… {pct}%', + 'wizard.indexing_current_dir': 'Skanowanie: {dir}', 'wizard.done_title': 'Grupa utworzona', 'wizard.done_message': 'Grupa jest gotowa. Node ją hostuje, a szyfrowanie jest skonfigurowane.', 'wizard.go_to_group': 'Przejdź do grupy', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js index c3220ae..0001051 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js @@ -453,6 +453,7 @@ export default { 'presence.online': 'Um node que hospeda este grupo está on-line', 'presence.offline': 'Nenhum node acessível para este grupo', 'presence.unknown': 'Ainda não verificado', + 'presence.indexing': 'Indexando — {pct}%', 'chat.load_older': 'Carregar {n} mensagens anteriores', 'chat.start_of_history': 'Início da conversa', 'chat.today': 'Hoje', @@ -560,6 +561,13 @@ export default { // Settings — node 'settings_node.detach_failed_continue': 'Não foi possível desanexar o grupo do node. Excluir do hub mesmo assim?', 'settings_node.roots': 'Diretórios compartilhados', + 'settings_node.scan_title': 'Varredura', + 'settings_node.scan_hint': 'Com que frequência o node reverifica seus diretórios em busca de mudanças que possa ter perdido, e quanto tempo espera após uma mudança antes de indexar um arquivo.', + 'settings_node.scan_reconcile_label': 'Intervalo de reverificação (minutos)', + 'settings_node.scan_debounce_label': 'Espera após uma mudança (segundos)', + 'settings_node.scan_save': 'Salvar', + 'settings_node.scan_saving': 'Salvando…', + 'settings_node.scan_saved': 'Salvo.', // Create group wizard 'wizard.title': 'Criar grupo', @@ -582,6 +590,9 @@ export default { 'wizard.step_add_roots': 'Adicionando diretórios', 'wizard.step_gek': 'Inicializando chave de criptografia', 'wizard.step_pair': 'Configurando pareamento', + 'wizard.step_index': 'Indexando arquivos', + 'wizard.indexing_progress': 'Indexando… {pct}%', + 'wizard.indexing_current_dir': 'Analisando: {dir}', 'wizard.done_title': 'Grupo criado', 'wizard.done_message': 'Seu grupo está pronto. Seu node o está hospedando e a criptografia está configurada.', 'wizard.go_to_group': 'Ir para o grupo', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js index 0151c4b..ad8bf53 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js @@ -428,6 +428,7 @@ export default { 'presence.online': '有一个服务本群组的 node 在线', 'presence.offline': '本群组没有可达的 node', 'presence.unknown': '尚未检测', + 'presence.indexing': '正在索引 — {pct}%', 'chat.load_older': '加载更早的 {n} 条消息', 'chat.start_of_history': '对话开头', 'chat.today': '今天', @@ -543,6 +544,13 @@ export default { // Settings — node 'settings_node.detach_failed_continue': '无法从 node 上分离此群组。是否仍然在 hub 上删除？', 'settings_node.roots': '共享目录', + 'settings_node.scan_title': '扫描', + 'settings_node.scan_hint': 'node 多久重新检查一次目录以发现可能错过的变化，以及文件变化后等待多久才建立索引。', + 'settings_node.scan_reconcile_label': '重新检查间隔（分钟）', + 'settings_node.scan_debounce_label': '变化后的等待时间（秒）', + 'settings_node.scan_save': '保存', + 'settings_node.scan_saving': '保存中…', + 'settings_node.scan_saved': '已保存。', // Create group wizard 'wizard.title': '创建群组', @@ -566,6 +574,9 @@ export default { 'wizard.step_add_roots': '添加目录', 'wizard.step_gek': '初始化加密密钥', 'wizard.step_pair': '设置配对', + 'wizard.step_index': '正在索引文件', + 'wizard.indexing_progress': '正在索引… {pct}%', + 'wizard.indexing_current_dir': '正在扫描：{dir}', 'wizard.done_title': '群组已创建', 'wizard.done_message': '您的群组已就绪。您的 node 正在托管它，加密已设置完成。', 'wizard.go_to_group': '前往群组', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/platform.js b/packages/meshbay-hub/src/meshbay_hub/static/platform.js index 30df80e..a5312e1 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/platform.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/platform.js @@ -266,6 +266,68 @@ export const node = { }; /** + * Poll a group's initial-scan progress on the local node (loopback), until + * it reports it is no longer scanning. For "add a directory" in Settings, + * where the group is already hosted — index-status is meaningful the whole + * time. NOT for a brand-new group the wizard just attached: see + * waitForGroupHosted below for why that needs a different exit condition. + * + * `onUpdate` is called with each {scanning, scanned_bytes, total_bytes, + * current_dir} snapshot, including the final one where scanning is false. + */ +export async function watchIndexProgress(groupId, onUpdate, { intervalMs = 500 } = {}) { + for (;;) { + let status; + try { + status = await node.call('GET', `/api/groups/${groupId}/index-status`); + } catch { + // The node went away mid-poll — stop rather than spin forever; the + // caller's own connection-status handling already covers that case. + return; + } + onUpdate(status); + if (!status.scanning) return; + await new Promise(resolve => setTimeout(resolve, intervalMs)); + } +} + +/** + * Create Group wizard only: wait for a brand-new group to actually become + * usable on the node, showing index-status along the way. + * + * Not the same wait as watchIndexProgress above. `/api/reload` returns as + * soon as the reload is scheduled (ops.start_reload) — before the node has + * even created an indexer for the group, let alone started scanning. A + * naive "poll index-status until scanning is false" would see the default + * idle answer on that very first poll and return immediately, and every + * group-scoped call after it (add a root, init the GEK) would still 404 + * with "not configured"/"not hosted" for as long as the real scan actually + * takes. The only answer that means "safe to proceed" is the group + * genuinely appearing in /api/groups (groups_ctx, daemon.py) — index-status + * is read purely for the progress bar. + */ +export async function waitForGroupHosted(groupId, onProgress, + { intervalMs = 500, timeoutMs = 30 * 60 * 1000 } = {}) { + const deadline = Date.now() + timeoutMs; + for (;;) { + try { + const status = await node.call('GET', `/api/groups/${groupId}/index-status`); + if (onProgress) onProgress(status); + } catch { /* keep waiting — the loopback API can be momentarily busy */ } + + try { + const list = await node.call('GET', '/api/groups'); + if (Array.isArray(list.groups) && list.groups.some((g) => g.id === groupId)) return; + } catch { /* keep waiting */ } + + if (Date.now() > deadline) { + throw new Error('The node did not finish attaching this group in time'); + } + await new Promise((r) => setTimeout(r, intervalMs)); + } +} + +/** * LAN cast relay — re-serve decrypted video segments over HTTP so a * Chromecast or Smart TV on the same Wi-Fi can play the stream. * diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css index 20b9f8c..2cb27d0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -989,6 +989,17 @@ button:disabled { opacity: 0.5; cursor: not-allowed; } font-size: 0.9em; color: var(--text); } +.settings-label input[type="number"] { + display: block; + width: 90px; + margin-top: 4px; + padding: 6px 8px; + border-radius: 6px; + border: 1px solid var(--border); + background: var(--bg-base); + color: var(--text); + font-size: 0.95em; +} .settings-value { font-size: 0.9em; @@ -1887,6 +1898,19 @@ a.transfer-name { .presence-online { background: var(--success); } .presence-offline { background: var(--error); } .presence-unknown { background: transparent; border-color: var(--text-dim); } +/* The node is scanning this group's files right now (initial import, or a + directory just added) — a state, not a verdict, so it pulses rather than + sitting on a fixed color; green at rest, green again once scanning stops + (driven by IndexProgress.scanning, guaranteed to turn back off — see + daemon.py _progress_pusher and webrtc_server.py's handshake ack). */ +.presence-indexing { + background: var(--success); + animation: presence-indexing-pulse 1.6s ease-in-out infinite; +} +@keyframes presence-indexing-pulse { + 0%, 100% { background: var(--success); } + 50% { background: #f59e0b; } +} .sidebar-item-name { overflow: hidden; @@ -2284,7 +2308,7 @@ a.transfer-name { align-items: center; justify-content: space-between; padding: 8px 12px; - background: var(--surface); + background: var(--bg-surface); border: 1px solid var(--border); border-radius: 8px; gap: 8px; @@ -2322,7 +2346,7 @@ a.transfer-name { align-items: center; gap: 10px; padding: 10px 14px; - background: var(--surface); + background: var(--bg-surface); border: 1px solid var(--border); border-radius: 8px; font-size: 0.9em; @@ -2337,6 +2361,37 @@ a.transfer-name { .wizard-step-error .wizard-step-icon { color: var(--error); } .wizard-step-pending { opacity: 0.5; } +/* ── Indexing progress (Create Group wizard, and "add a directory" in + Settings) — bytes-based, not file-count-based; see the two call sites of + platform.watchIndexProgress. ── */ +.index-progress { + padding: 10px 14px; + background: var(--bg-surface); + border: 1px solid var(--border); + border-radius: 8px; + font-size: 0.9em; +} +.index-progress-bar { + height: 6px; + border-radius: 3px; + background: var(--border); + overflow: hidden; + margin-bottom: 8px; +} +.index-progress-fill { + height: 100%; + background: var(--accent); + transition: width 0.3s ease; +} +.index-progress-label { font-weight: 600; } +.index-progress-dir { + color: var(--text-dim); + margin-top: 2px; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + .warning-msg { padding: 10px 14px; background: color-mix(in srgb, #f59e0b 15%, transparent); diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index da98253..4f6b656 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -103,8 +103,10 @@ class MeshBayTransport { set onStreamEnd(fn) { this._onStreamEnd = fn; } set onStreamError(fn) { this._onStreamError = fn; } set onIndexSync(fn) { this._onIndexSync = fn; } + set onIndexDelta(fn) { this._onIndexDelta = fn; } set onUploadPolicy(fn) { this._onUploadPolicy = fn; } set onAppsEnabled(fn) { this._onAppsEnabled = fn; } + set onIndexProgress(fn) { this._onIndexProgress = fn; } get sessionKeys() { return this._sessionKeys; } @@ -634,6 +636,29 @@ class MeshBayTransport { return msg; } + /** + * How often the node's reconciliation backstop runs, and how long it + * waits after a file's last write before hashing it (indexer.py + * DirectoryIndexer). Whole seconds only: the node builds the signing + * subject with Python's `%g` (drops a trailing ".0"), and the simplest + * way to always match it byte-for-byte from JS is to never send a + * fractional value in the first place. + */ + async setScanSettings(reconcileIntervalSecs, debounceSecs, signFn) { + const reconcile = Math.round(reconcileIntervalSecs); + const debounce = Math.round(debounceSecs); + const msg = await this._sendAndWait({ + type: 'set_scan_settings', v: '0.1', + reconcile_interval_secs: reconcile, debounce_secs: debounce, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp( + msg, 'set_scan_settings', `${reconcile},${debounce}`, signFn); + } + return msg; + } + async revokeMember(userId, signFn) { const msg = await this._sendAndWait({ type: 'member_revoke', v: '0.1', user_id: userId, @@ -1251,6 +1276,37 @@ class MeshBayTransport { this._onAppsEnabled(msg.apps || []); } + // The operator's node is scanning — never the entries themselves, just + // enough to animate a presence dot. Pushed periodically while it runs, + // plus once more on the transition back to idle (daemon.py + // _progress_pusher). UNLIKE member_upload_ack/apps_enabled_ack above, + // this is never a reply to anything this browser asked for — nobody + // calls _sendAndWait for it — so it MUST return here. Falling through + // to the "oldest pending" guess below hands it to whatever unrelated + // request happens to be waiting (a handshake, a chat history fetch), + // which then waits forever for its real answer while this one already + // "arrived" — and every message after that is one slot off too. Found + // live: a group mid-scan corrupted its own handshake and chat history + // this way, arriving roughly every 2s for as long as scanning ran. + if (msg.type === 'index_progress') { + if (this._onIndexProgress) { + this._onIndexProgress({ + scanning: Boolean(msg.scanning), + scanned_bytes: msg.scanned_bytes || 0, + total_bytes: msg.total_bytes || 0, + }); + } + return; + } + + // Same reasoning as index_progress: nobody awaits this one either, it + // is purely informational (group-settings.js does not currently act on + // it), so it must not be left to fall through to the oldest pending + // request. + if (msg.type === 'set_scan_settings_ack') { + return; + } + if (msg.type === 'index_sync' && msg.entries) { if (this._onIndexSync) this._onIndexSync(msg); const oldest = this._pending.entries().next(); @@ -1260,6 +1316,16 @@ class MeshBayTransport { return; } + // Incremental update — additions/deletions only, never the whole index. + // Only ever arrives after the full index this browser already has (the + // node's first push to a newly connected peer is always index_sync, see + // daemon.py _broadcast_index_change), so there is always a base to + // apply it to. + if (msg.type === 'index_delta') { + if (this._onIndexDelta) this._onIndexDelta(msg); + return; + } + if (msg.type === 'file_chunk') { const key = `chunk:${msg.file_id}:${msg.chunk_index}`; for (const [, handler] of this._pending) { @@ -1298,6 +1364,29 @@ class MeshBayTransport { return; } + // chat_hist_resp answers a `chat_hist` request, but under a different + // type string — unlike index_sync, which is asked for and answered under + // the same name, so the generic fallback below happens to work for it by + // accident. Without this check, whenever a chat_hist_resp arrives while + // something else this browser asked for (fetchIndex, even the handshake + // itself) is still the oldest pending entry, it gets handed to that + // instead: the request chat_hist_resp actually belongs to then hangs + // until _sendAndWait's own 30s timeout, and whatever it stole from + // resolves with the wrong shape entirely — reproduced live as a + // consistent ~30s hang immediately after a successful handshake, for one + // specific group and not others connected the same way, which is exactly + // what depending on response arrival order rather than on request type + // predicts: it fires only when the two responses happen to reorder. + if (msg.type === 'chat_hist_resp') { + const oldest = this._pending.entries().next(); + if (!oldest.done && oldest.value[1]._reqType === 'chat_hist') { + oldest.value[1].resolve(msg); + } else { + console.warn('[MeshBay] chat_hist_resp with no matching chat_hist pending'); + } + return; + } + // Everything above is routed by something in the message. What is left is // matched by arrival order, which is only ever a guess — and a wrong guess // here hands one request's answer to another, which then waits for a reply @@ -1349,6 +1438,16 @@ function _encodeValue(val, parts) { const b = new Uint8Array(5); b[0] = 0xce; new DataView(b.buffer).setUint32(1, val, false); parts.push(b); + } else if (val >= 0 && val <= Number.MAX_SAFE_INTEGER) { + // Same split as the 0xcf decoder case above, in reverse — without + // this, a value over 0xffffffff fell to the plain int32 branch + // below and silently wrapped to a wrong, unrelated number instead + // of failing loudly. + const b = new Uint8Array(9); b[0] = 0xcf; + const dv = new DataView(b.buffer); + dv.setUint32(1, Math.floor(val / 4294967296), false); + dv.setUint32(5, val % 4294967296, false); + parts.push(b); } else if (val >= -32 && val < 0) { parts.push(new Uint8Array([val & 0xff])); } else if (val >= -128 && val < 0) { @@ -1460,6 +1559,26 @@ function _decodeValue(buf, view, offset) { case 0xcc: return [buf[offset + 1], offset + 2]; case 0xcd: return [view.getUint16(offset + 1, false), offset + 3]; case 0xce: return [view.getUint32(offset + 1, false), offset + 5]; + // uint64/int64 — never emitted by this file's own encoder (a JS number + // above 0xffffffff falls to float64 there), but the node's real msgpack + // library sends a plain uint64 for any Python int over ~4.3 billion, and + // a raw byte count crosses that easily (found live: IndexProgress. + // scanned_bytes/total_bytes in the handshake ack, indexer.py, for a + // group whose total library size exceeds ~4 GB). Split into two 32-bit + // halves rather than DataView's getBigUint64/getBigInt64 — a BigInt + // would silently poison every arithmetic use of these fields elsewhere + // (percentage math, comparisons) — and every real byte count fits in a + // plain JS number well under Number.MAX_SAFE_INTEGER (2^53). + case 0xcf: { + const hi = view.getUint32(offset + 1, false); + const lo = view.getUint32(offset + 5, false); + return [hi * 4294967296 + lo, offset + 9]; + } + case 0xd3: { + const hi = view.getInt32(offset + 1, false); + const lo = view.getUint32(offset + 5, false); + return [hi * 4294967296 + lo, offset + 9]; + } case 0xcb: return [view.getFloat64(offset + 1, false), offset + 9]; case 0xd0: return [view.getInt8(offset + 1), offset + 2]; case 0xd1: return [view.getInt16(offset + 1, false), offset + 3]; diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index c68f999..af85340 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -45,7 +45,7 @@ from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config from meshbay_node.roots import RootSet, RootError from meshbay_node.hub_client import HubClient, HubConfig -from meshbay_node.indexer import DirectoryIndexer +from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.roster import Roster from meshbay_node.transport import ( @@ -119,6 +119,17 @@ class NodeDaemon: self._denylist = ( Denylist(path=config.data_dir / "denylist.json") if Denylist else None) self._chat_stores: dict[str, ChatStore] = {} + self._index_caches: dict[str, IndexCache] = {} + # Coalesces a burst of index changes (one per debounced watchdog + # event) into a single broadcast — see _on_index_change. 0.5s is + # short enough nobody notices the wait, long enough that dropping a + # few hundred files into a watched folder produces one push instead + # of one per file. + self._broadcast_coalesce_secs = 0.5 + self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {} + # group_id -> (version, {id: entry}) as of the last thing actually + # broadcast — the comparison point for the next delta. + self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None self._roster: Roster | None = None @@ -234,12 +245,31 @@ class NodeDaemon: log.info("No GEK yet for group %s — will accept first setup", group_cfg.name) + index_cache = IndexCache( + db_path=data_dir / group_cfg.id[:16] / "index_cache.db") + await index_cache.open() + self._index_caches[group_cfg.id] = index_cache + + # Read once at load, like member_upload/enabled_apps below — + # kept current in place afterwards by set_scan_settings + # (ops.py), which updates both this indexer object directly + # and roster.db, so a restart picks up the same values. + scan_settings = ( + await self._roster.scan_settings(group_cfg.id) + if self._roster else { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }) + indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, on_change=self._on_index_change, + cache=index_cache, + reconcile_secs=scan_settings["reconcile_interval_secs"], + debounce_secs=scan_settings["debounce_secs"], ) await indexer.start(defer_scan=True) self._indexers.append(indexer) @@ -253,6 +283,20 @@ class NodeDaemon: "gek": gek, "roots": roots, "index": indexer.index, + # Live reference, mutated in place by the indexer itself + # (see IndexProgress in indexer.py) — read, never copied, + # by the handshake ack and the periodic progress pusher. + "progress": indexer.progress, + # Bound method, called when a peer completes the + # handshake — resets reconcile's backoff (indexer.py + # _reconcile_loop) so the backstop is prompt again now + # that someone is actually looking. + "note_activity": indexer.note_activity, + # Shown to the operator in Settings, and kept current in + # place by set_scan_settings (ops.py) — same reasoning as + # member_upload below. + "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], + "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, # Admission policy comes from node.toml, never from the hub: # a hub that could declare a group open would be handed its key. @@ -456,6 +500,8 @@ class NodeDaemon: if gctx: self._tasks.append(asyncio.create_task( _bg_scan(idx, group_cfg.name, gctx))) + self._tasks.append(asyncio.create_task( + self._progress_pusher(idx))) # 12. Wait for shutdown stop_event = asyncio.Event() @@ -561,18 +607,40 @@ class NodeDaemon: if gek: log.info("GEK loaded for new group %s", group_cfg.id[:8]) + index_cache = IndexCache( + db_path=data_dir / group_cfg.id[:16] / "index_cache.db") + await index_cache.open() + self._index_caches[group_cfg.id] = index_cache + + scan_settings = ( + await self._roster.scan_settings(group_cfg.id) + if self._roster else { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }) + indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=sk_ed, gek=gek, on_change=self._on_index_change, + cache=index_cache, + reconcile_secs=scan_settings["reconcile_interval_secs"], + debounce_secs=scan_settings["debounce_secs"], ) - await indexer.start() + # Registered *before* start() runs its (blocking, possibly very + # long — see the StarWars benchmark) initial scan, specifically + # so /api/groups/{id}/index-status can see indexer.progress + # while a brand-new group is still scanning — this is the one + # group state that must stay visible during the very window the + # group is not yet authorized for member connections (below). self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer + await indexer.start() + data_dir.mkdir(parents=True, exist_ok=True) chat_db = data_dir / group_cfg.id[:16] / "chat.db" store = ChatStore(db_path=chat_db) @@ -583,6 +651,10 @@ class NodeDaemon: "gek": gek, "roots": roots, "index": indexer.index, + "progress": indexer.progress, + "note_activity": indexer.note_activity, + "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], + "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, "join_policy": group_cfg.join_policy, "member_upload": ( @@ -600,6 +672,11 @@ class NodeDaemon: log.info("Hot-loaded group %s (%s, %d roots)", group_cfg.name, group_cfg.id[:8], len(roots)) added_names.append(group_cfg.name) + # The initial scan above already ran to completion (indexer.start() + # is not deferred here), so this only matters for whatever scans + # this group as time goes on — a root added later, reconcile + # picking one back up. + self._tasks.append(asyncio.create_task(self._progress_pusher(indexer))) # ── Tear down removed groups ───────────────────────────────────── removed_names = [] @@ -618,6 +695,16 @@ class NodeDaemon: await store.close() except Exception: pass + cache = self._index_caches.pop(gid, None) + if cache: + try: + await cache.close() + except Exception: + pass + pending = self._pending_broadcasts.pop(gid, None) + if pending: + pending.cancel() + self._last_broadcast_snapshot.pop(gid, None) self._state["indexes"].pop(gid, None) self._state["indexers"].pop(gid, None) old_name = gid[:8] @@ -708,46 +795,149 @@ class NodeDaemon: log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None + async def _progress_pusher(self, indexer: DirectoryIndexer, + interval: float = 2.0) -> None: + """ + Watches indexer.progress and pushes a light INDEX_PROGRESS message to + this group's connected peers — never the index itself, that stays + _on_index_change's job. Runs for the node's whole lifetime: a scan + can start from several places (initial scan, a root added later, + reconcile picking a root back up), and this only needs to notice the + flag, not why it changed. + + The final push at the False transition is what lets a presence dot + reliably turn back off on an already-connected client — the + handshake ack only covers the moment of connecting. `interval` is a + parameter (not a bare constant) only so a test can drive this loop + without waiting on the real 2s cadence. + """ + was_scanning = False + while True: + await asyncio.sleep(interval) + progress = indexer.progress + now_scanning = progress.scanning + if now_scanning or was_scanning: + self._push_index_progress(indexer.group_id, progress) + was_scanning = now_scanning + + def _push_index_progress(self, group_id: str, progress) -> None: + if not self._webrtc: + return + msg = { + "type": MNP.INDEX_PROGRESS, + "v": MNP_VERSION, + "group_id": group_id, + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + } + pushed = 0 + for session in list(self._webrtc._sessions.values()): + if session._group_id == group_id: + try: + session._send(msg) + pushed += 1 + except Exception: + pass + if pushed: + log.debug("Index progress pushed to %d peer(s) for group %s", + pushed, group_id[:8]) + async def _on_index_change(self, indexer: DirectoryIndexer) -> None: - """Called when a DirectoryIndexer detects file changes.""" + """ + Called when a DirectoryIndexer detects file changes — once per + debounced watchdog event, so dropping N files into a watched folder + calls this N times in quick succession. Coalesces those into one + broadcast (_broadcast_index_change) rather than one push per file: + the timer is reset on every call and only fires once calls stop + arriving for _broadcast_coalesce_secs. + """ + group_id = indexer.group_id + loop = asyncio.get_event_loop() + pending = self._pending_broadcasts.pop(group_id, None) + if pending: + pending.cancel() + + def fire() -> None: + self._pending_broadcasts.pop(group_id, None) + asyncio.ensure_future(self._broadcast_index_change(indexer)) + + self._pending_broadcasts[group_id] = loop.call_later( + self._broadcast_coalesce_secs, fire) + + async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None: + """ + The actual push, run once per coalesced burst. Sends a full + INDEX_SYNC the first time a group is ever broadcast (no previous + snapshot to diff against — the client's own first fetchIndex() call + already covers that case) and an INDEX_DELTA every time after, + computed against the last thing this method actually sent. + """ group_id = indexer.group_id idx = indexer.index log.info("Index changed for group %s: %d files (v%d)", group_id[:8], idx.count, idx.version) - # 11.5 — Push updated index to connected WebRTC peers in this group + prev = self._last_broadcast_snapshot.get(group_id) + delta = None + if prev is not None: + prev_version, prev_entries = prev + previous = GroupIndex._snapshot( + idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries) + delta = idx.diff(previous) + self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) + + # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: - entries = [ - { - "id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at, + if delta is not None: + msg = { + "type": MNP.INDEX_DELTA, + "v": MNP_VERSION, + "group_id": idx.group_id, + "base_version": delta.base_version, + "version": delta.version, + "additions": [ + {"id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at} + for e in delta.additions + ], + "deletions": delta.deletions, + } + else: + msg = { + "type": MNP.INDEX_SYNC, + "v": MNP_VERSION, + "group_id": idx.group_id, + "version": idx.version, + "entries": [ + {"id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at} + for e in idx.entries + ], } - for e in idx.entries - ] - sync_msg = { - "type": MNP.INDEX_SYNC, - "v": MNP_VERSION, - "group_id": idx.group_id, - "version": idx.version, - "entries": entries, - } pushed = 0 for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: try: - session._send(sync_msg) + session._send(msg) pushed += 1 except Exception: pass if pushed: - log.info("Index pushed to %d WebRTC peers", pushed) + log.info("Index %s pushed to %d WebRTC peers", + "delta" if delta is not None else "sync", pushed) # 11.9 — Register file hashes with hub swarm table (public groups only, H7) group_cfg = next( (g for g in self._config.groups if g.id == group_id), None) if (self._hub and self._state.get("endpoint_hint") and group_cfg and group_cfg.visibility == "public"): - hashes = [e.id for e in idx.entries] + # Only the newly added hashes once there is a delta to know them + # from — registering the whole library again on every change is + # the same O(changes x library size) cost the delta above exists + # to avoid. + hashes = ([e.id for e in delta.additions] if delta is not None + else [e.id for e in idx.entries]) if hashes: endpoint = f"webrtc:{self._config.node.quic_port}" asyncio.ensure_future(self._register_swarm(hashes, endpoint)) @@ -772,6 +962,10 @@ class NodeDaemon: log.info("Shutting down...") self._state["status"] = "stopping" + for handle in self._pending_broadcasts.values(): + handle.cancel() + self._pending_broadcasts.clear() + for task in self._tasks: task.cancel() for task in self._tasks: @@ -795,6 +989,9 @@ class NodeDaemon: for store in self._chat_stores.values(): await store.close() + for cache in self._index_caches.values(): + await cache.close() + for indexer in self._indexers: await indexer.stop() diff --git a/packages/meshbay-node/src/meshbay_node/indexer/__init__.py b/packages/meshbay-node/src/meshbay_node/indexer/__init__.py index bb6231b..c92c730 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/__init__.py @@ -1,5 +1,6 @@ """Directory indexer and Mesh Group Index.""" from .indexer import DirectoryIndexer from .group_index import GroupIndex +from .cache import IndexCache -__all__ = ["DirectoryIndexer", "GroupIndex"] +__all__ = ["DirectoryIndexer", "GroupIndex", "IndexCache"] diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py new file mode 100644 index 0000000..c26ddf1 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py @@ -0,0 +1,95 @@ +""" +MeshBay Node — persistent (path, size, mtime) -> hash cache, one per group. + +Without this, every node restart re-reads and re-hashes every file in every +root, even when nothing changed — measured at 23 minutes for a 114 GB library +on a USB hard drive. This cache lets a scan skip the read entirely for a file +whose size and mtime still match what was hashed last time. + +It is a path-keyed accelerator only. The GroupIndex itself stays keyed by +content hash (see indexer.py's note on why two identical files are one +entry) — this cache never changes that, it only avoids recomputing a hash +that has not changed. +""" + +import logging +from dataclasses import dataclass +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS files ( + path TEXT PRIMARY KEY, + mtime REAL NOT NULL, + size INTEGER NOT NULL, + hash TEXT NOT NULL, + type TEXT NOT NULL, + added_at INTEGER NOT NULL +); +""" + + +@dataclass +class CachedEntry: + hash: str + type: str + added_at: int + + +class IndexCache: + """Async SQLite (path, size, mtime) -> hash cache for one group.""" + + def __init__(self, db_path: Path): + self._db_path = db_path + self._db: aiosqlite.Connection | None = None + + async def open(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._db = await aiosqlite.connect(str(self._db_path)) + await self._db.executescript(_SCHEMA) + await self._db.commit() + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None + + async def __aenter__(self): + await self.open() + return self + + async def __aexit__(self, *_): + await self.close() + + async def lookup(self, path: str, size: int, mtime: float) -> CachedEntry | None: + """ + A cache hit requires an EXACT match on both size and mtime. A mtime + touched without a content change is a false negative (an unnecessary + rehash) — accepted, since the alternative (trusting a stale hash) is + a silent wrong answer instead of an occasional wasted read. + """ + async with self._db.execute( + "SELECT hash, type, added_at FROM files " + "WHERE path = ? AND size = ? AND mtime = ?", + (path, size, mtime)) as cur: + row = await cur.fetchone() + return CachedEntry(hash=row[0], type=row[1], added_at=row[2]) if row else None + + async def put(self, path: str, size: int, mtime: float, hash: str, + type: str, added_at: int) -> None: + """ + Written only once a file has been hashed in full — never partway + through — so a crash mid-hash leaves no stale/partial row behind: the + next scan simply finds no cache entry and hashes the file again. + """ + await self._db.execute( + "INSERT INTO files (path, mtime, size, hash, type, added_at) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT(path) DO UPDATE SET " + "mtime = excluded.mtime, size = excluded.size, hash = excluded.hash, " + "type = excluded.type, added_at = excluded.added_at", + (path, mtime, size, hash, type, added_at)) + await self._db.commit() diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index 5dbdc5d..ec98667 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -86,6 +86,23 @@ class GroupIndex: def count(self) -> int: return len(self._entries) + def entries_by_id(self) -> dict: + """A snapshot copy, for diff() to compare a later version against — + see daemon.py _on_index_change, the only caller.""" + return dict(self._entries) + + @classmethod + def _snapshot(cls, group_id: str, sk_node: Ed25519PrivateKey, gek: bytes | None, + version: int, entries_by_id: dict) -> "GroupIndex": + """ + A lightweight stand-in for diff()'s `previous` argument — never + serialized or sent anywhere, just a comparison point built from an + earlier entries_by_id() snapshot rather than a live GroupIndex. + """ + idx = cls(group_id=group_id, sk_node=sk_node, gek=gek, version=version) + idx._entries = dict(entries_by_id) + return idx + # ── Serialisation ───────────────────────────────────────────────────────── def serialize(self) -> bytes: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 482b556..b479f7e 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -26,6 +26,7 @@ import asyncio import logging import time from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass from pathlib import Path from typing import Callable, Awaitable @@ -36,6 +37,7 @@ from watchdog.observers import Observer from meshbay_common.paths import fold, find_fold_collisions, long_path from meshbay_common.protocol import IndexEntry +from meshbay_node.indexer.cache import IndexCache from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import Root, RootSet @@ -75,6 +77,27 @@ def _is_indexable(path: Path) -> bool: _HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks +@dataclass +class IndexProgress: + """ + A snapshot of "is this indexer mid-scan right now", for the status shown + to the operator (Create Group wizard, adding a directory) and pushed to + connected members (a presence dot, never anything more specific — see + daemon.py/webrtc_server.py). Reset per _scan_root() call rather than + accumulated across a group's roots: the consumers that matter always + watch exactly one root being scanned. + + Mutated only from the asyncio loop thread (the hashing itself runs in an + executor thread, but never touches this), so no lock is needed. + """ + scanning: bool = False + scanned_bytes: int = 0 + total_bytes: int = 0 + # Basename only, deliberately not the full path — enough to show progress + # without broadcasting the operator's directory structure. + current_dir: str = "" + + def _virtual_dir(root: Root, file_path: Path) -> str: """ The directory a file appears in, as members see it: `"Films/2024"`. @@ -87,6 +110,17 @@ def _virtual_dir(root: Root, file_path: Path) -> str: return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}" +def _walk_root(root: Root) -> list[Path]: + """ + Blocking directory walk — always run via an executor, never awaited + directly in the asyncio loop. A tree of tens of thousands of files (or + one on a slow network share) can take seconds; run inline, that stalls + every other thing the daemon is doing — WebRTC sessions, chat, the admin + UI — for as long as it takes. + """ + return [p for p in root.path.rglob("*") if p.is_file()] + + def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: """Compute IndexEntry for a file. Blocking — run in executor. Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.) @@ -136,8 +170,17 @@ class DirectoryIndexer: # How often to re-check which roots are readable and reconcile the index # against what is actually on disk. Not a poll for changes — a backstop for # the events the OS did not deliver, and the way a re-plugged drive is - # noticed. - RECONCILE_SECS = 60.0 + # noticed. Watchdog already covers the common case in real time, so this + # does not need to run often to do its job; it backs off further still + # (see _reconcile_loop) when nothing has changed for a while, and a + # per-group operator setting (roster.py SETTING_RECONCILE_INTERVAL) can + # override the starting point. + DEFAULT_RECONCILE_SECS = 600.0 # 10 min + RECONCILE_BACKOFF_CAP = 7200.0 # 2 h — never sleeps longer than this + # How long to wait after the last event on a given path before acting on + # it — several writes to the same file in quick succession (a slow copy + # in several passes) collapse into one hash instead of one per write. + DEFAULT_DEBOUNCE_SECS = 2.0 def __init__( self, @@ -146,20 +189,43 @@ class DirectoryIndexer: sk_node: Ed25519PrivateKey, gek: bytes | None, on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None, + cache: IndexCache | None = None, + reconcile_secs: float = DEFAULT_RECONCILE_SECS, + debounce_secs: float = DEFAULT_DEBOUNCE_SECS, ): self.roots = roots self.group_id = group_id self.sk_node = sk_node self.gek = gek self.on_change = on_change + self.reconcile_secs = reconcile_secs + self.debounce_secs = debounce_secs + # Current backoff delay — starts at reconcile_secs, doubles on every + # tick that finds nothing changed (up to RECONCILE_BACKOFF_CAP), and + # resets the moment something real happens (a change, or a peer + # connecting — see note_activity()). + self._reconcile_delay = reconcile_secs + # Path -> (size, mtime, hash) accelerator, so a restart does not have + # to re-read a file it already hashed last time (see cache.py). None + # in tests that do not care about it — every hash is then a miss. + self._cache = cache self._index = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek) self._index.roots = roots.describe() - self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer") + # One worker, deliberately, not a real pool: hashing two files at once + # buys nothing here and can cost a lot. It only offloads the blocking + # read+hash off the asyncio loop; it was never used for concurrency — + # every call site awaits one run_in_executor before starting the next + # (see _hash_or_cached below) — and measured on a spinning USB drive, + # two interleaved multi-GB reads would seek-thrash against each other + # rather than go faster. Left at 1 so the number does not promise a + # concurrency this code never provided. + self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="indexer") self._observer: Observer | None = None self._loop: asyncio.AbstractEventLoop | None = None self._reconciler: asyncio.Task | None = None self._pending_timers: dict[str, asyncio.TimerHandle] = {} + self.progress = IndexProgress() @property def index(self) -> GroupIndex: @@ -204,21 +270,77 @@ class DirectoryIndexer: async def _scan_root(self, root: Root) -> int: log.info("Scanning %s (root %r) ...", root.path, root.name) - loop = asyncio.get_event_loop() count = 0 + loop = asyncio.get_event_loop() try: - files = [p for p in root.path.rglob("*") if p.is_file()] + files = await loop.run_in_executor(self._executor, _walk_root, root) except OSError as e: log.warning("Cannot scan root %r: %s", root.name, e) return 0 - for file_path in files: - entry = await loop.run_in_executor( - self._executor, _scan_file, root, file_path) - if entry: - self._index.add_entry(entry) - count += 1 + + # Sizes up front, off the same listing that already walked the tree — + # the progress bar's denominator, not a second pass over the disk. + sized: list[tuple[Path, int]] = [] + for p in files: + try: + sized.append((p, p.stat().st_size)) + except OSError: + continue + + self.progress.scanning = True + self.progress.scanned_bytes = 0 + self.progress.total_bytes = sum(size for _, size in sized) + self.progress.current_dir = "" + try: + for file_path, size in sized: + self.progress.current_dir = file_path.parent.name + entry = await self._hash_or_cached(root, file_path) + self.progress.scanned_bytes += size + if entry: + self._index.add_entry(entry) + count += 1 + finally: + # Must run even if a hash/IO error propagates out of the loop + # above — an indexing state that never turns back off is worse + # than the scan itself failing. + self.progress.scanning = False + self.progress.current_dir = "" return count + async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: + """ + Cache-aware replacement for a bare _scan_file() call: skips the + content read entirely when this path's (size, mtime) still match + what was hashed last time — the difference between a redundant full + rehash of a 100+ GB library on every restart and a stat()-only pass. + The only place that decides to actually read a file's bytes. + """ + if not _is_indexable(file_path): + return None + try: + st = file_path.stat() + except OSError: + return None + + if self._cache is not None: + cached = await self._cache.lookup(str(file_path), st.st_size, st.st_mtime) + if cached is not None: + return IndexEntry( + id=cached.hash, + name=file_path.name, + path=_virtual_dir(root, file_path), + size=st.st_size, + type=cached.type, + added_at=cached.added_at, + ) + + loop = asyncio.get_event_loop() + entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) + if entry and self._cache is not None: + await self._cache.put(str(file_path), st.st_size, st.st_mtime, + entry.id, entry.type, entry.added_at) + return entry + def _report_collisions(self) -> None: """ Names that are the same file on a case-insensitive filesystem. @@ -324,20 +446,38 @@ class DirectoryIndexer: async def _reconcile_loop(self) -> None: while True: try: - await asyncio.sleep(self.RECONCILE_SECS) - await self.reconcile() + await asyncio.sleep(self._reconcile_delay) + changed = await self.reconcile() + if changed: + self._reconcile_delay = self.reconcile_secs + else: + self._reconcile_delay = min( + self._reconcile_delay * 2, self.RECONCILE_BACKOFF_CAP) except asyncio.CancelledError: raise except Exception: log.exception("Reconcile failed — continuing") - async def reconcile(self) -> None: + def note_activity(self) -> None: + """ + Called when something makes a prompt reconcile worth having again — + today, a peer completing the handshake for this group + (webrtc_server.py). Someone is looking, so the backstop should be at + its normal cadence rather than however far backoff had stretched it. + """ + self._reconcile_delay = self.reconcile_secs + + async def reconcile(self) -> bool: """ Re-check availability, and rescan roots that came back. The only place a root's entries are dropped: when the root is readable and the files are genuinely gone. A root that is not readable is left untouched, which is the whole point. + + Returns whether anything actually changed — _reconcile_loop uses this + to back off when a pass finds nothing to do, rather than running at + the same cadence forever regardless of how quiet the root is. """ changed = self.roots.refresh_availability() touched = False @@ -367,6 +507,8 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + return touched + async def _sweep_available_roots(self) -> bool: """ Catch what the watcher missed: files gone, and files never announced. @@ -375,24 +517,18 @@ class DirectoryIndexer: absent has nothing to compare against, and comparing anyway is exactly the mistake this module exists to avoid. """ - loop = asyncio.get_event_loop() changed = False + loop = asyncio.get_event_loop() for root in self.roots: if not root.available: continue try: - on_disk = {p.resolve() for p in root.path.rglob("*") - if _is_indexable(p)} + on_disk, known = await loop.run_in_executor( + self._executor, self._sweep_scan_root, root) except OSError as e: log.warning("Cannot reconcile root %r: %s", root.name, e) continue - known: dict[Path, str] = {} - for entry in self._entries_under(root): - abs_path = self._entry_path(root, entry) - if abs_path: - known[abs_path] = entry.id - for missing in set(known) - on_disk: # Duplicate content is handled without a special case here: the # entry goes, and the add loop below re-indexes the surviving @@ -404,26 +540,46 @@ class DirectoryIndexer: log.info("Reconcile: %s is gone", missing) changed = True - for added in on_disk - set(known): - entry = await loop.run_in_executor( - self._executor, _scan_file, root, added) - if not entry: - continue - # The index is keyed by **content hash**, so two identical files - # at two paths are one entry and the path comparison above - # cannot see the second. Adding it anyway rewrites that entry's - # path every cycle, bumps the version, and pushes an index - # update to every connected peer once a minute — for ever. - # Measured on a live node: `clip.mp4` present at the root and in - # uploads/ with the same bytes. - if self._index.get_entry(entry.id) is not None: - log.debug("Reconcile: %s duplicates content already indexed " - "as %s — leaving the index alone", - added, entry.id[:8]) - continue - self._index.add_entry(entry) - log.info("Reconcile: %s appeared (missed event)", added) - changed = True + added_paths = on_disk - set(known) + if not added_paths: + continue + + added_sized: list[tuple[Path, int]] = [] + for p in added_paths: + try: + added_sized.append((p, p.stat().st_size)) + except OSError: + added_sized.append((p, 0)) + + self.progress.scanning = True + self.progress.scanned_bytes = 0 + self.progress.total_bytes = sum(size for _, size in added_sized) + self.progress.current_dir = "" + try: + for added, size in added_sized: + self.progress.current_dir = added.parent.name + entry = await self._hash_or_cached(root, added) + self.progress.scanned_bytes += size + if not entry: + continue + # The index is keyed by **content hash**, so two identical + # files at two paths are one entry and the path comparison + # above cannot see the second. Adding it anyway rewrites + # that entry's path every cycle, bumps the version, and + # pushes an index update to every connected peer once a + # minute — for ever. Measured on a live node: `clip.mp4` + # present at the root and in uploads/ with the same bytes. + if self._index.get_entry(entry.id) is not None: + log.debug("Reconcile: %s duplicates content already " + "indexed as %s — leaving the index alone", + added, entry.id[:8]) + continue + self._index.add_entry(entry) + log.info("Reconcile: %s appeared (missed event)", added) + changed = True + finally: + self.progress.scanning = False + self.progress.current_dir = "" return changed def _entries_under(self, root: Root) -> list[IndexEntry]: @@ -444,6 +600,21 @@ class DirectoryIndexer: except OSError: return None + def _sweep_scan_root(self, root: Root) -> tuple[set[Path], dict[Path, str]]: + """ + Blocking: the two disk-touching pieces of one reconcile pass for a + root, bundled so both run together in the executor rather than in + the asyncio loop — the tree walk, and resolving the real path of + every entry already known under this root (one syscall each). + """ + on_disk = {p.resolve() for p in root.path.rglob("*") if _is_indexable(p)} + known: dict[Path, str] = {} + for entry in self._entries_under(root): + abs_path = self._entry_path(root, entry) + if abs_path: + known[abs_path] = entry.id + return on_disk, known + def _restart_observer(self) -> None: """Re-schedule watches after roots appeared or disappeared.""" if self._observer: @@ -454,8 +625,6 @@ class DirectoryIndexer: # ── Internal update ─────────────────────────────────────────────────────── - _DEBOUNCE_SECS = 2.0 - def _schedule_update(self, file_path: Path, deleted: bool = False) -> None: """Called from watchdog thread — schedule debounced async update.""" if not self._loop: @@ -472,7 +641,7 @@ class DirectoryIndexer: self._pending_timers.pop(key, None) asyncio.ensure_future(self._update_entry(file_path, deleted)) - self._pending_timers[key] = self._loop.call_later(self._DEBOUNCE_SECS, fire) + self._pending_timers[key] = self._loop.call_later(self.debounce_secs, fire) def _remove_by_path(self, root: Root, file_path: Path) -> None: """Remove any existing entries that point at this file.""" @@ -511,9 +680,7 @@ class DirectoryIndexer: self._remove_by_path(root, file_path) if not deleted: - loop = asyncio.get_event_loop() - entry = await loop.run_in_executor( - self._executor, _scan_file, root, file_path) + entry = await self._hash_or_cached(root, file_path) if entry: self._index.add_entry(entry) log.debug("Indexed: %s (%s, %d bytes)", diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index c9d862a..154813d 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -22,6 +22,7 @@ in the adapter. from __future__ import annotations +import asyncio import logging import re from dataclasses import asdict @@ -730,12 +731,69 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict: return {"apps": apps, "group_id": group_id} +# ── Scan settings ──────────────────────────────────────────────────────────── + +async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float, + debounce_secs: float) -> dict: + """ + How often the indexer's reconciliation backstop runs, and how long a + changed file is left alone before being hashed (indexer.py + DirectoryIndexer). Persisted like set_member_upload/set_enabled_apps — + but there is also a *live* DirectoryIndexer object to update, since it + reads these once at construction and runs its own background loop with + them rather than consulting groups_ctx on every use. + """ + roster = _roster(state) + await roster.set_scan_settings(group_id, reconcile_interval_secs, debounce_secs, + set_by=state.get("node_user_id", "")) + indexer = state.get("indexers", {}).get(group_id) + if indexer: + indexer.reconcile_secs = reconcile_interval_secs + indexer.debounce_secs = debounce_secs + # Apply the new interval now rather than after whatever backoff had + # already stretched the wait to. + indexer.note_activity() + # Optional, unlike _group_ctx(): a group can be persisted here before it + # is hot-loaded (or in a test that only cares about the roster/indexer + # side), and that must not turn a successful write into a 404. + ctx = state.get("groups_ctx", {}).get(group_id) + if ctx is not None: + ctx["reconcile_interval_secs"] = reconcile_interval_secs + ctx["debounce_secs"] = debounce_secs + log.info("Scan settings for group %s: reconcile=%.0fs debounce=%.0fs", + group_id[:8], reconcile_interval_secs, debounce_secs) + return {"reconcile_interval_secs": reconcile_interval_secs, + "debounce_secs": debounce_secs, "group_id": group_id} + + # ── Reload ────────────────────────────────────────────────────────────────── async def reload_config(state: dict) -> dict: - """Hot-reload node.toml without dropping connections.""" + """Hot-reload node.toml without dropping connections. Blocks until the + reload actually finishes — see start_reload for why the loopback route + uses that instead.""" reload_fn = state.get("reload_fn") if not reload_fn: raise OpError("Reload not available", status=503) await reload_fn() return {"status": "reloaded"} + + +async def start_reload(state: dict) -> dict: + """ + Same as reload_config, but does not wait for the reload to finish. + + The loopback route uses this one: the Electron bridge caps every call at + a fixed 30s (main.js node:call), and hot-loading a brand-new group runs + its full initial scan synchronously inside _reload_config_inner() + (daemon.py) before that coroutine returns — minutes, not seconds, on a + real library (found against a 45 GB group on the same slow disk the + StarWars benchmark used). The reload keeps running on the daemon's own + event loop either way; add_root/remove_root below already fire it the + same way for exactly this reason. + """ + reload_fn = state.get("reload_fn") + if not reload_fn: + raise OpError("Reload not available", status=503) + asyncio.ensure_future(reload_fn()) + return {"status": "reloading"} diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index 1cc8cae..6eb7651 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -599,6 +599,39 @@ class Roster: json.dumps(sorted(apps)), set_by) return apps + # How often the indexer's reconciliation backstop runs, and how long it + # waits after the last change on a file before hashing it. Unset means + # the indexer's own defaults — an existing group's behaviour must not + # change because a node was upgraded. See indexer.py DirectoryIndexer + # for what these actually do and why the defaults are what they are. + SETTING_RECONCILE_INTERVAL = "reconcile_interval_secs" + SETTING_DEBOUNCE_SECS = "debounce_secs" + + async def scan_settings(self, group_id: str) -> dict: + # Imported here, not at module load: roster.py is loaded before the + # indexer package during startup, and this is the only place the two + # need each other's names. + from meshbay_node.indexer.indexer import DirectoryIndexer + + reconcile = await self.get_setting(group_id, self.SETTING_RECONCILE_INTERVAL) + debounce = await self.get_setting(group_id, self.SETTING_DEBOUNCE_SECS) + return { + "reconcile_interval_secs": ( + float(reconcile) if reconcile is not None + else DirectoryIndexer.DEFAULT_RECONCILE_SECS), + "debounce_secs": ( + float(debounce) if debounce is not None + else DirectoryIndexer.DEFAULT_DEBOUNCE_SECS), + } + + async def set_scan_settings(self, group_id: str, reconcile_interval_secs: float, + debounce_secs: float, set_by: str = "") -> dict: + await self.set_setting(group_id, self.SETTING_RECONCILE_INTERVAL, + str(float(reconcile_interval_secs)), set_by) + await self.set_setting(group_id, self.SETTING_DEBOUNCE_SECS, + str(float(debounce_secs)), set_by) + return await self.scan_settings(group_id) + async def create_invite( self, group_id: str, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index b6f572a..fa6c3e9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -63,6 +63,7 @@ from meshbay_common.adminop import ( OP_MEMBER_UNPIN, OP_MEMBER_UPLOAD, OP_APPS_ENABLED, + OP_SET_SCAN_SETTINGS, OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, @@ -85,6 +86,7 @@ from meshbay_common.join import ( from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import ops from meshbay_node.roots import ( RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, @@ -409,6 +411,8 @@ class WebRTCPeerSession: self._do_member_upload(msg) elif mtype == MNP.APPS_ENABLED: self._do_apps_enabled(msg) + elif mtype == MNP.SET_SCAN_SETTINGS: + self._do_set_scan_settings(msg) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: @@ -659,6 +663,20 @@ class WebRTCPeerSession: # setting (or one whose context has not loaded it yet) hides # nothing. "enabled_apps": list(self._group_ctx().get("enabled_apps") or []), + # So a client that connects mid-scan shows the indexing state + # immediately, instead of waiting for the next periodic + # INDEX_PROGRESS push. Never a path or filename — see + # IndexProgress in indexer.py. + "indexing": self._indexing_status(), + # Current values only — not enforced from here, just shown to + # the operator in Settings so the number on screen matches what + # the indexer is actually doing (set_scan_settings, ops.py). + "scan_settings": { + "reconcile_interval_secs": self._group_ctx().get( + "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS), + "debounce_secs": self._group_ctx().get( + "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS), + }, } if node_user_id: ack["node_user_id"] = node_user_id @@ -668,6 +686,13 @@ class WebRTCPeerSession: self._send(ack) self._audit("handshake") + # Someone is here now — reconcile's backstop should be prompt again + # rather than however far its backoff had stretched while nobody + # was connected (indexer.py DirectoryIndexer.note_activity). + note_activity = self._group_ctx().get("note_activity") + if note_activity: + note_activity() + async def _do_gek_bundle_fetch(self) -> None: """Serve the caller's wrapped GEK bundle during the handshake window.""" bundle_store = self._ctx.get("bundle_store") @@ -1648,6 +1673,69 @@ class WebRTCPeerSession: except Exception: pass + # Reconcile's backstop and the watchdog debounce (indexer.py + # DirectoryIndexer) — how hard the node works on the operator's own + # disk, not a member-facing permission. Signed for the same reason as + # apps_enabled: consistency of the authorization model, not because a + # wrong value here is itself dangerous. + MIN_RECONCILE_SECS = 10.0 + MAX_RECONCILE_SECS = 24 * 3600.0 + MIN_DEBOUNCE_SECS = 0.0 + MAX_DEBOUNCE_SECS = 300.0 + + def _do_set_scan_settings(self, msg: dict) -> None: + try: + reconcile = float(msg.get("reconcile_interval_secs")) + debounce = float(msg.get("debounce_secs")) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid scan settings"}) + return + if not (self.MIN_RECONCILE_SECS <= reconcile <= self.MAX_RECONCILE_SECS): + self._send({"type": "error", + "detail": f"reconcile_interval_secs must be between " + f"{self.MIN_RECONCILE_SECS:.0f} and " + f"{self.MAX_RECONCILE_SECS:.0f}"}) + return + if not (self.MIN_DEBOUNCE_SECS <= debounce <= self.MAX_DEBOUNCE_SECS): + self._send({"type": "error", + "detail": f"debounce_secs must be between " + f"{self.MIN_DEBOUNCE_SECS:.0f} and " + f"{self.MAX_DEBOUNCE_SECS:.0f}"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}") + + async def _admin_exec_set_scan_settings( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + try: + reconcile_s, debounce_s = pending["subject"].split(",") + reconcile, debounce = float(reconcile_s), float(debounce_s) + except (ValueError, KeyError): + self._send({"type": "error", "detail": "Invalid scan settings"}) + return + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"set_scan_settings:{pending['subject']}") + return + try: + result = await self._run_op( + ops.set_scan_settings, self._group_id or "", reconcile, debounce) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("set_scan_settings", pending["subject"]) + + notice = {"type": MNP.SET_SCAN_SETTINGS_ACK, "v": MNP_VERSION, **result} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + # ── Node management (D5) ───────────────────────────────────────────────── async def _do_node_status(self, msg: dict) -> None: @@ -2012,6 +2100,23 @@ class WebRTCPeerSession: return self._ctx["groups"][self._group_id] return self._ctx + def _indexing_status(self) -> dict: + """ + {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the + handshake ack and INDEX_PROGRESS pushes — never a path or filename, + that stays local to the operator's own admin UI. Absent "progress" + (context not loaded, or a group with no indexer at all) reads as + idle rather than erroring. + """ + progress = self._group_ctx().get("progress") + if progress is None: + return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0} + return { + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + } + def _peer_registry(self) -> dict: """ Connected peers for THIS group only. @@ -2626,6 +2731,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_APPS_ENABLED: self._spawn( self._admin_exec_apps_enabled(pending, transcript, sig_bytes)) + elif pending["op"] == OP_SET_SCAN_SETTINGS: + self._spawn( + self._admin_exec_set_scan_settings(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_ADD: self._spawn( self._admin_exec_root_add(pending, transcript, sig_bytes)) diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index b505f25..d5b3f94 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -26,6 +26,7 @@ from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ from meshbay_node import ops from meshbay_node.config import DEFAULT_CONFIG_PATH +from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_common.crypto import generate_gek, wrap_gek_aes from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR @@ -319,6 +320,39 @@ def create_ui_app(state: dict) -> FastAPI: asyncio.ensure_future(reload_fn()) return result + # asyncio.ensure_future above schedules the reload (and whatever initial + # scan it triggers) on the daemon's own event loop — it has no link to + # this HTTP request or to any browser tab. Closing the client that made + # this call does not cancel it: the scan is the node's own background + # work, not something borrowed from the request that started it. + + @app.get("/api/groups/{group_id}/index-status") + async def index_status(group_id: str): + """ + Polled by the Create Group wizard and by "add a directory" in + Settings — the same source either way, since both just start a scan + on this group's indexer. `current_dir` is a basename only, and is + never sent over MNP (see IndexProgress in indexer.py) — this route + is loopback-only, for the operator's own screen. + + Reads state["indexers"] rather than groups_ctx: a brand-new group is + registered there before its (possibly long) initial scan runs, but + is only added to groups_ctx once that scan finishes (it is not yet + authorized for member connections either way — see _reload_config) + — this is precisely the window the wizard needs to watch. + """ + indexer = state.get("indexers", {}).get(group_id) + progress = indexer.progress if indexer else None + if progress is None: + return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, + "current_dir": ""} + return { + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + "current_dir": progress.current_dir, + } + # ── Upload toggle (operator only, localhost) ───────────────────────── @app.put("/api/groups/{group_id}/member-upload") @@ -327,11 +361,26 @@ def create_ui_app(state: dict) -> FastAPI: state, group_id, bool(payload.get("allowed", False)), )) + # ── Scan settings (operator only, localhost) ────────────────────────── + + @app.put("/api/groups/{group_id}/scan-settings") + async def set_scan_settings(group_id: str, payload: dict): + return await _op(lambda: ops.set_scan_settings( + state, group_id, + float(payload.get("reconcile_interval_secs", + DirectoryIndexer.DEFAULT_RECONCILE_SECS)), + float(payload.get("debounce_secs", + DirectoryIndexer.DEFAULT_DEBOUNCE_SECS)), + )) + # ── Reload config ──────────────────────────────────────────────────── @app.post("/api/reload") async def reload_config(): - return await _op(lambda: ops.reload_config(state)) + # start_reload, not reload_config: this must return before a + # brand-new group's synchronous initial scan finishes (minutes, not + # seconds, on a real library) — see ops.start_reload for why. + return await _op(lambda: ops.start_reload(state)) # ── Chat endpoints ─────────────────────────────────────────────────────── diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index ab1613d..b367a20 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -237,6 +237,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 # real value would make this test wait 0.5s daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=2) daemon._state["endpoint_hint"] = "node123" @@ -256,6 +257,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu daemon._webrtc = mock_webrtc await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) # let the coalescing timer fire mock_session._send.assert_called_once() msg = mock_session._send.call_args[0][0] @@ -288,6 +290,7 @@ async def test_daemon_index_change_registers_swarm_for_public_group( data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=2) daemon._state["endpoint_hint"] = "node123" @@ -317,6 +320,7 @@ async def test_daemon_index_change_skips_other_group_peers( data_dir=tmp_path / "data", ) daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 daemon._hub = AsyncMock() daemon._hub.register_swarm = AsyncMock(return_value=0) daemon._state["endpoint_hint"] = "node123" @@ -340,6 +344,145 @@ async def test_daemon_index_change_skips_other_group_peers( daemon._webrtc = mock_webrtc await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) same_group._send.assert_called_once() other_group._send.assert_not_called() + + +# ── INDEX_DELTA (phase 4) ──────────────────────────────────────────────────── + +def _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32, + visibility="private"): + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id=group_id, name="test-group", shared_dir=str(shared_dir), + visibility=visibility, quic_port=29010, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 + daemon._hub = AsyncMock() + daemon._hub.register_swarm = AsyncMock(return_value=0) + daemon._state["endpoint_hint"] = "node123" + return daemon + + +@pytest.mark.asyncio +async def test_first_broadcast_is_full_sync_second_is_delta(tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + first = session._send.call_args_list[0].args[0] + assert first["type"] == "index_sync" + assert len(first["entries"]) == indexer.index.count + + # Nothing actually changed in the index between the two calls, but + # _on_index_change does not know or care why it was called — the + # SECOND broadcast must still be a delta, now that there is a + # previous snapshot to diff against. + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + second = session._send.call_args_list[1].args[0] + assert second["type"] == "index_delta" + assert second["additions"] == [] + assert second["deletions"] == [] + + +@pytest.mark.asyncio +async def test_delta_reflects_additions_and_deletions(tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + removed_id = indexer.index.entries[0].id + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p1": session} + + await daemon._on_index_change(indexer) # first: full sync, establishes the snapshot + await asyncio.sleep(0.05) + + # A real change: one entry removed, one added. + indexer.index.remove_entry(removed_id) + from meshbay_common.protocol import IndexEntry + new_entry = IndexEntry(id="new-file-id", name="new.mp4", path="shared", + size=10, type="video", added_at=0) + indexer.index.add_entry(new_entry) + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + delta_msg = session._send.call_args_list[1].args[0] + assert delta_msg["type"] == "index_delta" + assert delta_msg["deletions"] == [removed_id] + assert [a["id"] for a in delta_msg["additions"]] == ["new-file-id"] + + +@pytest.mark.asyncio +async def test_a_burst_of_changes_produces_one_broadcast(tmp_path, shared_dir, gek): + """Coalescing: several _on_index_change calls in quick succession (one + per debounced watchdog event) must collapse into a single push.""" + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek) + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p1": session} + + for _ in range(5): + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + session._send.assert_called_once() + + +@pytest.mark.asyncio +async def test_swarm_registration_only_sends_new_hashes_after_the_first( + tmp_path, shared_dir, gek): + daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, visibility="public") + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="a" * 32, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + total_files = indexer.index.count + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + assert len(daemon._hub.register_swarm.call_args_list[0].args[0]) == total_files + + from meshbay_common.protocol import IndexEntry + indexer.index.add_entry(IndexEntry(id="new-file-id", name="new.mp4", + path="shared", size=10, type="video", + added_at=0)) + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + assert daemon._hub.register_swarm.call_count == 2 + assert daemon._hub.register_swarm.call_args_list[1].args[0] == ["new-file-id"], \ + "only the newly added hash must be (re-)registered, not the whole library" diff --git a/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py new file mode 100644 index 0000000..7cb74cb --- /dev/null +++ b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py @@ -0,0 +1,329 @@ +""" +Adding a group (Create Group wizard, or "add a directory" to an existing +one) fires `_reload_config()` without awaiting it (`asyncio.ensure_future`, +ui/app.py) — the request handler, and whatever browser tab triggered it, +return immediately. This is deliberate: the initial scan behind it can take +a very long time (measured at 23 minutes for a 114 GB library on a slow +disk), and none of that work belongs to the HTTP request or the WebRTC +session that happened to start it. + +This test proves the scan is genuinely independent of its caller: it starts +the reload the same way the real endpoint does — schedules it and does not +await it, standing in for "the browser tab that made the call was closed" — +then does something else, and only afterwards checks that the reload +finished and the new group became available on its own. +""" + +import asyncio +import base64 +import os + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from unittest.mock import AsyncMock, MagicMock, patch + +from meshbay_node import ops +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +import meshbay_node.indexer.indexer as indexer_mod + + +def _free_port() -> int: + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _toml(data_dir, first_group_dir, second_group_id=None, second_group_dir=None) -> str: + # data_dir MUST come before any [section] header — TOML has no notion of + # "back to top-level" once a table is open, so a bare `key = value` line + # placed after [node] becomes node.data_dir, not the top-level data_dir + # load_config() actually reads. Silently falls back to the real default + # (~/.local/share/meshbay) instead of erroring, which is exactly how this + # test once ran a whole daemon — including _shutdown()'s unlink of + # ui-token — against the developer's real, already-running node. + text = f""" +data_dir = "{data_dir}" + +[hub] +url = "http://localhost:9999" +username = "testuser" + +[node] +quic_port = {_free_port()} +ui_port = {_free_port()} + +[[groups]] +id = "{"a" * 32}" +name = "first" +shared_dir = "{first_group_dir}" +visibility = "private" +""" + if second_group_id: + text += f""" +[[groups]] +id = "{second_group_id}" +name = "slow-new-group" +shared_dir = "{second_group_dir}" +visibility = "private" +""" + return text + + +def _mock_keystore_keys(sk_ed): + sk_x = X25519PrivateKey.generate() + pk_x_raw = sk_x.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + mock_keys = MagicMock() + mock_keys.sk_ed25519 = sk_ed + mock_keys.pk_ed25519_b64 = "test" + mock_keys.sk_x25519 = sk_x + mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode() + return mock_keys + + +@pytest.mark.asyncio +async def test_hot_loaded_group_finishes_scanning_without_anyone_awaiting_the_reload( + tmp_path): + first_dir = tmp_path / "first" + first_dir.mkdir() + (first_dir / "readme.txt").write_bytes(b"hello") + + second_dir = tmp_path / "second" + second_dir.mkdir() + for i in range(3): + (second_dir / f"file{i}.bin").write_bytes(os.urandom(64)) + second_group_id = "b" * 32 + + data_dir = tmp_path / "data" + config_path = tmp_path / "node.toml" + config_path.write_text(_toml(data_dir, first_dir)) + + from meshbay_node.config import load_config + daemon = NodeDaemon(load_config(config_path), config_path=config_path) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = _mock_keystore_keys(sk_node) + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = sk_node.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + # Slow the second group's hashing down (a stand-in for a large/slow + # library) so there is a real window in which "nobody is awaiting this" + # actually means something, without needing a genuinely huge file. + real_scan_file = indexer_mod._scan_file + + def slow_scan_file(root, path): + import time + time.sleep(0.15) + return real_scan_file(root, path) + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch.object(indexer_mod, "_scan_file", slow_scan_file): + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=15) + except (asyncio.TimeoutError, Exception): + pass + + run_task = asyncio.create_task(run_daemon()) + try: + for _ in range(50): + if daemon._state.get("status") == "running": + break + await asyncio.sleep(0.05) + assert daemon._state["status"] == "running" + assert second_group_id not in daemon._state.get("groups_ctx", {}) + + # Add the second group to the config on disk, the way the wizard's + # attach + /api/reload would leave it, then fire the reload exactly + # as ui/app.py does: scheduled, NOT awaited. + config_path.write_text(_toml(data_dir, first_dir, + second_group_id, second_dir)) + reload_task = asyncio.ensure_future(daemon._reload_config()) + + # Stand in for "the browser tab is gone": do something completely + # unrelated to the reload, and explicitly do not await it here. + await asyncio.sleep(0.01) + assert second_group_id not in daemon._state.get("groups_ctx", {}), \ + "the scan (3 files x 0.15s) cannot have finished yet" + + # Only now catch up with the background work, from a place that + # has no relationship to whatever originally triggered it. + await asyncio.wait_for(reload_task, timeout=5) + + assert second_group_id in daemon._state["groups_ctx"], \ + "the new group must be usable once its scan finishes, " \ + "regardless of whether anything was still watching the reload" + new_indexer = daemon._state["indexers"][second_group_id] + assert new_indexer.index.count == 3 + assert new_indexer.progress.scanning is False + finally: + shutdown_event.set() + await daemon._shutdown() + run_task.cancel() + try: + await run_task + except (asyncio.CancelledError, Exception): + pass + + +@pytest.mark.asyncio +async def test_group_scoped_ops_404_until_listed_then_succeed(tmp_path): + """ + The wizard's own sequence, reproduced against the real ops layer: attach + a brand-new group, fire the reload the way /api/reload now does + (ops.start_reload — scheduled, not awaited), and hit the group-scoped + calls that come right after in the UI (add a root, init the GEK) while + the scan is still running. + + Found live: "Attaching to node" no longer times out (ops.start_reload + returns immediately), but the very next wizard step then failed with + "Group not configured on this node" / "Group not hosted on this node" — + the group is not in daemon._state["config"].groups or ["groups_ctx"] + until _reload_config_inner() finishes, scan included, which is *after* + ops.start_reload has already returned. This locks in both halves: the + 404 while the scan runs, and success once ops.list_groups() actually + lists the group — the exact condition the wizard's own wait + (platform.waitForGroupHosted, app.js) polls for. + """ + first_dir = tmp_path / "first" + first_dir.mkdir() + (first_dir / "readme.txt").write_bytes(b"hello") + + second_dir = tmp_path / "second" + second_dir.mkdir() + for i in range(3): + (second_dir / f"file{i}.bin").write_bytes(os.urandom(64)) + second_group_id = "c" * 32 + extra_root_dir = tmp_path / "extra_root" + extra_root_dir.mkdir() + + data_dir = tmp_path / "data2" + config_path = tmp_path / "node2.toml" + config_path.write_text(_toml(data_dir, first_dir)) + + from meshbay_node.config import load_config + daemon = NodeDaemon(load_config(config_path), config_path=config_path) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = _mock_keystore_keys(sk_node) + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = sk_node.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + real_scan_file = indexer_mod._scan_file + + def slow_scan_file(root, path): + import time + time.sleep(0.15) + return real_scan_file(root, path) + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch.object(indexer_mod, "_scan_file", slow_scan_file): + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=15) + except (asyncio.TimeoutError, Exception): + pass + + run_task = asyncio.create_task(run_daemon()) + try: + for _ in range(50): + if daemon._state.get("status") == "running": + break + await asyncio.sleep(0.05) + assert daemon._state["status"] == "running" + + # The same file write ops.attach_group does (a raw text append), + # then the same fire-and-forget reload /api/reload now does. + config_path.write_text(_toml(data_dir, first_dir, + second_group_id, second_dir)) + reload_task = asyncio.ensure_future(ops.start_reload(daemon._state)) + + await asyncio.sleep(0.01) + listing = await ops.list_groups(daemon._state) + assert second_group_id not in [g["id"] for g in listing["groups"]], \ + "the scan (3 files x 0.15s) cannot have finished this fast" + + # Exactly the wizard's next two steps, hit mid-scan. + with pytest.raises(ops.OpError) as add_root_exc: + await ops.add_root(daemon._state, second_group_id, + str(extra_root_dir)) + assert add_root_exc.value.status == 404 + + with pytest.raises(ops.OpError) as gek_exc: + await ops.set_gek(daemon._state, second_group_id) + assert gek_exc.value.status == 404 + + # Now wait the way platform.waitForGroupHosted (app.js) does: + # poll list_groups(), not index-status, until the group is + # actually there. + for _ in range(100): + listing = await ops.list_groups(daemon._state) + if second_group_id in [g["id"] for g in listing["groups"]]: + break + await asyncio.sleep(0.05) + else: + pytest.fail("group never appeared in list_groups()") + + await asyncio.wait_for(reload_task, timeout=5) + + # Both calls that 404'd above must now succeed. + add_result = await ops.add_root(daemon._state, second_group_id, + str(extra_root_dir)) + assert add_result["status"] == "added" + + gek_result = await ops.set_gek(daemon._state, second_group_id) + assert gek_result["status"] == "ok" + finally: + shutdown_event.set() + await daemon._shutdown() + run_task.cancel() + try: + await run_task + except (asyncio.CancelledError, Exception): + pass diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py new file mode 100644 index 0000000..db0e37e --- /dev/null +++ b/packages/meshbay-node/tests/test_index_cache.py @@ -0,0 +1,78 @@ +"""Tests for the (path, size, mtime) -> hash cache (indexer/cache.py).""" + +import pytest + +from meshbay_node.indexer.cache import IndexCache + + +@pytest.fixture +async def cache(tmp_path): + c = IndexCache(db_path=tmp_path / "index_cache.db") + await c.open() + yield c + await c.close() + + +@pytest.mark.asyncio +async def test_put_then_lookup_hits(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + + hit = await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) + + assert hit is not None + assert hit.hash == "abc123" + assert hit.type == "video" + assert hit.added_at == 42 + + +@pytest.mark.asyncio +async def test_lookup_misses_on_unknown_path(cache): + assert await cache.lookup("/lib/never-seen.mkv", size=1, mtime=1.0) is None + + +@pytest.mark.asyncio +async def test_lookup_misses_on_different_mtime(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + + assert await cache.lookup("/lib/a.mkv", size=1000, mtime=222.0) is None + + +@pytest.mark.asyncio +async def test_lookup_misses_on_different_size(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + + assert await cache.lookup("/lib/a.mkv", size=2000, mtime=111.0) is None + + +@pytest.mark.asyncio +async def test_put_overwrites_previous_row_for_same_path(cache): + await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="old", + type="video", added_at=1) + await cache.put("/lib/a.mkv", size=2000, mtime=222.0, hash="new", + type="video", added_at=2) + + assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None + hit = await cache.lookup("/lib/a.mkv", size=2000, mtime=222.0) + assert hit.hash == "new" + + +@pytest.mark.asyncio +async def test_cache_survives_reopen(tmp_path): + db_path = tmp_path / "index_cache.db" + + c1 = IndexCache(db_path=db_path) + await c1.open() + await c1.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123", + type="video", added_at=42) + await c1.close() + + c2 = IndexCache(db_path=db_path) + await c2.open() + hit = await c2.lookup("/lib/a.mkv", size=1000, mtime=111.0) + await c2.close() + + assert hit is not None + assert hit.hash == "abc123" diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py new file mode 100644 index 0000000..52cd78b --- /dev/null +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -0,0 +1,165 @@ +""" +Indexing status visible node -> client: handshake ack field, the loopback +status route for the Create Group wizard / "add a directory", and the +periodic INDEX_PROGRESS push to already-connected peers. Never the index +itself (see test_daemon.py for that) and never anything sent to the hub. +""" + +import asyncio + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from unittest.mock import MagicMock +from fastapi.testclient import TestClient + +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer.indexer import IndexProgress +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from meshbay_node.ui.app import create_ui_app + + +def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * 32): + index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + group_ctx = {"index": index} + if progress is not None: + group_ctx["progress"] = progress + session._ctx = {"groups": {group_id: group_ctx}} + session._group_id = group_id + return session + + +# ── _indexing_status() ─────────────────────────────────────────────────────── + +def test_indexing_status_defaults_idle_when_no_progress_tracked(): + session = _session_with_progress(None) + assert session._indexing_status() == { + "scanning": False, "scanned_bytes": 0, "total_bytes": 0} + + +def test_indexing_status_reflects_live_progress(): + progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000, + current_dir="StarWars") + session = _session_with_progress(progress) + + status = session._indexing_status() + + assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000} + assert "current_dir" not in status, \ + "the directory name is operator-local detail, never sent to a member" + + +# ── /api/groups/{id}/index-status (loopback) ──────────────────────────────── + +def _ui_client(state: dict) -> TestClient: + return TestClient(create_ui_app({"status": "running", "groups_ctx": {}, + "indexes": {}, **state})) + + +def test_index_status_route_idle_for_unknown_group(): + client = _ui_client({"indexers": {}}) + resp = client.get("/api/groups/unknown-group/index-status") + assert resp.status_code == 200 + assert resp.json() == {"scanning": False, "scanned_bytes": 0, + "total_bytes": 0, "current_dir": ""} + + +def test_index_status_route_reflects_indexer_progress(): + fake_indexer = MagicMock() + fake_indexer.progress = IndexProgress( + scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000, + current_dir="2024") + client = _ui_client({"indexers": {"g" * 32: fake_indexer}}) + + resp = client.get(f"/api/groups/{'g' * 32}/index-status") + + assert resp.json() == { + "scanning": True, "scanned_bytes": 1_000_000, + "total_bytes": 4_000_000_000, "current_dir": "2024", + } + + +# ── _push_index_progress / _progress_pusher ───────────────────────────────── + +def _daemon(tmp_path) -> NodeDaemon: + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(), + groups=[], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + return NodeDaemon(config) + + +@pytest.mark.asyncio +async def test_push_index_progress_only_reaches_same_group_peers(tmp_path): + daemon = _daemon(tmp_path) + + same_group = MagicMock() + same_group._group_id = "a" * 32 + same_group._send = MagicMock() + other_group = MagicMock() + other_group._group_id = "b" * 32 + other_group._send = MagicMock() + + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": same_group, "p2": other_group} + daemon._webrtc = mock_webrtc + + progress = IndexProgress(scanning=True, scanned_bytes=10, total_bytes=100) + daemon._push_index_progress("a" * 32, progress) + + same_group._send.assert_called_once() + msg = same_group._send.call_args[0][0] + assert msg["type"] == "index_progress" + assert msg["group_id"] == "a" * 32 + assert msg["scanning"] is True + assert msg["scanned_bytes"] == 10 + assert msg["total_bytes"] == 100 + other_group._send.assert_not_called() + + +@pytest.mark.asyncio +async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path): + daemon = _daemon(tmp_path) + + session = MagicMock() + session._group_id = "a" * 32 + session._send = MagicMock() + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + indexer = MagicMock() + indexer.group_id = "a" * 32 + indexer.progress = IndexProgress(scanning=True, scanned_bytes=0, total_bytes=100) + + task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05)) + try: + # Two ticks while still scanning. + await asyncio.sleep(0.12) + assert session._send.call_count >= 2 + assert all(c.args[0]["scanning"] is True for c in session._send.call_args_list) + + # Scan finishes between ticks. + indexer.progress.scanning = False + calls_before = session._send.call_count + await asyncio.sleep(0.07) + assert session._send.call_count == calls_before + 1, \ + "exactly one final push must follow the False transition" + assert session._send.call_args.args[0]["scanning"] is False + + # Nothing further once idle. + calls_after_final = session._send.call_count + await asyncio.sleep(0.15) + assert session._send.call_count == calls_after_final, \ + "no more pushes once idle and already reported" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index c304361..68ccc4c 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -9,7 +9,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import generate_gek -from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache +import meshbay_node.indexer.indexer as indexer_mod from conftest import one_root from meshbay_node.keystore import NodeKeys @@ -190,3 +191,290 @@ async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek): wire = indexer.index.serialize() recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) assert recovered.count == indexer.index.count + + +# ── Cache-aware scanning ─────────────────────────────────────────────────────── + +@pytest.fixture +async def index_cache(tmp_path): + c = IndexCache(db_path=tmp_path / "index_cache.db") + await c.open() + yield c + await c.close() + + +@pytest.mark.asyncio +async def test_second_scan_with_same_cache_hashes_nothing( + shared_dir, sk_node, gek, index_cache): + """ + The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same + on-disk cache) that finds every file's (size, mtime) unchanged must not + read a single byte of file content. + """ + first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await first.initial_scan() + assert first.index.count == 4 + + calls = [] + real_scan_file = indexer_mod._scan_file + + def spy(root, path): + calls.append(path) + return real_scan_file(root, path) + + indexer_mod._scan_file = spy + try: + second = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await second.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}" + assert second.index.count == first.index.count + assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries} + + +@pytest.mark.asyncio +async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache): + d = tmp_path / "shared" + d.mkdir() + f = d / "movie.mkv" + f.write_bytes(b"original content") + + first = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await first.initial_scan() + old_id = first.index.entries[0].id + + # Change both content and mtime, as any real edit would. + f.write_bytes(b"a completely different, longer payload") + os.utime(f, (time.time() + 5, time.time() + 5)) + + second = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await second.initial_scan() + + assert second.index.count == 1 + assert second.index.entries[0].id != old_id + + +@pytest.mark.asyncio +async def test_scan_interrupted_partway_leaves_only_completed_files_cached( + tmp_path, sk_node, gek, index_cache): + """ + A cache row is only ever written after a file is fully hashed (cache.py), + so a crash mid-scan cannot leave a stale/partial row — the next scan just + treats the not-yet-cached files as new, and finishes the job. + """ + d = tmp_path / "shared" + d.mkdir() + names = [f"file{i}.bin" for i in range(5)] + for i, name in enumerate(names): + (d / name).write_bytes(os.urandom(64) * (i + 1)) + + real_scan_file = indexer_mod._scan_file + hashed_before_crash = [] + + def crash_after_three(root, path): + if len(hashed_before_crash) >= 3: + raise RuntimeError("simulated crash mid-scan") + entry = real_scan_file(root, path) + hashed_before_crash.append(path) + return entry + + indexer_mod._scan_file = crash_after_three + try: + crashing = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + with pytest.raises(RuntimeError): + await crashing.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert len(hashed_before_crash) == 3 + + # A normal rescan (same cache) must still end up with all 5 files + # correctly indexed, hashing only the ones the crash never got to. + calls = [] + + def spy(root, path): + calls.append(path) + return real_scan_file(root, path) + + indexer_mod._scan_file = spy + try: + resumed = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek, cache=index_cache) + await resumed.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert resumed.index.count == 5 + assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}" + + +# ── Progress state ─────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek) + assert indexer.progress.scanning is False + + await indexer.initial_scan() + + total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file()) + assert indexer.progress.scanning is False, "must end idle, not stuck scanning" + assert indexer.progress.scanned_bytes == total_size + assert indexer.progress.total_bytes == total_size + + +@pytest.mark.asyncio +async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek): + d = tmp_path / "shared" + d.mkdir() + (d / "a.bin").write_bytes(os.urandom(64)) + (d / "b.bin").write_bytes(os.urandom(64)) + + real_scan_file = indexer_mod._scan_file + + def boom(root, path): + raise RuntimeError("simulated failure mid-scan") + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek) + indexer_mod._scan_file = boom + try: + with pytest.raises(RuntimeError): + await indexer.initial_scan() + finally: + indexer_mod._scan_file = real_scan_file + + assert indexer.progress.scanning is False, \ + "an exception mid-scan must not leave the scanning flag stuck on" + + +# ── Off-loop directory walks, reconcile backoff ───────────────────────────── + +@pytest.mark.asyncio +async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek): + d = tmp_path / "shared" + d.mkdir() + (d / "f.bin").write_bytes(b"x") + + real_walk = indexer_mod._walk_root + + def slow_walk(root): + time.sleep(0.2) + return real_walk(root) + + indexer_mod._walk_root = slow_walk + ticks = 0 + + async def ticker(): + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + ticker_task = asyncio.create_task(ticker()) + try: + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + finally: + indexer_mod._walk_root = real_walk + ticker_task.cancel() + try: + await ticker_task + except asyncio.CancelledError: + pass + + assert ticks >= 5, ( + "the event loop must keep running other tasks while a directory " + f"walk is in progress in the executor — only {ticks} ticks happened " + "during a 0.2s walk") + + +@pytest.mark.asyncio +async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=0.01) + await indexer.initial_scan() + assert indexer._reconcile_delay == 0.01 + + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.sleep(0.2) + assert indexer._reconcile_delay > 0.01, \ + "several no-change ticks must have grown the delay" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + # A low, instance-only cap so the clamp is observable without waiting + # through dozens of real doublings up to the real 7200s ceiling. + indexer.RECONCILE_BACKOFF_CAP = 0.05 + indexer._reconcile_delay = 0.04 + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.sleep(0.15) + assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_note_activity_resets_backoff(shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=10.0) + await indexer.initial_scan() + indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff + + indexer.note_activity() + + assert indexer._reconcile_delay == 10.0 + + +@pytest.mark.asyncio +async def test_reconcile_backoff_resets_when_something_actually_changes( + shared_dir, sk_node, gek): + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek, reconcile_secs=0.02) + await indexer.initial_scan() + indexer._reconcile_delay = 5.0 # pretend it had already backed off a lot + + # A fake reconcile() rather than a real filesystem change: the real + # sweep's timing (disk I/O, the executor round trip) would race against + # this test's own sleeps. What matters here is only _reconcile_loop's + # reaction to "something changed", not reconcile()'s own detection logic + # — that is covered separately (test_root_availability.py). + reconciled_once = asyncio.Event() + + async def fake_reconcile(): + reconciled_once.set() + return True + + indexer._reconcile_delay = 0.01 + indexer.reconcile = fake_reconcile + + task = asyncio.create_task(indexer._reconcile_loop()) + try: + await asyncio.wait_for(reconciled_once.wait(), timeout=2.0) + assert indexer._reconcile_delay == 0.02, \ + "a real change must reset the delay back to the base interval" + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py index d2ccc0d..83758ae 100644 --- a/packages/meshbay-node/tests/test_ops.py +++ b/packages/meshbay-node/tests/test_ops.py @@ -9,6 +9,7 @@ appeared: the adapters must be thin, and the operations must not decide who may call them. """ +import asyncio import inspect from pathlib import Path @@ -221,3 +222,32 @@ async def test_reload_config_without_fn_is_refused(tmp_path): state = _state(tmp_path) with pytest.raises(ops.OpError, match="Reload not available"): await ops.reload_config(state) + + +async def test_start_reload_returns_before_reload_fn_finishes(tmp_path): + """The loopback route uses this one: a brand-new group's initial scan + can take minutes, and the Electron bridge caps every loopback call at + 30s (main.js node:call) — start_reload must not block on it.""" + state = _state(tmp_path) + release = asyncio.Event() + called = [] + + async def slow_reload(): + await release.wait() + called.append(True) + state["reload_fn"] = slow_reload + + out = await asyncio.wait_for(ops.start_reload(state), timeout=1.0) + + assert out["status"] == "reloading" + assert not called, "start_reload must return before reload_fn finishes" + + release.set() + await asyncio.sleep(0) # let the still-running reload_fn task complete + assert called, "reload_fn must still actually run, just not be waited on" + + +async def test_start_reload_without_fn_is_refused(tmp_path): + state = _state(tmp_path) + with pytest.raises(ops.OpError, match="Reload not available"): + await ops.start_reload(state) diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py new file mode 100644 index 0000000..719b988 --- /dev/null +++ b/packages/meshbay-node/tests/test_scan_settings_policy.py @@ -0,0 +1,205 @@ +""" +The operator can tune how often the indexer's reconciliation backstop runs, +and how long it waits after a file's last write before hashing it. + +Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py: +changed by a signed operator instruction, stored on the node rather than the +hub. Unlike those two, there is also a *live* DirectoryIndexer object to +update — see test_set_scan_settings_updates_the_live_indexer below. +""" + +import os +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.adminop import OP_SET_SCAN_SETTINGS +from meshbay_common.crypto import generate_gek +from meshbay_node import ops +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture +def gek(): + return generate_gek() + + +@pytest.fixture +def shared_dir(tmp_path): + d = tmp_path / "shared" + d.mkdir() + (d / "video.mkv").write_bytes(os.urandom(256)) + return d + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = None + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_out_of_range_reconcile_interval_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": 1.0, "debounce_secs": 2.0}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_out_of_range_debounce_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": 600.0, "debounce_secs": 99999.0}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_non_numeric_values_are_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": "not-a-number", "debounce_secs": 2.0}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False + + session._do_set_scan_settings( + {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Who may change it ─────────────────────────────────────────────────────── + +async def test_changing_it_needs_a_signature(tmp_path): + """The request only ever produces a challenge — nothing is applied + until a signature over the transcript verifies.""" + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_set_scan_settings( + {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0}) + + assert issued == [(OP_SET_SCAN_SETTINGS, "600,2")] + + +# ── Where it is stored ────────────────────────────────────────────────────── + +async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + defaults = await roster.scan_settings("g1") + assert defaults == { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }, "unset must mean the indexer's own defaults, or an upgrade " \ + "changes behaviour for every existing group" + + await roster.set_scan_settings("g1", 1200.0, 5.0, set_by="op") + assert await roster.scan_settings("g1") == { + "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0} + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.scan_settings("g1") == { + "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0} + assert await reopened.scan_settings("g2") == { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }, "one group's setting must not answer for another" + finally: + await reopened.close() + + +# ── Applying it to the live indexer ───────────────────────────────────────── + +async def test_set_scan_settings_updates_the_live_indexer(tmp_path, shared_dir, gek): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + indexer = DirectoryIndexer( + roots=one_root(shared_dir), group_id="g1", + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await indexer.initial_scan() + indexer._reconcile_delay = 5000.0 # simulate a long-idle backoff + state = {"roster": roster, "indexers": {"g1": indexer}} + + try: + result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0) + + assert result == {"reconcile_interval_secs": 1800.0, "debounce_secs": 3.0, + "group_id": "g1"} + assert indexer.reconcile_secs == 1800.0 + assert indexer.debounce_secs == 3.0 + assert indexer._reconcile_delay == 1800.0, \ + "the new interval must apply right away, not after whatever " \ + "backoff had already stretched the wait to" + assert await roster.scan_settings("g1") == { + "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0} + finally: + await roster.close() + + +async def test_set_scan_settings_without_a_live_indexer_still_persists(tmp_path): + """A group hosted on the node but with no running indexer in this + process (e.g. a test, or a group not yet hot-loaded) must not crash — + the setting still lands in roster.db for whenever it is.""" + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + state = {"roster": roster, "indexers": {}} + + try: + result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0) + assert result["reconcile_interval_secs"] == 1800.0 + assert await roster.scan_settings("g1") == { + "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0} + finally: + await roster.close() |