diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-10-12 00:17:32 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-10-12 00:17:32 +0200 |
| commit | 14911aae21ec4975f7a805b91e3c4867cb41974b (patch) | |
| tree | 8c00230a88053905fd77938f3a9539efeb4c3094 /packages | |
| parent | f6f107e02c7a4aa7326c9a82310a239327d81ed6 (diff) | |
| download | meshbay-14911aae21ec4975f7a805b91e3c4867cb41974b.tar.gz | |
feat: let the operator pause and resume a root's indexing (MNP 6.2)
A whole-root walk now checks its root before every file: removed or
ejected, it stops instead of reading the drive to the end; paused, it
finishes from the hash cache alone. A paused root is not watched, not
swept by reconcile and survives a restart (roster), its indexed files
stay served. Resume runs the reconcile sweep over the root, so entries
and their enrichment are kept.
Signed root_pause / root_resume over MNP, loopback routes, CLI
`root pause|resume`, a pause button on the indexing dock and
pause/resume in the roots table of Group settings.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'packages')
42 files changed, 1793 insertions, 165 deletions
diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/keys/Transcripts.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/keys/Transcripts.kt index a48c4cf..b2d3d68 100644 --- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/keys/Transcripts.kt +++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/keys/Transcripts.kt @@ -171,7 +171,7 @@ class Transcripts(private val now: () -> Double = { System.currentTimeMillis() / "invite_cancel", "member_revoke", "apps_enabled", "set_scan_settings", "tmdb_config", "tmdb_enabled", "tmdb_override", "tmdb_rematch", "musicbrainz_enabled", "root_remove", "root_eject", "root_plug", - "app_directories", "chat_directory", "chat_link_preview", "search_listed", + "root_pause", "root_resume", "app_directories", "chat_directory", "chat_link_preview", "search_listed", "chat_epoch", "chat_purge", ) const val PREFIX_JOIN = "meshbay:join:v1" diff --git a/packages/meshbay-client/src/main.js b/packages/meshbay-client/src/main.js index 1af4386..33388e8 100644 --- a/packages/meshbay-client/src/main.js +++ b/packages/meshbay-client/src/main.js @@ -2396,6 +2396,8 @@ Its log: ${nodeLogHint()}`); updateRoot: (a) => ['PATCH', root(a), anObject(a.updates)], ejectRoot: (a) => ['PUT', `${root(a)}/eject`], plugRoot: (a) => ['PUT', `${root(a)}/plug`], + pauseRoot: (a) => ['PUT', `${root(a)}/pause`], + resumeRoot: (a) => ['PUT', `${root(a)}/resume`], removeRoot: (a) => ['DELETE', root(a)], initGek: (a) => ['POST', `${group(a)}/gek${a.rotate ? '?rotate=true' : ''}`], pairOperator: () => ['POST', '/api/operator/pair'], diff --git a/packages/meshbay-client/src/transcripts.js b/packages/meshbay-client/src/transcripts.js index ffba3bf..8119d71 100644 --- a/packages/meshbay-client/src/transcripts.js +++ b/packages/meshbay-client/src/transcripts.js @@ -42,7 +42,7 @@ const ADMIN_OPS = new Set([ 'invite_cancel', 'member_revoke', 'apps_enabled', 'set_scan_settings', 'tmdb_config', 'tmdb_enabled', 'tmdb_override', 'tmdb_rematch', 'musicbrainz_enabled', 'root_remove', 'root_eject', 'root_plug', - 'app_directories', 'chat_directory', 'chat_link_preview', 'search_listed', + 'root_pause', 'root_resume', 'app_directories', 'chat_directory', 'chat_link_preview', 'search_listed', 'chat_epoch', 'chat_purge', ]); diff --git a/packages/meshbay-common/src/meshbay_common/__init__.py b/packages/meshbay-common/src/meshbay_common/__init__.py index e8a9579..b86dec2 100644 --- a/packages/meshbay-common/src/meshbay_common/__init__.py +++ b/packages/meshbay-common/src/meshbay_common/__init__.py @@ -276,5 +276,11 @@ __version__ = "0.19.0" # 6.1 (2026-10-08) adds the signed `chat_purge` / `chat_purge_ack`: the operator # deletes a group's stored chat. Additive — a 6.0 node answers nothing, as for # any unknown type, and nothing else changes. The floor stays at 4.0. -MNP_VERSION = "6.1" +# +# 6.2 (2026-10-11) adds the signed `root_pause` / `root_resume` and their acks: +# the operator stops a root's indexing and starts it again, and each root in +# the roots table carries `paused`. Additive — a 6.1 node answers nothing to +# either, and a client that ignores `paused` shows a paused root as an ordinary +# one. The floor stays at 4.0. +MNP_VERSION = "6.2" MHP_VERSION = "0.1" diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index f59b420..43d2a7f 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -113,6 +113,11 @@ OP_CHAT_EPOCH = "chat_epoch" OP_CHAT_PURGE = "chat_purge" OP_ROOT_EJECT = "root_eject" OP_ROOT_PLUG = "root_plug" +# Stop and restart the indexing of one root (MNP 6.2). The subject is the root +# name, like eject and plug. Pausing narrows what the node does and resuming +# returns to what the operator already shared, so neither widens anything. +OP_ROOT_PAUSE = "root_pause" +OP_ROOT_RESUME = "root_resume" # Also gone with 6.0, because no client ever sent them: `gek_rotate`, # `member_unpin`, `transfer_limits` and `group_detach`. Rotating the group key, # forgetting an identity, the per-member transfer caps and no longer hosting a diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 5073b66..e5ceaa7 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -252,6 +252,12 @@ class MNP: ROOT_EJECT_ACK = "root_eject_ack" ROOT_PLUG = "root_plug" # operator → node: re-enable an ejected root ROOT_PLUG_ACK = "root_plug_ack" + # Operator → node: stop / restart indexing one root (MNP 6.2). The acks are + # broadcast, like eject's, so every open table shows the new state. + ROOT_PAUSE = "root_pause" + ROOT_PAUSE_ACK = "root_pause_ack" + ROOT_RESUME = "root_resume" + ROOT_RESUME_ACK = "root_resume_ack" # Member → node: who is in this group and which device keys they hold, with # the countersignature that admitted each one. Scoped to one group and # answers any member of it, because the point is that a member 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 48b1c80..3e3b483 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/group-page.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/group-page.js @@ -510,6 +510,9 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, setNodeRoots(msg.roots); } }; + // The dock's pause button, for a node that is not on this machine. + const pauseFromDock = (rootName) => + transport.pauseRoot(groupId, rootName, transport.signFn); // 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 @@ -526,7 +529,7 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, // The operator's indexing dock. The push names no root, so the // page that opened the roots table names it. if (transport.memberRole === 'operator') { - reportIndexPush(groupId, status, nodeRootsRef.current); + reportIndexPush(groupId, status, nodeRootsRef.current, pauseFromDock); } }; setOperatorPaired(transport.memberRole === 'operator'); @@ -637,7 +640,8 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, } } if (transport.memberRole === 'operator' && ack.indexing) { - reportIndexPush(groupId, ack.indexing, indexMsg.roots || nodeRootsRef.current); + reportIndexPush(groupId, ack.indexing, indexMsg.roots || nodeRootsRef.current, + pauseFromDock); } } catch (err) { if (cancelled) return; 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 1a367be..a666c29 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js @@ -55,8 +55,9 @@ async function readLocalNode(node, groupId, attempts = NODE_CHECK_ATTEMPTS, * mode="live" — a hosted group. What widens the node's sharing — adding * a directory, its `writable` and `removable` switches — goes * through the loopback API only, so only on the node's own - * machine. Removing, ejecting and plugging are signed ops - * over MNP, or the loopback API when there is no connection. + * machine. Removing, ejecting, plugging, pausing and + * resuming are signed ops over MNP, or the loopback API when + * there is no connection. * mode="local" — the wizard, before the group exists. Changes are held in * an array the caller owns; nothing is persisted until the * group is attached. @@ -72,7 +73,7 @@ async function readLocalNode(node, groupId, attempts = NODE_CHECK_ATTEMPTS, * * Props: * roots — the node's current roots: { name, path, writable, - * removable, ejected, available, kind } + * removable, ejected, paused, available, kind } * groupId — the group id * transport — MeshBayTransport instance, or null when not connected * signFn — signing function for admin ops @@ -153,7 +154,7 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, // signature proves that the operator's key signed, not that the operator // meant it: in a browser that key is driven by code the hub serves. // - // Narrowing — remove, eject, plug — MNP first, then loopback. + // Narrowing — remove, eject, plug, pause, resume — MNP first, then loopback. const overMnp = !isLocal && transport && transport.connected; const overLoopback = !isLocal && !overMnp && nodeAvail; const onNodeMachine = !isLocal && nodeAvail; @@ -220,6 +221,18 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, else throw new Error(t('node.root_no_route')); }), [overMnp, overLoopback, transport, groupId, signFn, run]); + const doPauseRoot = useCallback((rootName) => run(async () => { + if (overMnp) await transport.pauseRoot(groupId, rootName, signFn); + else if (overLoopback) await platform.node.op('pauseRoot', { groupId, rootName }); + else throw new Error(t('node.root_no_route')); + }), [overMnp, overLoopback, transport, groupId, signFn, run]); + + const doResumeRoot = useCallback((rootName) => run(async () => { + if (overMnp) await transport.resumeRoot(groupId, rootName, signFn); + else if (overLoopback) await platform.node.op('resumeRoot', { groupId, rootName }); + else throw new Error(t('node.root_no_route')); + }), [overMnp, overLoopback, transport, groupId, signFn, run]); + const doRemoveRoot = useCallback(async (rootName) => { if (isLocal) { if (onLocalRootsChange) { @@ -321,6 +334,8 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, <span class="node-root-badge node-root-badge-warn">${t('group.root_ejected')}</span>`} ${!isLocal && r.available === false && !r.ejected && html` <span class="node-root-badge node-root-badge-warn">${t('node.unavailable')}</span>`} + ${!isLocal && r.paused && html` + <span class="node-root-badge">${t('node.root_paused')}</span>`} </td> ${hasPaths && html` <td class="sdt-col-path" title=${r.path || ''}>${r.path || ''}</td>`} @@ -347,6 +362,14 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, `} <td class="sdt-col-actions"> ${canEdit && !isLocal && html` + <button class="sdt-action-btn" disabled=${busy || !!r.ejected} + title=${r.paused ? t('node.root_resume') : t('node.root_pause')} + aria-label=${r.paused ? t('node.root_resume') : t('node.root_pause')} + onClick=${() => r.paused ? doResumeRoot(r.name) : doPauseRoot(r.name)}> + <${Icon} name=${r.paused ? 'play' : 'pause'} /> + </button> + `} + ${canEdit && !isLocal && html` <button class="sdt-action-btn" disabled=${busy || !r.removable} title=${r.ejected ? t('group.root_plug') : t('group.root_eject')} onClick=${() => r.ejected ? doPlugRoot(r.name) : doEjectRoot(r.name)}> diff --git a/packages/meshbay-hub/src/meshbay_hub/static/index-dock-model.js b/packages/meshbay-hub/src/meshbay_hub/static/index-dock-model.js index bd4586b..ed81e7c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/index-dock-model.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/index-dock-model.js @@ -91,8 +91,9 @@ function loud(job, since, now) { * * `memory` is per group: when the current walk was first seen, whether its row * has been shown (a burst that crossed the threshold stays up until it ends), - * whether the operator hid it, and when it finished. Pure: same arguments, same - * answer, so a render that runs twice does not flash anything twice. + * whether the operator hid it, which walk they paused, and when it finished. + * Pure: same arguments, same answer, so a render that runs twice does not flash + * anything twice. */ export function dockRows(jobs, memory, now) { const next = {}; @@ -105,9 +106,12 @@ export function dockRows(jobs, memory, now) { const since = prev.busy && prev.sig === sig ? prev.since : now; const shown = Boolean(prev.busy && prev.shown) || loud(job, since, now); const hidden = Boolean(prev.busy && prev.hidden); - next[gid] = { busy: true, sig, since, shown, hidden, doneAt: 0 }; - if (shown && !hidden) rows.push({ ...job, state: 'running' }); - } else if (prev.busy && prev.shown && !prev.hidden) { + // The node takes a moment to stop, and may go on to the next root + // queued: only the walk that was paused stays off screen. + const paused = prev.busy && prev.pausedSig === sig ? sig : ''; + next[gid] = { busy: true, sig, since, shown, hidden, pausedSig: paused, doneAt: 0 }; + if (shown && !hidden && !paused) rows.push({ ...job, state: 'running' }); + } else if (prev.busy && prev.shown && !prev.hidden && !prev.pausedSig) { next[gid] = { busy: false, doneAt: now }; rows.push({ ...job, state: 'done' }); } else if (prev.doneAt && now - prev.doneAt < DONE_MS) { @@ -124,3 +128,14 @@ export function hideRow(memory, groupId) { if (!prev || !prev.busy) return memory; return { ...memory, [groupId]: { ...prev, hidden: true } }; } + +/** + * The operator paused the walk this row shows. Its row goes at once rather than + * when the node's next report arrives, and no "finished" follows it — the root + * is not finished, it is paused, and its settings row says so. + */ +export function pauseRow(memory, groupId) { + const prev = memory[groupId]; + if (!prev || !prev.busy) return memory; + return { ...memory, [groupId]: { ...prev, pausedSig: prev.sig } }; +} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/index-dock.js b/packages/meshbay-hub/src/meshbay_hub/static/index-dock.js index 6c89eb6..47a5818 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/index-dock.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/index-dock.js @@ -6,7 +6,7 @@ import { navigate, session } from './hub-client.js'; import { useStickyBand } from './sticky.js'; import { Icon } from './icon.js'; import { - fromLoopback, fromPush, mergeActivity, dockRows, hideRow, percent, queuedCount, + fromLoopback, fromPush, mergeActivity, dockRows, hideRow, pauseRow, percent, queuedCount, } from './index-dock-model.js'; // ── Indexing dock ─────────────────────────────────────────────────────────── @@ -29,6 +29,9 @@ const ABSENT_POLL_MS = 30000; // rather than a prop through App: the page and the dock are far apart, and a // push every two seconds must not re-render the whole application. const pushed = new Map(); +// How to pause a root of that group over the page's connection, signed by its +// operator — the dock holds no transport of its own. +const pushedPause = new Map(); const listeners = new Set(); /** @@ -36,11 +39,17 @@ const listeners = new Set(); * member keeps the sidebar dot. `status` is the push (or the handshake ack's * `indexing`); `roots` the roots table the page opened, which names `root_pos`. * `null` when the page lets go of the group: nothing will update the entry any - * more, and a stale "scanning" would stay on screen for ever. + * more, and a stale "scanning" would stay on screen for ever. `pause(rootName)` + * is what the row's pause button calls when no node answers on this machine. */ -export function reportIndexPush(groupId, status, roots) { - if (status) pushed.set(groupId, fromPush(groupId, status, roots)); - else pushed.delete(groupId); +export function reportIndexPush(groupId, status, roots, pause) { + if (status) { + pushed.set(groupId, fromPush(groupId, status, roots)); + if (pause) pushedPause.set(groupId, pause); + } else { + pushed.delete(groupId); + pushedPause.delete(groupId); + } for (const fn of listeners) fn(); } @@ -96,7 +105,7 @@ function usePushedActivity() { return snapshot; } -function DockRow({ row, onOpen, onHide }) { +function DockRow({ row, error, onOpen, onPause, onHide }) { const done = row.state === 'done'; const waiting = !done && !row.scanning; const listing = !done && row.scanning && !row.totalBytes; @@ -109,7 +118,9 @@ function DockRow({ row, onOpen, onHide }) { const where = row.root ? `${row.groupName} › ${row.root}` : row.groupName; let meta = ''; - if (listing) { + if (error) { + meta = error; + } else if (listing) { meta = t('indexdock.listing'); } else if (!done && !waiting) { meta = [ @@ -133,7 +144,8 @@ function DockRow({ row, onOpen, onHide }) { onClick=${onOpen}> <span class="index-dock-head"> <span class="index-dock-title"><strong>${status}</strong> ${where}</span> - ${meta && html`<span class="index-dock-meta">${meta}</span>`} + ${meta && html`<span class="index-dock-meta ${error ? 'index-dock-error' : ''}" + role=${error ? 'alert' : null}>${meta}</span>`} </span> <span class="index-progress-bar index-dock-bar"> <span class="index-progress-fill ${moving ? 'index-dock-indeterminate' : ''}" @@ -141,6 +153,12 @@ function DockRow({ row, onOpen, onHide }) { </span> ${next && html`<span class="index-dock-next">${next}</span>`} </button> + ${!done && onPause && html` + <button type="button" class="index-dock-hide" title=${t('indexdock.pause')} + aria-label=${t('indexdock.pause')} onClick=${onPause}> + <${Icon} name="pause" /> + </button> + `} ${!done && html` <button type="button" class="index-dock-hide" title=${t('indexdock.hide')} aria-label=${t('indexdock.hide')} onClick=${onHide}> @@ -156,6 +174,7 @@ export function IndexingDock({ groups }) { const remote = usePushedActivity(); const memory = useRef({}); const [, setTick] = useState(0); + const [errors, setErrors] = useState({}); // Publishes `--index-dock-h` for the sidebar, as the music bar publishes its // own; withdrawn when the dock renders nothing. const band = useStickyBand('--index-dock-h'); @@ -173,11 +192,36 @@ export function IndexingDock({ groups }) { return () => clearInterval(id); }, [pending]); + // A whole-root walk names its root; a burst of file events does not, and + // there is nothing to pause. The local node first, as for the figures. + const pauser = (row) => { + if (row.state !== 'running' || !row.scanning || !row.root) return null; + if (local[row.groupId]) { + return (rootName) => platform.node.op('pauseRoot', { groupId: row.groupId, rootName }); + } + return pushedPause.get(row.groupId) || null; + }; + + const pause = async (row, via) => { + const before = memory.current; + memory.current = pauseRow(memory.current, row.groupId); + setErrors((e) => ({ ...e, [row.groupId]: '' })); + setTick((n) => n + 1); + try { + await via(row.root); + } catch (err) { + // Still running: put the row back, and say why. + memory.current = { ...memory.current, [row.groupId]: before[row.groupId] }; + setErrors((e) => ({ ...e, [row.groupId]: platform.bridgeMessage(err) })); + } + }; + if (!rows.length) return null; return html` <div class="index-dock" ref=${band}> ${rows.map((row) => html` - <${DockRow} key=${row.groupId} row=${row} + <${DockRow} key=${row.groupId} row=${row} error=${errors[row.groupId] || ''} + onPause=${pauser(row) && (() => pause(row, pauser(row)))} onOpen=${() => { session.openGroupTab = { groupId: row.groupId, tab: 'settings' }; navigate('/group/' + row.groupId); 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 c190efa..264cd5b 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -880,6 +880,7 @@ export default { 'indexdock.next_count': "Wartend: {count}", 'indexdock.open': "Einstellungen dieser Gruppe öffnen", 'indexdock.hide': "Bis zur nächsten Indizierung ausblenden", + 'indexdock.pause': "Indexierung dieses Ordners pausieren", 'chat.load_older': '{n} ältere Nachrichten laden', 'chat.start_of_history': 'Anfang des Gesprächs', 'chat.today': 'Heute', @@ -1001,6 +1002,9 @@ export default { 'node.remove_root': 'Entfernen', 'node.root_no_signing_key': 'Kein Signaturschlüssel verfügbar — koppeln Sie dieses Gerät zuerst mit dem Node', 'node.root_no_route': 'Keine Verbindung zum Node — verbinden Sie sich damit oder verwenden Sie die App auf dem Rechner, der ihn hostet', + 'node.root_pause': "Indexierung pausieren", + 'node.root_resume': "Indexierung fortsetzen", + 'node.root_paused': "Indexierung pausiert", 'node.root_remove_last': 'Eine Gruppe braucht mindestens ein Verzeichnis', 'node.root_path': 'Pfad', 'node.directory': 'Verzeichnis', 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 df85791..a2d7699 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -993,6 +993,7 @@ export default { 'indexdock.next_count': "{count} more waiting", 'indexdock.open': "Open this group's settings", 'indexdock.hide': "Hide until the next indexing", + 'indexdock.pause': "Pause indexing this folder", 'chat.load_older': 'Load {n} older messages', 'chat.start_of_history': 'Beginning of the conversation', 'chat.today': 'Today', @@ -1069,6 +1070,9 @@ export default { 'node.remove_root': 'Remove', 'node.root_no_signing_key': 'No signing key available — pair this device with the node first', 'node.root_no_route': 'No route to the node — connect to it, or use the app on the machine hosting it', + 'node.root_pause': "Pause indexing", + 'node.root_resume': "Resume indexing", + 'node.root_paused': "indexing paused", 'node.root_remove_last': 'A group needs at least one directory', 'node.root_path': 'Path', 'node.root_added': 'Directory added.', 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 50c08cc..e7d313e 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -874,6 +874,7 @@ export default { 'indexdock.next_count': "En espera: {count}", 'indexdock.open': "Abrir los ajustes de este grupo", 'indexdock.hide': "Ocultar hasta la próxima indexación", + 'indexdock.pause': "Pausar la indexación de esta carpeta", 'chat.load_older': 'Cargar {n} mensajes anteriores', 'chat.start_of_history': 'Inicio de la conversación', 'chat.today': 'Hoy', @@ -995,6 +996,9 @@ export default { 'node.remove_root': 'Eliminar', 'node.root_no_signing_key': 'No hay clave de firma disponible: empareja primero este dispositivo con el nodo', 'node.root_no_route': 'No hay ruta al nodo: conéctate a él o usa la aplicación en la máquina que lo aloja', + 'node.root_pause': "Pausar la indexación", + 'node.root_resume': "Reanudar la indexación", + 'node.root_paused': "indexación en pausa", 'node.root_remove_last': 'Un grupo necesita al menos un directorio', 'node.root_path': 'Ruta', 'node.directory': 'Directorio', 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 a56c9ef..e1994dd 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -877,6 +877,7 @@ export default { 'indexdock.next_count': "En attente : {count}", 'indexdock.open': "Ouvrir les paramètres de ce groupe", 'indexdock.hide': "Masquer jusqu'à la prochaine indexation", + 'indexdock.pause': "Mettre en pause l'indexation de ce dossier", 'chat.load_older': 'Charger {n} messages plus anciens', 'chat.start_of_history': 'Début de la conversation', 'chat.today': 'Aujourd’hui', @@ -1004,6 +1005,9 @@ export default { 'node.remove_root': 'Retirer', 'node.root_no_signing_key': 'Aucune clé de signature disponible — appairez d\'abord cet appareil avec le nœud', 'node.root_no_route': 'Aucune route vers le nœud — connectez-vous à lui, ou utilisez l\'application sur la machine qui l\'héberge', + 'node.root_pause': "Mettre l'indexation en pause", + 'node.root_resume': "Reprendre l'indexation", + 'node.root_paused': "indexation en pause", 'node.root_remove_last': 'Un groupe a besoin d\'au moins un répertoire', 'node.root_path': 'Chemin', 'node.root_added': 'Répertoire ajouté.', 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 de071ee..eb96a49 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -876,6 +876,7 @@ export default { 'indexdock.next_count': "In attesa: {count}", 'indexdock.open': "Apri le impostazioni di questo gruppo", 'indexdock.hide': "Nascondi fino alla prossima indicizzazione", + 'indexdock.pause': "Sospendi l'indicizzazione di questa cartella", 'chat.load_older': 'Carica {n} messaggi precedenti', 'chat.start_of_history': 'Inizio della conversazione', 'chat.today': 'Oggi', @@ -1003,6 +1004,9 @@ export default { 'node.remove_root': 'Rimuovi', 'node.root_no_signing_key': 'Nessuna chiave di firma disponibile: associa prima questo dispositivo al nodo', 'node.root_no_route': 'Nessuna via verso il nodo: connettiti a esso oppure usa l\'applicazione sulla macchina che lo ospita', + 'node.root_pause': "Sospendi l'indicizzazione", + 'node.root_resume': "Riprendi l'indicizzazione", + 'node.root_paused': "indicizzazione sospesa", 'node.root_remove_last': 'Un gruppo ha bisogno di almeno una directory', 'node.root_path': 'Percorso', 'node.directory': 'Directory', 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 d7261c6..2a4584a 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -864,6 +864,7 @@ export default { 'indexdock.next_count': "待機中: {count}", 'indexdock.open': "このグループの設定を開く", 'indexdock.hide': "次のインデックスまで隠す", + 'indexdock.pause': "このフォルダーのインデックス作成を一時停止", 'chat.load_older': '以前のメッセージを {n} 件読み込む', 'chat.start_of_history': '会話のはじまり', 'chat.today': '今日', @@ -989,6 +990,9 @@ export default { 'node.remove_root': '削除', 'node.root_no_signing_key': '署名鍵がありません — 先にこの端末をノードとペアリングしてください', 'node.root_no_route': 'ノードへの経路がありません — 接続するか、ノードを動かしているマシンでアプリを使ってください', + 'node.root_pause': "インデックス作成を一時停止", + 'node.root_resume': "インデックス作成を再開", + 'node.root_paused': "インデックス作成を一時停止中", 'node.root_remove_last': 'グループには少なくとも 1 つのディレクトリが必要です', 'node.root_path': 'パス', 'node.directory': 'ディレクトリ', 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 a628fb9..ce4acbe 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -878,6 +878,7 @@ export default { 'indexdock.next_count': "Wachtend: {count}", 'indexdock.open': "Instellingen van deze groep openen", 'indexdock.hide': "Verbergen tot de volgende indexering", + 'indexdock.pause': "Indexeren van deze map pauzeren", 'chat.load_older': '{n} oudere berichten laden', 'chat.start_of_history': 'Begin van het gesprek', 'chat.today': 'Vandaag', @@ -1005,6 +1006,9 @@ export default { 'node.remove_root': 'Verwijderen', 'node.root_no_signing_key': 'Geen ondertekeningssleutel beschikbaar — koppel dit apparaat eerst aan de node', 'node.root_no_route': 'Geen route naar de node — maak verbinding, of gebruik de app op de machine die hem host', + 'node.root_pause': "Indexeren pauzeren", + 'node.root_resume': "Indexeren hervatten", + 'node.root_paused': "indexeren gepauzeerd", 'node.root_remove_last': 'Een groep heeft minstens één map nodig', 'node.root_path': 'Pad', 'node.directory': 'Map', 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 9014664..72ace3a 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -898,6 +898,7 @@ export default { 'indexdock.next_count': "W kolejce: {count}", 'indexdock.open': "Otwórz ustawienia tej grupy", 'indexdock.hide': "Ukryj do następnego indeksowania", + 'indexdock.pause': "Wstrzymaj indeksowanie tego folderu", 'chat.load_older': 'Wczytaj {n} starszych wiadomości', 'chat.start_of_history': 'Początek rozmowy', 'chat.today': 'Dziś', @@ -1029,6 +1030,9 @@ export default { 'node.remove_root': 'Usuń', 'node.root_no_signing_key': 'Brak klucza podpisu — najpierw sparuj to urządzenie z węzłem', 'node.root_no_route': 'Brak połączenia z węzłem — połącz się z nim albo użyj aplikacji na komputerze, który go hostuje', + 'node.root_pause': "Wstrzymaj indeksowanie", + 'node.root_resume': "Wznów indeksowanie", + 'node.root_paused': "indeksowanie wstrzymane", 'node.root_remove_last': 'Grupa wymaga co najmniej jednego katalogu', 'node.root_path': 'Ścieżka', 'node.directory': 'Katalog', 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 e6972fb..ad46dc2 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 @@ -875,6 +875,7 @@ export default { 'indexdock.next_count': "Na fila: {count}", 'indexdock.open': "Abrir as configurações deste grupo", 'indexdock.hide': "Ocultar até a próxima indexação", + 'indexdock.pause': "Pausar a indexação desta pasta", 'chat.load_older': 'Carregar {n} mensagens anteriores', 'chat.start_of_history': 'Início da conversa', 'chat.today': 'Hoje', @@ -996,6 +997,9 @@ export default { 'node.remove_root': 'Remover', 'node.root_no_signing_key': 'Nenhuma chave de assinatura disponível — pareie este dispositivo com o nó primeiro', 'node.root_no_route': 'Sem rota até o nó — conecte-se a ele ou use o aplicativo na máquina que o hospeda', + 'node.root_pause': "Pausar a indexação", + 'node.root_resume': "Retomar a indexação", + 'node.root_paused': "indexação pausada", 'node.root_remove_last': 'Um grupo precisa de pelo menos um diretório', 'node.root_path': 'Caminho', 'node.directory': 'Diretório', 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 96f7f5f..ccaf059 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 @@ -852,6 +852,7 @@ export default { 'indexdock.next_count': "等待中:{count}", 'indexdock.open': "打开此群组的设置", 'indexdock.hide': "隐藏,直到下一次索引", + 'indexdock.pause': "暂停索引此文件夹", 'chat.load_older': '加载更早的 {n} 条消息', 'chat.start_of_history': '对话开头', 'chat.today': '今天', @@ -977,6 +978,9 @@ export default { 'node.remove_root': '移除', 'node.root_no_signing_key': '没有可用的签名密钥 — 请先将本设备与节点配对', 'node.root_no_route': '无法连接到节点 — 请先连接,或在运行该节点的机器上使用应用', + 'node.root_pause': "暂停索引", + 'node.root_resume': "继续索引", + 'node.root_paused': "索引已暂停", 'node.root_remove_last': '每个群组至少需要一个目录', 'node.root_path': '路径', 'node.directory': '目录', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css index 67d52b9..f6cb57c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -3591,6 +3591,7 @@ h2 .gn-owner, h3 .gn-owner { font-size: 0.55em; } .sdt-action-btn:hover:not(:disabled) { background: var(--bg-raised); color: var(--text); } .sdt-action-btn:disabled { opacity: 0.3; cursor: default; } .sdt-action-danger:hover:not(:disabled) { color: var(--error); border-color: var(--error); } +.sdt-action-btn .icon { width: 1em; height: 1em; display: block; } .sdt-dir-name { display: inline-flex; align-items: center; gap: 6px; font-weight: 500; } .sdt-dir-name .icon { width: 16px; height: 16px; flex-shrink: 0; } .sdt-row-ejected { opacity: 0.5; } @@ -4832,6 +4833,7 @@ h2 .gn-owner, h3 .gn-owner { font-size: 0.55em; } white-space: nowrap; font-variant-numeric: tabular-nums; } +.index-dock-meta.index-dock-error { color: var(--error); white-space: normal; } .index-dock-bar { display: block; margin: 6px 0 0; } .index-dock-bar .index-progress-fill { display: block; } .index-dock-done .index-progress-fill { background: var(--success); } diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport-admin.js b/packages/meshbay-hub/src/meshbay_hub/static/transport-admin.js index 7bc77af..aea156b 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport-admin.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport-admin.js @@ -279,6 +279,27 @@ extendTransport(class { return msg; } + // Stop / restart indexing one root (MNP 6.2). Already-indexed files stay + // listed while it is paused. + async pauseRoot(groupId, rootName, signFn) { + return this._rootIndexOp('root_pause', groupId, rootName, signFn); + } + + async resumeRoot(groupId, rootName, signFn) { + return this._rootIndexOp('root_resume', groupId, rootName, signFn); + } + + async _rootIndexOp(type, groupId, rootName, signFn) { + const msg = await this._sendAndWait({ + type, v: '6.2', group_id: groupId, root_name: rootName, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, type, rootName, signFn); + } + return msg; + } + /** * Ask the node for a one-time pairing code admitting `userId` to this group. * diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index ee6645f..c86d668 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -196,7 +196,7 @@ function _aborted() { // unsaved-looking draft after a save that had worked. const BROADCAST_ACK_TYPES = new Set([ 'root_eject_ack', 'root_plug_ack', 'root_remove_ack', - 'app_directories_ack', 'chat_directory_ack', 'chat_link_preview_ack', + 'root_pause_ack', 'root_resume_ack', 'app_directories_ack', 'chat_directory_ack', 'chat_link_preview_ack', 'chat_epoch_ack', 'chat_purge_ack', 'search_listed_ack', ]); @@ -223,7 +223,7 @@ const ADMIN_OP_TYPES = new Set([ 'tmdb_override', 'tmdb_rematch', 'tmdb_config', 'tmdb_enabled', 'musicbrainz_enabled', 'file_delete', 'dir_delete', 'apps_enabled', 'set_scan_settings', 'member_revoke', - 'root_remove', 'root_eject', 'root_plug', + 'root_remove', 'root_eject', 'root_plug', 'root_pause', 'root_resume', 'app_directories', 'chat_directory', 'chat_link_preview', 'chat_epoch', 'chat_purge', 'search_listed', 'invite_create', 'invite_link_create', 'invite_cancel', ]); @@ -2141,11 +2141,13 @@ class MeshBayTransport { this._onMusicbrainzEnabled(Boolean(msg.enabled)); } - // A root was ejected, plugged or removed. Broadcast by the node to every - // peer, so everyone's table updates without waiting for the next index_sync. + // A root was ejected, plugged, paused, resumed or removed. Broadcast by the + // node to every peer, so everyone's table updates without waiting for the + // next index_sync. // A root added or a flag changed — on the node's own machine, never over // MNP — arrives in the index_delta the node pushes. if (msg.type === 'root_eject_ack' || msg.type === 'root_plug_ack' + || msg.type === 'root_pause_ack' || msg.type === 'root_resume_ack' || msg.type === 'root_remove_ack') { if (this._onRootsChanged) this._onRootsChanged(msg); } diff --git a/packages/meshbay-hub/tests/test_indexing_dock.py b/packages/meshbay-hub/tests/test_indexing_dock.py index e960950..1619b57 100644 --- a/packages/meshbay-hub/tests/test_indexing_dock.py +++ b/packages/meshbay-hub/tests/test_indexing_dock.py @@ -51,15 +51,18 @@ def _job(**over) -> dict: return job -def _frames(tmp_path, frames: list[tuple[int, list[dict]]], hide_at: int | None = None): +def _frames(tmp_path, frames: list[tuple[int, list[dict]]], hide_at: int | None = None, + pause_at: int | None = None): """Feed dockRows a sequence of (time ms, jobs) and return each frame's rows.""" return _run(tmp_path, f""" const frames = {json.dumps(frames)}; const hideAt = {json.dumps(hide_at)}; + const pauseAt = {json.dumps(pause_at)}; let memory = {{}}; const out = []; for (const [now, jobs] of frames) {{ if (hideAt === now) memory = m.hideRow(memory, 'g1'); + if (pauseAt === now) memory = m.pauseRow(memory, 'g1'); const r = m.dockRows(jobs, memory, now); memory = r.memory; out.push(r.rows.map((row) => [row.groupId, row.state])); @@ -120,6 +123,24 @@ def test_a_hidden_row_stays_hidden_until_the_group_is_idle(tmp_path): # ── Where it comes from ───────────────────────────────────────────────────── @needs_node +def test_a_paused_walk_leaves_at_once_and_is_not_called_finished(tmp_path): + """The node takes a moment to stop: its next reports still say scanning.""" + idle = _job(scanning=False, kind="") + assert _frames(tmp_path, [ + (0, [_job()]), (1000, [_job()]), (2000, [_job()]), (3000, [idle]), + ], pause_at=1000) == [[["g1", "running"]], [], [], []] + + +@needs_node +def test_pausing_one_root_does_not_hide_the_next_one_queued(tmp_path): + first = _job(root="results", queued=["archive"]) + second = _job(root="archive") + assert _frames(tmp_path, [ + (0, [first]), (1000, [first]), (2000, [second]), + ], pause_at=1000) == [[["g1", "running"]], [], [["g1", "running"]]] + + +@needs_node def test_the_loopback_answer_wins_and_the_hub_names_the_group(tmp_path): out = _run(tmp_path, """ const local = { g1: m.fromLoopback({ group_id: 'g1', group_name: 'node name', @@ -173,6 +194,17 @@ def test_only_an_operator_connection_feeds_the_dock(): "a member's connection must keep the sidebar dot only") +def test_the_pause_button_reaches_the_node_by_either_door(): + """The local node over loopback; any other through the page's connection, + signed by its operator — the dock holds no transport of its own.""" + dock = DOCK.read_text(encoding="utf-8") + assert "node.op('pauseRoot'" in dock + page = GROUP_PAGE.read_text(encoding="utf-8") + assert "transport.pauseRoot(groupId, rootName, transport.signFn)" in page + for at in [m.start() for m in re.finditer(r"reportIndexPush\(groupId, (?!null)", page)]: + assert "pauseFromDock" in page[at:at + 140] + + def test_leaving_a_group_clears_its_row(): page = GROUP_PAGE.read_text(encoding="utf-8") cleanup = page[page.index(" return () => {\n cancelled = true;"):] diff --git a/packages/meshbay-node/src/meshbay_node/cli/groups.py b/packages/meshbay-node/src/meshbay_node/cli/groups.py index fca4800..b402930 100644 --- a/packages/meshbay-node/src/meshbay_node/cli/groups.py +++ b/packages/meshbay-node/src/meshbay_node/cli/groups.py @@ -33,6 +33,8 @@ def group(args) -> None: flags.append("removable") if r.get("ejected"): flags.append("ejected") + if r.get("paused"): + flags.append("paused") flag_str = f" ({', '.join(flags)})" if flags else "" live = "" if r.get("available", True) else " [UNAVAILABLE]" print(f" root {r['name']}{flag_str}{live}") @@ -128,6 +130,8 @@ def root(args) -> None: flags.append("removable") if r.get("ejected"): flags.append("EJECTED") + if r.get("paused"): + flags.append("PAUSED") avail = "available" if r.get("available", True) else "UNAVAILABLE" flags.append(avail) print(f" {r['name']:<20} {', '.join(flags)}") @@ -215,8 +219,29 @@ def root(args) -> None: print(f"plugged root {name} — files are visible again") return - print("usage: meshbay-node root list|add|remove|set|eject|plug [name] " - "[--group NAME]") + if sub == "pause": + name = args.target + if not name: + print("usage: meshbay-node root pause <name> [--group NAME]") + sys.exit(1) + _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/pause", + method="PUT") + print(f"paused root {name} — indexed files stay listed, nothing new " + f"is indexed until resumed") + return + + if sub == "resume": + name = args.target + if not name: + print("usage: meshbay-node root resume <name> [--group NAME]") + sys.exit(1) + _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/resume", + method="PUT") + print(f"resumed root {name} — indexing continues in the background") + return + + print("usage: meshbay-node root list|add|remove|set|eject|plug|pause|resume " + "[name] [--group NAME]") sys.exit(1) diff --git a/packages/meshbay-node/src/meshbay_node/cli/parser.py b/packages/meshbay-node/src/meshbay_node/cli/parser.py index cfc4704..2a593b6 100644 --- a/packages/meshbay-node/src/meshbay_node/cli/parser.py +++ b/packages/meshbay-node/src/meshbay_node/cli/parser.py @@ -20,7 +20,7 @@ def build_parser() -> argparse.ArgumentParser: "| operator pair: pair a " "browser with this node | member list|invite|cancel|revoke|unpin " "| group list|add|remove " - "| root list|add|remove|set|eject|plug " + "| root list|add|remove|set|eject|plug|pause|resume " "| gek init|rotate | file list|rm " "| video rematch: re-resolve TMDB matches for a group's " "videos " @@ -44,7 +44,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("subcommand", nargs="?", help="'pair' for operator; list|invite|revoke|unpin for " "member; list|add|remove for group; " - "list|add|remove|set|eject|plug for root; " + "list|add|remove|set|eject|plug|pause|resume for root; " "init|rotate for gek; " "list|rm for file; rematch for video; show|clear for " "denylist; list|add|remove|reset for stun; " diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 4d89f97..022553e 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -1286,7 +1286,8 @@ class NodeDaemon(EnrichmentMixin): async def _build_roots(self, group_cfg) -> RootSet: """ - Build a group's RootSet from node.toml, with the ejected state restored. + Build a group's RootSet from node.toml, with the ejected and paused + state restored. node.toml carries configuration (`writable`, `removable`); the roster carries the runtime answer to "is this drive ejected right now". They @@ -1298,12 +1299,14 @@ class NodeDaemon(EnrichmentMixin): if self._roster: ejected = await self._roster.ejected_roots(group_cfg.id) auto = await self._roster.auto_ejected_roots(group_cfg.id) - if ejected: - for spec in specs: - name = spec.get("name") or Path(spec.get("path", "")).name - if fold(name) in ejected: - spec["ejected"] = True - spec["ejected_auto"] = fold(name) in auto + paused = await self._roster.paused_roots(group_cfg.id) + for spec in specs: + name = spec.get("name") or Path(spec.get("path", "")).name + if fold(name) in ejected: + spec["ejected"] = True + spec["ejected_auto"] = fold(name) in auto + if fold(name) in paused: + spec["paused"] = True return RootSet.build(specs) # Every application that keeps directories. This is the one list, and it diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 4eeeec0..4d68212 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -396,7 +396,7 @@ class DirectoryIndexer: await off_disk(self.roots, self.roots.refresh_availability) await self._plug_back_recognised() total = 0 - waiting = [r.name for r in self.roots if r.available] + waiting = [r.name for r in self.roots if r.available and not r.paused] self._queue(waiting) try: for root in self.roots: @@ -405,8 +405,9 @@ class DirectoryIndexer: "are not indexed yet and will appear when it returns", root.name, root.path) continue - waiting.remove(root.name) - self._unqueue([root.name]) + if root.name in waiting: + waiting.remove(root.name) + self._unqueue([root.name]) total += await self._scan_root(root) finally: self._unqueue(waiting) @@ -417,9 +418,21 @@ class DirectoryIndexer: total, len(self.roots)) async def _scan_root(self, root: Root, *, kind: str = "scan") -> int: + """ + Walk one root and index what is on it. + + A paused root — paused before the walk or part-way through it — is + listed from the hash cache alone: what the node already hashed comes + back, nothing new is read. It cannot simply be skipped, because the + index is rebuilt at every start and a rescan drops the root's entries + before it walks. A root removed or ejected while this runs is not + finished at all: its entries are dropped, or frozen on a drive that is + leaving, so reading the rest of it is work for nothing. + """ log.info("Scanning %s (root %r) ...", root.path, root.name) count = 0 loop = asyncio.get_event_loop() + quiet = root.paused # `scanning` flips on here, before the walk — not after it and the # sizing pass, which for a large or slow (network/USB) root can @@ -432,7 +445,8 @@ class DirectoryIndexer: # had not yet turned on, even though the node was already several # seconds into walking and sizing a large root — total_bytes is # unknown at this point, so it starts at 0 and is corrected below. - self._begin_job(root.name, root=root, kind=kind) + if not quiet: + self._begin_job(root.name, root=root, kind=kind) try: try: files = await loop.run_in_executor(self._executor, _walk_root, root) @@ -447,14 +461,30 @@ class DirectoryIndexer: # loop right here, blocking the whole daemon for a large root). sized = await loop.run_in_executor(self._executor, _size_files, files) - self.progress.total_bytes = sum(size for _, size in sized) - self.progress.files_total = len(sized) - self.progress.current_dir = "" - 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 - self.progress.files_done += 1 + if not quiet: + self.progress.total_bytes = sum(size for _, size in sized) + self.progress.files_total = len(sized) + self.progress.current_dir = "" + for done, (file_path, size) in enumerate(sized): + live = self._live(root) + if live is None or live.ejected: + log.info("Scan of root %r stopped after %d of %d files: " + "removed or ejected", root.name, done, len(sized)) + break + if not quiet and live.paused: + log.info("Root %r paused after %d of %d files — the rest is " + "listed from the hash cache only", root.name, done, + len(sized)) + # Off the operator's screen now; what is left reads no file. + self._end_job() + quiet = True + if not quiet: + self.progress.current_dir = file_path.parent.name + entry = await (self._cached(root, file_path) if quiet + else self._hash_or_cached(root, file_path)) + if not quiet: + self.progress.scanned_bytes += size + self.progress.files_done += 1 if entry: self._index.add_entry(entry) count += 1 @@ -462,7 +492,8 @@ class DirectoryIndexer: # 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._end_job() + if not quiet: + self._end_job() return count def _begin_job(self, current_dir: str, total_bytes: int = 0, *, @@ -520,7 +551,13 @@ class DirectoryIndexer: if name in self.progress.queued: self.progress.queued.remove(name) - async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None: + async def _cached(self, root: Root, file_path: Path) -> IndexEntry | None: + """What `_hash_or_cached` would return without reading the file, or + None: a paused root's file the cache does not know is left out.""" + return await self._hash_or_cached(root, file_path, read=False) + + async def _hash_or_cached(self, root: Root, file_path: Path, *, + read: bool = True) -> 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 @@ -553,6 +590,8 @@ class DirectoryIndexer: hash_version=cached.hash_version, ), file_path, st) + if not read: + return None 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: @@ -643,7 +682,7 @@ class DirectoryIndexer: self._observer = Observer() watched = 0 for root in self.roots: - if not root.available: + if not root.available or root.paused: continue try: self._observer.schedule(handler, str(root.path), recursive=True) @@ -742,8 +781,26 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + def _live(self, root: Root) -> Root | None: + """ + This set's own object for `root`, or None once it is gone from the set. + + A retarget replaces every Root object, so a walk that started before + one holds a stale copy: the flags an operator sets afterwards are on + this one. + """ + for r in self.roots: + if r.folded == root.folded and r.path == root.path: + return r + return None + def _holds(self, root: Root) -> bool: - return any(r.folded == root.folded and r.path == root.path for r in self.roots) + return self._live(root) is not None + + def _stopped(self, root: Root) -> bool: + """Removed, ejected or paused since this walk began: stop reading it.""" + live = self._live(root) + return live is None or live.ejected or live.paused async def _scan_added_roots(self, added: list[Root]) -> None: waiting = [r.name for r in added] @@ -867,69 +924,86 @@ class DirectoryIndexer: Only touches roots that are readable right now — a root whose volume is absent has nothing to compare against, and comparing anyway is exactly - the mistake this module exists to avoid. + the mistake this module exists to avoid. A paused root is left alone: + the operator stopped its indexing, and this pass is indexing. """ changed = False - loop = asyncio.get_event_loop() for root in self.roots: - if not root.available: - continue - try: - 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) + if not root.available or root.paused: continue - - 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 - # copy under its own path, because the id is then absent. A - # dedicated "find the survivor" lookup was written first and - # deleted — it rehashed every file under the root on any single - # deletion, and a test proved it changed nothing. - self._index.remove_entry(known[missing]) - log.info("Reconcile: %s is gone", missing) + if await self._sweep_root(root): changed = True + return changed - added_paths = on_disk - set(known) - if not added_paths: - continue + async def _sweep_root(self, root: Root, *, kind: str = "reconcile") -> bool: + """ + One root's half of the sweep. Also what a resume runs: it indexes only + the files the index lacks, where a `_scan_root` would replace every + entry it met and lose what the enrichment passes had put there. + """ + changed = False + loop = asyncio.get_event_loop() + try: + 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) + return False - 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)) + 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 + # copy under its own path, because the id is then absent. A + # dedicated "find the survivor" lookup was written first and + # deleted — it rehashed every file under the root on any single + # deletion, and a test proved it changed nothing. + self._index.remove_entry(known[missing]) + log.info("Reconcile: %s is gone", missing) + changed = True + + added_paths = on_disk - set(known) + if not added_paths: + return changed - self._begin_job("", sum(size for _, size in added_sized), root=root, - kind="reconcile", files_total=len(added_sized)) + added_sized: list[tuple[Path, int]] = [] + for p in added_paths: 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 - self.progress.files_done += 1 - 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) + added_sized.append((p, p.stat().st_size)) + except OSError: + added_sized.append((p, 0)) + + self._begin_job("", sum(size for _, size in added_sized), root=root, + kind=kind, files_total=len(added_sized)) + try: + for done, (added, size) in enumerate(added_sized): + if self._stopped(root): + log.info("Indexing root %r stopped after %d of %d new files", + root.name, done, len(added_sized)) + break + self.progress.current_dir = added.parent.name + entry = await self._hash_or_cached(root, added) + self.progress.scanned_bytes += size + self.progress.files_done += 1 + 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) + if kind == "reconcile": log.info("Reconcile: %s appeared (missed event)", added) - changed = True - finally: - self._end_job() + changed = True + finally: + self._end_job() return changed def _entries_under(self, root: Root) -> list[IndexEntry]: @@ -1170,6 +1244,71 @@ class DirectoryIndexer: if self.on_change: await self.on_change(self) + # ── Pause and resume ────────────────────────────────────────────────────── + + async def pause_root(self, root_name: str) -> None: + """ + Stop indexing a root: the walk under way ends at the next file, and + neither the watcher nor reconcile touch it until it is resumed. Its + entries stay — they are files on a readable disk, served as before. + """ + root = self.roots.by_name(root_name) + if root is None or root.paused: + return + root.paused = True + self.progress.queued[:] = [n for n in self.progress.queued + if fold(n) != root.folded] + log.info("Root %r paused — %d entries kept, nothing more is indexed " + "until it is resumed", root.name, len(self._entries_under(root))) + self._restart_observer() + await self.publish_roots() + + async def resume_root(self, root_name: str) -> None: + """ + Index a paused root again, from where it stopped. + + Returns once the walk is queued, not when it ends: on a large drive + that is hours, and the operator's request has nothing to wait for — + the progress band shows the walk. Run as a task this indexer owns, like + a plug, so the session that asked can close without cancelling it. + """ + root = self.roots.by_name(root_name) + if root is None or not root.paused: + return + root.paused = False + log.info("Root %r resumed", root.name) + self._restart_observer() + await self.publish_roots() + if not root.available: + return + self._queue([root.name]) + task = asyncio.create_task(self._finish_resume(root)) + self._scan_tasks.add(task) + task.add_done_callback(self._scan_tasks.discard) + + async def _finish_resume(self, root: Root) -> None: + changed = False + waiting = True + try: + async with self._scan_lock: + self._unqueue([root.name]) + waiting = False + # Paused, ejected or removed again while it waited. + live = self._live(root) + if live is not None and live.available and not self._stopped(live): + changed = await self._sweep_root(live, kind="scan") + except asyncio.CancelledError: + raise + except Exception: + log.exception("Resuming root %r failed", root.name) + finally: + if waiting: + self._unqueue([root.name]) + if changed: + self._index.version = int(time.time()) + if self.on_change: + await self.on_change(self) + # ── Internal update ─────────────────────────────────────────────────────── def _schedule_update(self, file_path: Path, deleted: bool = False, @@ -1278,7 +1417,7 @@ class DirectoryIndexer: await self.on_change(self) return - if not root.available: + if not root.available or root.paused: return # On Windows watchdog cannot tell what a deleted path was -- it is diff --git a/packages/meshbay-node/src/meshbay_node/ops/__init__.py b/packages/meshbay-node/src/meshbay_node/ops/__init__.py index fc5b7ef..c88a909 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/ops/__init__.py @@ -84,8 +84,10 @@ from meshbay_node.ops.node_toml import ( from meshbay_node.ops.roots import ( add_root, eject_root, + pause_root, plug_root, remove_root, + resume_root, update_root, ) from meshbay_node.ops.settings import ( @@ -139,6 +141,7 @@ __all__ = [ "list_transfers", "open_chat_epoch", "pair_operator", + "pause_root", "plug_root", "prune_chat", "purge_chat", @@ -149,6 +152,7 @@ __all__ = [ "rematch_video", "remove_root", "resolve_user", + "resume_root", "revoke_member", "set_app_directories", "set_app_directory", diff --git a/packages/meshbay-node/src/meshbay_node/ops/roots.py b/packages/meshbay-node/src/meshbay_node/ops/roots.py index c272978..187e134 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/roots.py +++ b/packages/meshbay-node/src/meshbay_node/ops/roots.py @@ -1,4 +1,4 @@ -"""A group's directories: adding, removing, changing and ejecting them.""" +"""A group's directories: adding, removing, changing, ejecting and pausing them.""" from __future__ import annotations @@ -126,6 +126,12 @@ async def remove_root(state: dict, group_id: str, root_name: str) -> dict: _remove_roots_block(conf_path, group_id, resolved) cfg.roots.pop(match_idx) + # A directory added back later under the same name starts indexed, not + # paused by a decision about the one removed here. + roster = state.get("roster") + if roster: + await roster.set_root_paused(group_id, rname, False, + set_by=state.get("node_user_id", "")) # Not mutating the live set here either — see `add_root`. Dropping the # root from it would leave `retarget` unable to tell that its entries @@ -304,3 +310,68 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict: log.info("Root plugged: %s in group %s", root_name, group_id[:8]) return {"status": "plugged", "name": root_name, "group_id": group_id, "roots": roots.describe()} + + +def _live_root(state: dict, group_id: str, root_name: str): + """ + The group's live RootSet and the root named `root_name` in it. + + The indexer's set when the group has no context yet: a group added while + the node runs gets one only once its first scan ends, and that scan — a + whole new library — is the one most worth pausing. + """ + config = _config(state) + if not any(g.id == group_id for g in config.groups): + raise OpError("Group not configured on this node", status=404) + ctx = state.get("groups_ctx", {}).get(group_id) + indexer = state.get("indexers", {}).get(group_id) + roots: RootSet | None = (ctx.get("roots") if ctx else None) or ( + indexer.roots if indexer else None) + if not roots: + raise OpError("Group has no roots", status=503) + root = roots.by_name(root_name) + if root is None: + raise OpError(f"No root named {root_name!r} in this group", status=404) + return roots, root + + +async def pause_root(state: dict, group_id: str, root_name: str) -> dict: + """Stop indexing a root; what is already indexed stays listed.""" + roots, root = _live_root(state, group_id, root_name) + if root.paused: + return {"status": "already_paused", "name": root.name, + "group_id": group_id, "roots": roots.describe()} + + # Persisted first: a restart between the two must not start the scan the + # operator stopped. + await _roster(state).set_root_paused( + group_id, root.name, True, set_by=state.get("node_user_id", "")) + # The flag on the context's set too, as `eject_root` does, for a context + # whose indexer a retarget replaced. + indexer = state.get("indexers", {}).get(group_id) + if indexer: + await indexer.pause_root(root.name) + root.paused = True + + log.info("Root paused: %s in group %s", root.name, group_id[:8]) + return {"status": "paused", "name": root.name, "group_id": group_id, + "roots": roots.describe()} + + +async def resume_root(state: dict, group_id: str, root_name: str) -> dict: + """Index a paused root again; the walk runs in the background.""" + roots, root = _live_root(state, group_id, root_name) + if not root.paused: + return {"status": "already_resumed", "name": root.name, + "group_id": group_id, "roots": roots.describe()} + + await _roster(state).set_root_paused( + group_id, root.name, False, set_by=state.get("node_user_id", "")) + indexer = state.get("indexers", {}).get(group_id) + if indexer: + await indexer.resume_root(root.name) + root.paused = False + + log.info("Root resumed: %s in group %s", root.name, group_id[:8]) + return {"status": "resumed", "name": root.name, "group_id": group_id, + "roots": roots.describe()} diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 7dded1f..61fdc9b 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -182,6 +182,10 @@ class Root: # such a root comes back on its own once its files are there again # (`DirectoryIndexer._reconcile`). An operator's eject never does. auto: bool = False + # The operator stopped indexing it: not walked, not watched, not re-checked, + # and what was already indexed stays listed. Independent of `available` — + # a paused root is readable, and its files are served. + paused: bool = False available: bool = True @property @@ -341,14 +345,16 @@ class RootSet: # Backward compat: old configs use `upload` instead of `writable` writable = bool(spec.get("writable", spec.get("upload", False))) - # `ejected` is runtime state, not configuration — it reaches here - # only from the roster, restored at startup so a drive ejected - # before a restart does not come back on its own. + # `ejected` and `paused` are runtime state, not configuration — they + # reach here only from the roster, restored at startup so a drive + # ejected before a restart does not come back on its own, and a + # paused scan does not start again. root = Root(name=name, path=path, kind=kind, writable=writable, removable=bool(spec.get("removable", False)), ejected=bool(spec.get("ejected", False)), auto=bool(spec.get("ejected_auto", False)), + paused=bool(spec.get("paused", False)), available=not bool(spec.get("ejected", False))) _refuse_nesting(root, roots) roots.append(root) @@ -496,7 +502,8 @@ class RootSet: "available": r.available, "writable": r.writable, "removable": r.removable, - "ejected": r.ejected} + "ejected": r.ejected, + "paused": r.paused} # `with_paths` is for the operator's *own* channels only — the # loopback API and the CLI reading it, both of which already # require being on this machine with the run token. A member is diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index a5d1967..9336e38 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -737,6 +737,31 @@ class Roster: return {r["key"][len(prefix):] for r in rows if r["key"].startswith(prefix) and r["value"] == "auto"} + # Whether the operator paused a root's indexing. Kept here beside + # `root_ejected:` for the same reasons: runtime state rather than + # configuration, and a pause that a restart undid would start the very scan + # the operator stopped. + SETTING_ROOT_PAUSED_PREFIX = "root_paused:" + + @classmethod + def root_paused_key(cls, root_name: str) -> str: + return cls.SETTING_ROOT_PAUSED_PREFIX + fold(root_name) + + async def set_root_paused(self, group_id: str, root_name: str, + paused: bool, set_by: str = "") -> None: + await self.set_setting(group_id, self.root_paused_key(root_name), + "1" if paused else "0", set_by) + + async def paused_roots(self, group_id: str) -> set[str]: + """The folded names of this group's paused roots — matched like `ejected_roots`.""" + prefix = self.SETTING_ROOT_PAUSED_PREFIX + async with self._db.execute( + "SELECT key, value FROM group_settings WHERE group_id = ?", + (group_id,)) as cur: + rows = await cur.fetchall() + return {r["key"][len(prefix):] for r in rows + if r["key"].startswith(prefix) and r["value"] == "1"} + async def get_setting(self, group_id: str, key: str, default: str | None = None) -> str | None: async with self._db.execute( diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py index 7b3b1c9..aab3691 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py @@ -24,8 +24,10 @@ from meshbay_common.adminop import ( OP_MEMBER_REVOKE, OP_MUSICBRAINZ_ENABLED, OP_ROOT_EJECT, + OP_ROOT_PAUSE, OP_ROOT_PLUG, OP_ROOT_REMOVE, + OP_ROOT_RESUME, OP_SEARCH_LISTED, OP_SET_SCAN_SETTINGS, OP_TMDB_CONFIG, @@ -72,6 +74,8 @@ _ADMIN_EXECUTORS = { OP_CHAT_PURGE: "_admin_exec_chat_purge", OP_ROOT_EJECT: "_admin_exec_root_eject", OP_ROOT_PLUG: "_admin_exec_root_plug", + OP_ROOT_PAUSE: "_admin_exec_root_pause", + OP_ROOT_RESUME: "_admin_exec_root_resume", } diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py index 5d4ed94..0ceefb5 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py @@ -71,6 +71,8 @@ _HANDLERS = { MNP.ROOT_REMOVE: ("_do_root_remove", INLINE), MNP.ROOT_EJECT: ("_do_root_eject", INLINE), MNP.ROOT_PLUG: ("_do_root_plug", INLINE), + MNP.ROOT_PAUSE: ("_do_root_pause", INLINE), + MNP.ROOT_RESUME: ("_do_root_resume", INLINE), MNP.KEYPAIR_BUNDLE_STORE: ("_do_keypair_bundle_store", SPAWNED), MNP.KEYPAIR_BUNDLE_DELETE: ("_do_keypair_bundle_delete", SPAWNED), MNP.USER_BLOB_STORE: ("_do_user_blob_store", SPAWNED), diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py index 237a359..c1f424d 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py @@ -1,5 +1,5 @@ """The operator's controls over a group's roots and scan pacing that MNP carries: -removing, ejecting and plugging a root. What widens the sharing, and the node's +removing, ejecting, plugging, pausing and resuming a root. What widens the sharing, and the node's own status, settings, roster and denylist, are the loopback API's and the CLI's (MNP 6.0).""" @@ -8,8 +8,10 @@ import logging from meshbay_common import MNP_VERSION from meshbay_common.adminop import ( OP_ROOT_EJECT, + OP_ROOT_PAUSE, OP_ROOT_PLUG, OP_ROOT_REMOVE, + OP_ROOT_RESUME, OP_SET_SCAN_SETTINGS, ) from meshbay_common.protocol import MNP @@ -201,6 +203,64 @@ class NodeOpsMixin: except Exception: pass + def _do_root_pause(self, msg: dict) -> None: + self._ask_root_index_op(msg, OP_ROOT_PAUSE) + + def _do_root_resume(self, msg: dict) -> None: + self._ask_root_index_op(msg, OP_ROOT_RESUME) + + def _ask_root_index_op(self, msg: dict, op: str) -> None: + target_group = str(msg.get("group_id", self._group_id or "")).strip() + root_name = str(msg.get("root_name", "")).strip() + if not target_group or not root_name: + self._send({"type": "error", "detail": "Missing group_id or root_name"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + op, root_name, + payload={"group_id": target_group, "root_name": root_name}, + group_id=target_group) + + async def _admin_exec_root_pause( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + await self._exec_root_index_op( + pending, transcript, sig, OP_ROOT_PAUSE, ops.pause_root, MNP.ROOT_PAUSE_ACK) + + async def _admin_exec_root_resume( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + await self._exec_root_index_op( + pending, transcript, sig, OP_ROOT_RESUME, ops.resume_root, + MNP.ROOT_RESUME_ACK) + + async def _exec_root_index_op(self, pending: dict, transcript: bytes, sig: bytes, + op: str, run, ack: str) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"{op}:{pending['subject'][:24]}") + return + p = pending["payload"] + try: + result = await self._run_op(run, p["group_id"], p["root_name"]) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + except Exception as e: + log.error("%s failed: %s", op, e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + return + self._audit(op, p["root_name"]) + # Broadcast, like eject's: every open roots table shows the new state. + notice = {"type": ack, "v": MNP_VERSION, **result} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + async def _retarget_indexer(self, group_id: str) -> None: """ Pick up a root that was just removed from node.toml. diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 03db8b6..9933750 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -449,6 +449,16 @@ def create_ui_app(state: dict) -> FastAPI: """Bring an ejected folder back.""" return await _op(lambda: ops.plug_root(state, group_id, root_name)) + @app.put("/api/groups/{group_id}/roots/{root_name}/pause") + async def pause_root(group_id: str, root_name: str): + """Stop indexing a folder. What is already indexed stays listed.""" + return await _op(lambda: ops.pause_root(state, group_id, root_name)) + + @app.put("/api/groups/{group_id}/roots/{root_name}/resume") + async def resume_root(group_id: str, root_name: str): + """Index a paused folder again, from where it stopped.""" + return await _op(lambda: ops.resume_root(state, group_id, root_name)) + @app.delete("/api/groups/{group_id}/roots/{root_name}") async def remove_root(group_id: str, root_name: str): """Remove a folder from a group. At least one must remain.""" diff --git a/packages/meshbay-node/tests/golden/cli.json b/packages/meshbay-node/tests/golden/cli.json index 48d2f74..d069510 100644 --- a/packages/meshbay-node/tests/golden/cli.json +++ b/packages/meshbay-node/tests/golden/cli.json @@ -4,7 +4,7 @@ "asked": [], "exit": 0, "stderr": "", - "stdout": "usage: meshbay-node [-h] [--hub-url HUB_URL] [--username USERNAME] [--dir DIR] [--yes]\n [--config CONFIG] [--group GROUP] [--link] [--writable] [--no-writable]\n [--removable] [--no-removable] [--open] [--name NAME]\n [--log-level {DEBUG,INFO,WARNING,ERROR}]\n [{init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}]\n [subcommand] [target] [value]\n\nMeshBay Node daemon\n\npositional arguments:\n {init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}\n init: provision config + keystore | reset: erase all node state | status:\n node state and keys | operator pair: pair a browser with this node |\n member list|invite|cancel|revoke|unpin | group list|add|remove | root\n list|add|remove|set|eject|plug | gek init|rotate | file list|rm | video\n rematch: re-resolve TMDB matches for a group's videos | chat\n status|rotate|encrypt-history|prune | denylist show|clear | stun\n list|add|remove|reset | transfers show|set|max-size|per-member: live\n transfer slots, the node-wide caps, the largest single upload, and how\n many one member may run at once in a group | reload: re-read node.toml\n (hot; systemd or the loopback API) | restart-daemon: restart the node\n (systemd unit, the Windows autostart launcher, or the service task,\n whichever applies) | autostart install|remove|start|stop|status (Windows:\n run meshbay-node at each sign-in, no admin) | service\n install|remove|start|stop|status (Windows: run at boot, before sign-in,\n needs admin once to install) | calibrate-argon2: benchmark\n subcommand 'pair' for operator; list|invite|revoke|unpin for member; list|add|remove\n for group; list|add|remove|set|eject|plug for root; init|rotate for gek;\n list|rm for file; rematch for video; show|clear for denylist;\n list|add|remove|reset for stun; show|set|max-size|per-member for\n transfers; install|remove|start|stop|status for autostart and for service\n target username for member invite|revoke|unpin (an optional e-mail label with\n --link, a link id for member cancel); group name for group add; file id\n for file rm; identifier for denylist clear; download cap for transfers\n set; size in GB for transfers max-size\n value the second value where a verb takes two: the upload cap for transfers set\n\noptions:\n -h, --help show this help message and exit\n --hub-url HUB_URL hub URL, for init (e.g. https://meshbay.org)\n --username USERNAME hub username, for init\n --dir DIR shared directory, for group add\n --yes skip the confirmation for destructive commands\n --config CONFIG Config file path\n --group GROUP group id (optional if only one is configured)\n --link member invite: an invitation link, for someone who may have no account yet\n (valid 7 days, single use)\n --writable root accepts member uploads (root add/set)\n --no-writable root is read-only (root add/set, group add)\n --removable mark root as removable (root set/add)\n --no-removable mark root as not removable (root set)\n --open group add: anyone the hub lists the group to may join (default: by\n invitation only)\n --name NAME root name (root add; defaults to directory basename)\n --log-level {DEBUG,INFO,WARNING,ERROR}\n", + "stdout": "usage: meshbay-node [-h] [--hub-url HUB_URL] [--username USERNAME] [--dir DIR] [--yes]\n [--config CONFIG] [--group GROUP] [--link] [--writable] [--no-writable]\n [--removable] [--no-removable] [--open] [--name NAME]\n [--log-level {DEBUG,INFO,WARNING,ERROR}]\n [{init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}]\n [subcommand] [target] [value]\n\nMeshBay Node daemon\n\npositional arguments:\n {init,reset,status,gek-init,gek,operator,member,group,root,file,video,chat,denylist,stun,transfers,reload,restart-daemon,autostart,service,calibrate-argon2}\n init: provision config + keystore | reset: erase all node state | status:\n node state and keys | operator pair: pair a browser with this node |\n member list|invite|cancel|revoke|unpin | group list|add|remove | root\n list|add|remove|set|eject|plug|pause|resume | gek init|rotate | file\n list|rm | video rematch: re-resolve TMDB matches for a group's videos |\n chat status|rotate|encrypt-history|prune | denylist show|clear | stun\n list|add|remove|reset | transfers show|set|max-size|per-member: live\n transfer slots, the node-wide caps, the largest single upload, and how\n many one member may run at once in a group | reload: re-read node.toml\n (hot; systemd or the loopback API) | restart-daemon: restart the node\n (systemd unit, the Windows autostart launcher, or the service task,\n whichever applies) | autostart install|remove|start|stop|status (Windows:\n run meshbay-node at each sign-in, no admin) | service\n install|remove|start|stop|status (Windows: run at boot, before sign-in,\n needs admin once to install) | calibrate-argon2: benchmark\n subcommand 'pair' for operator; list|invite|revoke|unpin for member; list|add|remove\n for group; list|add|remove|set|eject|plug|pause|resume for root;\n init|rotate for gek; list|rm for file; rematch for video; show|clear for\n denylist; list|add|remove|reset for stun; show|set|max-size|per-member for\n transfers; install|remove|start|stop|status for autostart and for service\n target username for member invite|revoke|unpin (an optional e-mail label with\n --link, a link id for member cancel); group name for group add; file id\n for file rm; identifier for denylist clear; download cap for transfers\n set; size in GB for transfers max-size\n value the second value where a verb takes two: the upload cap for transfers set\n\noptions:\n -h, --help show this help message and exit\n --hub-url HUB_URL hub URL, for init (e.g. https://meshbay.org)\n --username USERNAME hub username, for init\n --dir DIR shared directory, for group add\n --yes skip the confirmation for destructive commands\n --config CONFIG Config file path\n --group GROUP group id (optional if only one is configured)\n --link member invite: an invitation link, for someone who may have no account yet\n (valid 7 days, single use)\n --writable root accepts member uploads (root add/set)\n --no-writable root is read-only (root add/set, group add)\n --removable mark root as removable (root set/add)\n --no-removable mark root as not removable (root set)\n --open group add: anyone the hub lists the group to may join (default: by\n invitation only)\n --name NAME root name (root add; defaults to directory basename)\n --log-level {DEBUG,INFO,WARNING,ERROR}\n", "systemctl": [] }, "autostart no-such-sub": { @@ -550,7 +550,7 @@ "asked": [], "exit": 1, "stderr": "", - "stdout": "usage: meshbay-node root list|add|remove|set|eject|plug [name] [--group NAME]\n", + "stdout": "usage: meshbay-node root list|add|remove|set|eject|plug|pause|resume [name] [--group NAME]\n", "systemctl": [] }, "root plug media": { diff --git a/packages/meshbay-node/tests/golden/dispatch.json b/packages/meshbay-node/tests/golden/dispatch.json index a39c57d..384c38f 100644 --- a/packages/meshbay-node/tests/golden/dispatch.json +++ b/packages/meshbay-node/tests/golden/dispatch.json @@ -144,6 +144,14 @@ "_admin_exec_root_eject" ] }, + "root_pause": { + "audit": [], + "log": [], + "sent": [], + "spawned": [ + "_admin_exec_root_pause" + ] + }, "root_plug": { "audit": [], "log": [], @@ -160,6 +168,14 @@ "_admin_exec_root_remove" ] }, + "root_resume": { + "audit": [], + "log": [], + "sent": [], + "spawned": [ + "_admin_exec_root_resume" + ] + }, "search_listed": { "audit": [], "log": [], @@ -2020,7 +2036,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2343,7 +2359,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2362,7 +2378,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2381,7 +2397,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2400,7 +2416,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2670,7 +2686,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2684,7 +2700,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2698,7 +2714,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2712,7 +2728,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2726,7 +2742,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2740,7 +2756,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2754,7 +2770,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -2768,7 +2784,7 @@ "messages": [], "req_id": 4242, "type": "chat_hist_resp", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -3983,7 +3999,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -4002,7 +4018,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -4021,7 +4037,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -4040,7 +4056,7 @@ "subject": "gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -10159,7 +10175,7 @@ "subject": "link:gggggggggggggggggggggggggggggggg", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -12258,7 +12274,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -12277,7 +12293,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -12296,7 +12312,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13397,7 +13413,7 @@ "req_id": 4242, "token": null, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13412,7 +13428,7 @@ "x" ], "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13425,7 +13441,7 @@ "req_id": 4242, "token": 7, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13438,7 +13454,7 @@ "req_id": 4242, "token": "x", "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13451,7 +13467,7 @@ "req_id": 4242, "token": null, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13466,7 +13482,7 @@ "x" ], "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13479,7 +13495,7 @@ "req_id": 4242, "token": 7, "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13492,7 +13508,7 @@ "req_id": 4242, "token": "x", "type": "pong", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13827,7 +13843,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13846,7 +13862,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -13865,7 +13881,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14030,6 +14046,379 @@ "sent": [], "spawned": [] }, + "root_pause | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | member | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | operator | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause | operator | lists": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "['x']", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_pause", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "['x']", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_pause | operator | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "7", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_pause", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "7", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_pause | operator | strings": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "x", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_pause", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "x", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_pause_ack | member | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | member | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | member | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | member | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_pause_ack | operator | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, "root_plug | challenged | bare": { "audit": [], "log": [], @@ -14200,7 +14589,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14219,7 +14608,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14238,7 +14627,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14573,7 +14962,7 @@ "subject": "['x']", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14592,7 +14981,7 @@ "subject": "7", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14611,7 +15000,7 @@ "subject": "x", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -14776,6 +15165,379 @@ "sent": [], "spawned": [] }, + "root_resume | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | member | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "No authorized key for this", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | operator | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Missing group_id or root_name", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume | operator | lists": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "['x']", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_resume", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "['x']", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_resume | operator | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "7", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_resume", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "7", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_resume | operator | strings": { + "audit": [], + "log": [], + "sent": [ + { + "group_id": "x", + "node_pk": "iojj3XQJ8ZX9UtstPLpdcspnCb8dlBIb83SIAbQPb1w=", + "nonce": "<volatile>", + "op": "root_resume", + "op_id": "<volatile>", + "req_id": 4242, + "subject": "x", + "ts": "<volatile>", + "type": "admin_challenge", + "v": "6.2" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | challenged | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | bare": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | lists": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | numbers": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | fresh | strings": { + "audit": [], + "log": [], + "sent": [ + { + "detail": "Handshake required", + "req_id": 4242, + "type": "error" + } + ], + "spawned": [] + }, + "root_resume_ack | member | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | member | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | member | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | member | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | bare": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | lists": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | numbers": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, + "root_resume_ack | operator | strings": { + "audit": [], + "log": [ + "WARNING Unknown MNP message type on DataChannel: %s" + ], + "sent": [], + "spawned": [] + }, "search_listed | challenged | bare": { "audit": [], "log": [], @@ -17238,7 +18000,7 @@ "subject": "{\"language\":null,\"token\":null}", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] @@ -17281,7 +18043,7 @@ "subject": "{\"language\":\"x\",\"token\":\"sha256:2d711642b726b04401627ca9fbac32f5c8530fb1903cc4db02258717921a4881\"}", "ts": "<volatile>", "type": "admin_challenge", - "v": "6.1" + "v": "6.2" } ], "spawned": [] diff --git a/packages/meshbay-node/tests/test_root_availability.py b/packages/meshbay-node/tests/test_root_availability.py index 7cc890f..d31e11b 100644 --- a/packages/meshbay-node/tests/test_root_availability.py +++ b/packages/meshbay-node/tests/test_root_availability.py @@ -120,7 +120,7 @@ async def test_members_are_told_which_roots_are_unavailable(tmp_path): idx = await _indexer(_roots(films)) assert idx.index.roots == [ {"name": "Films", "kind": "generic", "available": True, - "writable": True, "removable": False, "ejected": False}] + "writable": True, "removable": False, "ejected": False, "paused": False}] (films / "a.mkv").unlink() films.rmdir() diff --git a/packages/meshbay-node/tests/test_root_paths_are_operator_only.py b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py index 646c7b4..edbc62f 100644 --- a/packages/meshbay-node/tests/test_root_paths_are_operator_only.py +++ b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py @@ -102,7 +102,7 @@ def test_the_cli_only_reads_fields_the_payload_carries(monkeypatch, tmp_path, ca path = Path("/tmp/x") name = "x" kind = "generic" - writable = removable = ejected = False + writable = removable = ejected = paused = False available = True offered = RootSet(roots=[_Any()]).describe(with_paths=True)[0] diff --git a/packages/meshbay-node/tests/test_root_pause.py b/packages/meshbay-node/tests/test_root_pause.py new file mode 100644 index 0000000..d9a4fcb --- /dev/null +++ b/packages/meshbay-node/tests/test_root_pause.py @@ -0,0 +1,313 @@ +""" +Pausing a root's indexing, and the walk that has to notice it. + +A scan of a large drive runs for hours, and nothing could stop one: the hashing +loop never looked at the root again once it had the file list. Removing the +directory let it read the whole drive anyway and threw the result away at the +end; ejecting it went on reading a drive the operator was about to unplug. + +A pause has to outlast the walk it stops. Reconcile indexes every file the index +lacks, so a walk that was merely cancelled was redone at the next tick; the flag +is what reconcile, the watcher and a restart all read. And it keeps what was +indexed: the index is rebuilt at every start, so a paused root is listed from the +hash cache rather than left empty. +""" + +import asyncio +from pathlib import Path +from types import SimpleNamespace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node import ops +from meshbay_node.config import GroupConfig, RootSpec +from meshbay_node.indexer import indexer as indexer_mod +from meshbay_node.indexer.cache import IndexCache +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.roster import Roster + +pytestmark = pytest.mark.asyncio + +GID = "g" * 32 + + +def _films(tmp_path: Path, n: int = 6) -> Path: + films = tmp_path / "Films" + films.mkdir() + for i in range(n): + (films / f"f{i}.mkv").write_bytes(f"film {i}".encode()) + return films + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _count_reads(monkeypatch, after=None, at: int = 0) -> list[str]: + """Record every file actually read; run `after` once `at` have been.""" + read: list[str] = [] + real = indexer_mod._scan_file + + def counting(root, path): + read.append(path.name) + if after is not None and len(read) == at: + after() + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", counting) + return read + + +# ── The walk notices ───────────────────────────────────────────────────────── + +async def test_a_pause_stops_the_walk_at_the_next_file(tmp_path, monkeypatch): + roots = RootSet.build([{"path": str(_films(tmp_path))}]) + idx = _indexer(roots) + read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), at=2) + + await idx.initial_scan() + + assert len(read) == 2, f"read {len(read)} files after the pause" + assert len(_names(idx)) == 2, "the files hashed before the pause stay listed" + assert idx.progress.scanning is False + + +async def test_a_root_removed_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch): + films = _films(tmp_path) + other = tmp_path / "Other" + other.mkdir() + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + + def remove(): + idx.roots = RootSet.build([{"path": str(other)}]) + + read = _count_reads(monkeypatch, remove, at=1) + await idx.initial_scan() + + assert len(read) == 1, f"went on reading a removed root: {read}" + + +async def test_a_root_ejected_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch): + roots = RootSet.build([{"path": str(_films(tmp_path)), "removable": True}]) + idx = _indexer(roots) + read = _count_reads(monkeypatch, lambda: idx.eject_root("Films"), at=1) + + await idx.initial_scan() + + assert len(read) == 1, f"went on reading an ejected drive: {read}" + + +async def test_a_retarget_does_not_hide_the_pause_from_a_running_walk( + tmp_path, monkeypatch): + """The walk holds the Root it started with; a reload replaces every one.""" + films = _films(tmp_path) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + + def reload_then_pause(): + idx.roots = RootSet.build([{"path": str(films)}]) + idx.roots.roots[0].paused = True + + read = _count_reads(monkeypatch, reload_then_pause, at=1) + await idx.initial_scan() + + assert len(read) == 1 + + +# ── The pause outlasts the walk ────────────────────────────────────────────── + +async def test_reconcile_leaves_a_paused_root_alone(tmp_path): + films = _films(tmp_path, n=1) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + await idx.pause_root("Films") + + (films / "new.mkv").write_bytes(b"new film") + await idx.reconcile() + + assert "new.mkv" not in _names(idx), "reconcile redid the work the pause stopped" + + +async def test_the_watcher_leaves_a_paused_root_alone(tmp_path): + films = _films(tmp_path, n=1) + idx = _indexer(RootSet.build([{"path": str(films)}])) + await idx.initial_scan() + await idx.pause_root("Films") + + new = films / "new.mkv" + new.write_bytes(b"new film") + await idx._update_entry(new, deleted=False) + + assert "new.mkv" not in _names(idx) + + +async def test_a_paused_root_is_listed_from_the_cache_and_nothing_is_read( + tmp_path, monkeypatch): + """ + After a restart the index is empty until the walk refills it. Skipping a + paused root would empty it for members; reading it would undo the pause. + """ + films = _films(tmp_path, n=3) + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + try: + await _indexer(RootSet.build([{"path": str(films)}]), cache=cache).initial_scan() + (films / "unhashed.mkv").write_bytes(b"never hashed") + + roots = RootSet.build([{"path": str(films), "paused": True}]) + idx = _indexer(roots, cache=cache) + read = _count_reads(monkeypatch) + await idx.initial_scan() + + assert read == [] + assert _names(idx) == {"f0.mkv", "f1.mkv", "f2.mkv"} + assert idx.progress.scanning is False + finally: + await cache.close() + + +async def test_a_pause_mid_rescan_keeps_the_entries_it_had(tmp_path, monkeypatch): + """A rescan drops the root's entries before it walks; stopping it there + would leave the operator with half a library.""" + films = _films(tmp_path, n=4) + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + try: + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots, cache=cache) + await idx.initial_scan() + for f in films.iterdir(): + f.write_bytes(f.read_bytes() + b" edited") # every cache row stale + + read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), + at=1) + await idx._rescan_root(roots.roots[0]) + + assert len(read) == 1 + assert len(_names(idx)) == 1 + finally: + await cache.close() + + +# ── Resume ─────────────────────────────────────────────────────────────────── + +async def test_resume_indexes_what_is_missing_and_keeps_what_was_enriched(tmp_path): + films = _films(tmp_path, n=2) + idx = _indexer(RootSet.build([{"path": str(films)}])) + await idx.initial_scan() + for e in idx.index.entries: + e.artist = "Some Artist" + await idx.pause_root("Films") + (films / "new.mkv").write_bytes(b"new film") + + await idx.resume_root("Films") + await asyncio.gather(*idx._scan_tasks) + + assert "new.mkv" in _names(idx) + kept = [e for e in idx.index.entries if e.name != "new.mkv"] + assert all(e.artist == "Some Artist" for e in kept), \ + "resume replaced entries it should have left alone" + + +async def test_resume_returns_before_the_walk_ends(tmp_path): + films = _films(tmp_path, n=1) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + await idx.pause_root("Films") + + async with idx._scan_lock: + await asyncio.wait_for(idx.resume_root("Films"), timeout=1) + assert idx.progress.queued == ["Films"] + await asyncio.gather(*idx._scan_tasks) + assert idx.progress.queued == [] + + +# ── The flag is the operator's, and survives ───────────────────────────────── + +def _state(tmp_path: Path, roots: RootSet, roster: Roster) -> dict: + index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}}, + "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]), + "roster": roster, + "node_user_id": "operator", + } + + +async def test_pause_and_resume_persist_through_the_roster(tmp_path): + films = _films(tmp_path, n=1) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + state = _state(tmp_path, RootSet.build([{"path": str(films)}]), roster) + + out = await ops.pause_root(state, GID, "films") + assert out["status"] == "paused" + assert out["roots"][0]["paused"] is True + assert out["roots"][0]["available"] is True, "a paused root is still served" + assert await roster.paused_roots(GID) == {"films"} + assert (await ops.pause_root(state, GID, "Films"))["status"] == "already_paused" + + out = await ops.resume_root(state, GID, "Films") + assert out["status"] == "resumed" + assert await roster.paused_roots(GID) == set() + finally: + await roster.close() + + +async def test_removing_a_root_forgets_its_pause(tmp_path): + """A directory added back later under the same name starts indexed.""" + films = _films(tmp_path, n=1) + keep = tmp_path / "Keep" + keep.mkdir() + conf = tmp_path / "node.toml" + conf.write_text( + f'[[groups]]\nid = "{GID}"\nname = "g"\n\n' + f'[[groups.roots]]\npath = "{films.as_posix()}"\n\n' + f'[[groups.roots]]\npath = "{keep.as_posix()}"\n') + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + state = _state(tmp_path, RootSet.build( + [{"path": str(films)}, {"path": str(keep)}]), roster) + state["config_path"] = str(conf) + state["config"] = SimpleNamespace(groups=[GroupConfig( + id=GID, roots=[RootSpec(path=str(films)), RootSpec(path=str(keep))])]) + await ops.pause_root(state, GID, "Films") + + await ops.remove_root(state, GID, "Films") + + assert await roster.paused_roots(GID) == set() + finally: + await roster.close() + + +async def test_a_new_group_can_be_paused_during_its_first_scan(tmp_path): + """A group added while the node runs has an indexer, and no context until + that first scan ends — the scan most worth pausing.""" + films = _films(tmp_path, n=1) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + state = _state(tmp_path, roots, roster) + state["groups_ctx"] = {} + state["indexers"] = {GID: idx} + + out = await ops.pause_root(state, GID, "Films") + + assert out["status"] == "paused" + assert roots.roots[0].paused is True + finally: + await roster.close() diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py index 0fe9f03..30bbc8a 100644 --- a/packages/meshbay-node/tests/test_roots.py +++ b/packages/meshbay-node/tests/test_roots.py @@ -290,9 +290,9 @@ def test_describe_reports_what_a_member_needs(tmp_path): described = roots.describe() assert described == [ {"name": "Media", "kind": "generic", "available": True, - "writable": True, "removable": False, "ejected": False}, + "writable": True, "removable": False, "ejected": False, "paused": False}, {"name": "Music", "kind": "audio", "available": True, - "writable": False, "removable": True, "ejected": False}, + "writable": False, "removable": True, "ejected": False, "paused": False}, ] # Deliberately no paths: a member is told what exists and whether it is # readable, not where on the operator's disk it lives. @@ -301,15 +301,16 @@ def test_describe_reports_what_a_member_needs(tmp_path): def test_describe_names_a_root_and_never_a_path(tmp_path): """ - What a member is told about a root: that it exists, and whether it is - readable and writable. Never where on the operator's disk it lives — that - is the operator's own view (`with_paths`), over their own channel. + What a member is told about a root: that it exists, whether it is + readable and writable, and whether its indexing is paused. Never where on + the operator's disk it lives — that is the operator's own view + (`with_paths`), over their own channel. """ (tmp_path / "Media").mkdir() roots = RootSet.build([_spec(tmp_path / "Media", writable=True)]) described = roots.describe()[0] assert set(described) == {"name", "kind", "available", "writable", - "removable", "ejected"} + "removable", "ejected", "paused"} # ── SAFE_UPLOAD_NAME ──────────────────────────────────────────────────────── |