diff options
Diffstat (limited to 'packages')
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() |