diff options
46 files changed, 1848 insertions, 176 deletions
diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md index 9c63c52..d0b5475 100644 --- a/docs/MESHBAY_DESIGN.md +++ b/docs/MESHBAY_DESIGN.md @@ -16,7 +16,7 @@ > them — it names the invariant that holds today, not the incident that produced > it. §13 is the register of those labels. > -> Wire versions at the time of writing: **MNP 6.1** (oldest peer accepted 4.0), +> Wire versions at the time of writing: **MNP 6.2** (oldest peer accepted 4.0), > **MHP 0.1**, packages **0.19.0**. The normative source for the wire format is > `MESHBAY_NODE_PROTOCOL.md`; this document states the design the protocol > serves, not its byte layout. @@ -1231,7 +1231,7 @@ from anything in the response. | `invite_link_create`, `invite_cancel` | the operator | | group key init and rotation | **local admin API or CLI only**; the node generates the key itself | | root add, root `writable`/`removable`, hosting a group, transfer limits | **local admin API or CLI only** (MNP 6.0) | -| root remove/eject/plug, `apps_enabled`, app directories | operator-signed | +| root remove/eject/plug/pause/resume, `apps_enabled`, app directories | operator-signed | | ~~`gek_bundle_store`~~ | **the message does not exist.** No member ever hands the node key material | `gek_bundle_store` was deleted rather than gated. The operator's X25519 public key @@ -1641,8 +1641,9 @@ A signed op proves that the operator's key signed, not that they meant it: in a browser that key is driven by code the hub serves (T3), and in the desktop application by a renderer that parses content from nodes. Either could otherwise have shared any folder on the machine, writable, from anywhere. The operator still -sees every root and its flags from any browser; removing, ejecting and plugging stay -signed ops, because they narrow what is shared or restore what already was. +sees every root and its flags from any browser; removing, ejecting, plugging, pausing +and resuming stay signed ops, because they narrow what is shared or restore what +already was. > **There is one answer to "may this member write", and it is the root.** A single > flag over the group cannot express "this library is published read-only and that @@ -1670,6 +1671,12 @@ states: - **`available`** — computed as `not ejected and is_live()`. This is what clients and the indexer see. +A third, **`paused`**, is the operator's alone and independent of both: the root is +readable and served, and its indexing is stopped (§6.3). Persisted in `roster.db` +beside `ejected`, for the same reason: a pause that a restart undid would start the +scan the operator stopped. Removing a root forgets its pause, so a directory added +back under the same name starts indexed. + The distinction matters: between clicking eject and physically unplugging, `is_live()` is still true, and without `ejected` the availability sweep would immediately flip the root back. @@ -1755,6 +1762,24 @@ reason to show progress. `index_progress` is deliberately **not** sealed: counters only, every 2 s. +**A walk looks at its root before every file.** A root removed from the set or +ejected while it is walked is not read to the end: its entries are dropped, or +frozen on a drive that is leaving. The walk reads the root from the current set, +not the object it started with, because a reload replaces every root. + +**A pause outlasts the walk it stops.** Reconcile indexes every file the index +lacks, so a walk that was merely cancelled is redone at the next tick. A paused +root is therefore not watched, not swept by reconcile, and not hashed by any walk: + +- **A walk that meets a pause finishes from the hash cache alone.** What the node + already hashed stays listed; nothing new is read. Stopping dead is not enough, + because the index is rebuilt at every start and a rescan drops the root's + entries before it walks — either would leave a paused root half empty. +- **Resuming runs the reconcile sweep over that root, never a scan.** A scan + replaces every entry it meets, and with them what the enrichment passes put + there; the sweep indexes only what the index lacks. The request returns once + the walk is queued, and the progress band shows it. + ### 6.4 Uploads Five protections, and they are the substance: @@ -1989,7 +2014,7 @@ before its key is linked or a group exists. ``` meshbay-node status meshbay-node group list|add|remove -meshbay-node root list|add|remove|set|eject|plug +meshbay-node root list|add|remove|set|eject|plug|pause|resume meshbay-node gek init|rotate meshbay-node operator pair meshbay-node member list|invite [--link]|cancel|revoke|unpin diff --git a/docs/MESHBAY_HTTP_API.md b/docs/MESHBAY_HTTP_API.md index 1bcb55d..73d079d 100644 --- a/docs/MESHBAY_HTTP_API.md +++ b/docs/MESHBAY_HTTP_API.md @@ -229,6 +229,8 @@ draws a new one at each start and deletes the file when it stops. | PATCH | `/api/groups/{group_id}/roots/{root_name}` | Make a folder writable or removable, or not. | | PUT | `/api/groups/{group_id}/roots/{root_name}/eject` | Eject a removable folder so its disk can be unplugged. | | PUT | `/api/groups/{group_id}/roots/{root_name}/plug` | Bring an ejected folder back. | +| PUT | `/api/groups/{group_id}/roots/{root_name}/pause` | Stop indexing a folder. | +| PUT | `/api/groups/{group_id}/roots/{root_name}/resume` | Index a paused folder again, from where it stopped. | | DELETE | `/api/groups/{group_id}/roots/{root_name}` | Remove a folder from a group. | | GET | `/api/groups/{group_id}/index-status` | One group's indexing progress. | | GET | `/api/index-status` | Every group's indexing progress. | diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md index 994c908..207d640 100644 --- a/docs/MESHBAY_NODE_PROTOCOL.md +++ b/docs/MESHBAY_NODE_PROTOCOL.md @@ -1168,6 +1168,7 @@ broadcast, every connected peer in the group learns the change without reconnect | `musicbrainz_enabled` | `enabled` | operator | `musicbrainz_enabled_ack{enabled}` | yes | | `root_remove` | the root name | operator | `root_remove_ack` | no | | `root_eject`, `root_plug` | the root name | operator | `root_eject_ack` / `root_plug_ack` | yes | +| `root_pause`, `root_resume` | the root name | operator | `root_pause_ack` / `root_resume_ack` | yes | | `app_directories` | `<app>:<dir>,<dir>,...` | operator | `app_directories_ack{app, dirs}` | yes | | `chat_directory` | the path | operator | `chat_directory_ack{path}` | yes | | `chat_link_preview` | `on\|off` | operator | `chat_link_preview_ack{enabled}` | yes | @@ -1189,7 +1190,8 @@ proves that the operator's key signed, not that the operator meant it: in a brow that key is driven by code the hub serves (T3), and in the desktop application by a renderer that parses content from nodes. Either could have shared any folder on the operator's machine, writable, from anywhere. What is left here narrows (`root_remove`, -`root_eject`) or restores what the operator already shared (`root_plug`). The +`root_eject`, `root_pause`) or restores what the operator already shared (`root_plug`, +`root_resume`). The operator still sees the table from any browser: it rides in `index_sync` and `index_delta`, which is also how a change made on the node's machine reaches every open page. @@ -1327,11 +1329,13 @@ folder just created, or one emptied, does not exist as far as the UI is concerne a member cannot tell "the drive is unplugged" from "it is all still there" — an unavailable root is listed, with its content frozen rather than hidden. -Each root is described as `{name, kind, available, writable, removable, ejected}` — and -never a path: a member is told what exists and whether it is readable, never where on -the operator's disk it lives. `roots` rides on `index_delta` as well as `index_sync`, +Each root is described as `{name, kind, available, writable, removable, ejected, +paused}` — and never a path: a member is told what exists and whether it is readable, +never where on the operator's disk it lives. `paused` is the operator having stopped +the root's indexing: its files are served as before, and nothing new is indexed until +it is resumed. `roots` rides on `index_delta` as well as `index_sync`, because a full index is only ever sent on request: without it, a root added, removed, -ejected or plugged would leave every connected client's directory table stale until +ejected, plugged or paused would leave every connected client's directory table stale until somebody reloaded the page, and the delta that tells them something changed would be the one message unable to say what. @@ -2138,6 +2142,7 @@ it back (§3.5). | `chat_link_preview` / `_ack` | C→N / N⇒C | signed | whether the node unfurls posted links | | `search_listed` / `_ack` | C→N / N⇒C | signed | whether members' cross-group Search lists this group | | `root_eject` / `_ack`, `root_plug` / `_ack` | C→N / N⇒C | signed | take a removable root offline, put it back | +| `root_pause` / `_ack`, `root_resume` / `_ack` | C→N / N⇒C | signed | stop a root's indexing, start it again from where it stopped (6.2) | | `apps_enabled` / `_ack` | C→N / N⇒C | signed | which group apps are shown | | `set_scan_settings` / `_ack` | C→N / N⇒C | signed | reconcile interval and debounce | | `tmdb_config` / `_ack` | C→N / N⇒C | signed | node-wide TMDB token and language | @@ -2182,7 +2187,7 @@ message: ## 13. Versioning and compatibility -MNP versions independently of the package version. Current: **`6.1`**; oldest peer +MNP versions independently of the package version. Current: **`6.2`**; oldest peer accepted: **`4.0`**. 4.0 is the floor: a member presents a short-lived node-audience token bound to one node @@ -2208,6 +2213,10 @@ script in its page either. 6.1 adds the signed `chat_purge` (§10.4), additively: a 6.0 node gives no answer, as for any unknown type, and nothing else changes. +6.2 adds the signed `root_pause` and `root_resume` (§10.4) and `paused` on each root of +the roots table, additively: a 6.1 node gives no answer to either message, and a client +that ignores `paused` shows a paused root as an ordinary one. + The two numbers are separate on purpose. `MNP_VERSION` says what this build speaks; `MNP_MIN_SUPPORTED` says what it will talk to, and moving the second is a decision about whether an older peer can still do anything useful: diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md index 5649b30..42103d6 100644 --- a/docs/USERGUIDE.md +++ b/docs/USERGUIDE.md @@ -695,6 +695,13 @@ catalogued. Group → Settings has the re-check interval and how long the node waits after a file changes before indexing it. Members see indexing progress while it runs. +**Pausing.** Indexing a large drive takes hours. The pause button on the +indexing band, or **Pause indexing** beside the directory in Group → Settings, +stops it at the next file. The files already indexed stay visible to members. +Nothing else in that directory is indexed until you click **Resume indexing**, +not even a file a member uploads into it, and the pause survives a restart. +Resuming picks up where it stopped: files already indexed are not read again. + One thing that surprises people: **two identical files in one group are one entry**, because a file is identified by its content. A scan that reports ten files and indexes nine has not lost anything. @@ -1058,6 +1065,7 @@ meshbay-node reset erase all node state (destructive) meshbay-node group list|add|remove groups this node hosts meshbay-node root list|add|remove|set directories in a group meshbay-node root eject|plug removable drives +meshbay-node root pause|resume stop and restart a directory's indexing meshbay-node gek init|rotate the group's encryption key meshbay-node operator pair authorise a browser 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 ──────────────────────────────────────────────────────── |