diff options
45 files changed, 1931 insertions, 157 deletions
diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md index 0703993..f9ceaa2 100644 --- a/docs/MESHBAY_DESIGN.md +++ b/docs/MESHBAY_DESIGN.md @@ -1783,6 +1783,53 @@ root is therefore not watched, not swept by reconcile, and not hashed by any wal there; the sweep indexes only what the index lacks. The request returns once the walk is queued, and the progress band shows it. +**A disk that stops answering costs a file, then a root, never the scan.** A read +on a failing drive, a share that dropped or a file a sync client has not fetched +does not fail: it does not return. The hash ran on the indexer's one thread with +nothing to bound it, so one such file stopped its group's scan where it stood: +the roots behind it unread, the enrichment that follows a scan never started, a +pause unable to take effect since it is looked at between two files, and nothing +in the log naming the file, because a file is named there only when it fails. + +- **What is timed is silence, not work.** Every directory entry listed and every + megabyte read counts as an answer (`_beat`). A listing of a million entries and + a file on a slow link take as long as they take; neither goes sixty seconds + (`STALL_SECS`) without one. A call waiting its turn behind another is not + timed for the wait. +- **The thread is left behind, because nothing can interrupt it.** It stays in + the kernel for as long as the kernel keeps it, and a new one takes the calls + that were waiting (`_DiskThread.abandon`). They are daemon threads: the + interpreter joins a pool's threads on its way out, and a node with one read + that never returned could not stop. +- **A root's walk asks the disk nothing from the event loop.** Each file's stat + is taken with the listing, on the indexer's thread, and handed down to the + hash. Asked again on the loop before each hash, as it was, it is the loop + that waits on a silent disk, and with it every connection of the node. The + one file a watcher event names is still asked for there. +- **The file is listed as unread and not asked for again until it changes.** + Reconcile meets it at every pass and would wait a minute on it each time. A + file the system refuses outright (a permission, a lock) is listed with what the + system said and is asked again, since asking costs nothing. A restart, or the + root going away and coming back, forgets the list. +- **Three in a row, with none read in between, and the root is paused.** A minute + a file is weeks on a music library, with a thread left behind each time. The + root is paused by the node as the operator would pause it (above): the rest of + the walk comes from the hash cache, the watcher and reconcile leave it alone, + what is indexed stays served, the next root is walked, and Resume starts it + again and gets past the files already known to give no answer. A listing that + stops answering pauses its root at once, having no file to give up on; a rescan + that gets no listing puts back the entries it had dropped. +- **This pause is not persisted.** The roster holds the operator's pauses. A + restart tries the root again, which is what brings back one whose drive or + share has returned, at the cost of three minutes when it has not. +- **The operator is told, on their own machine only.** Ten seconds of silence is + logged and shown in the progress band with the file; the group's shared + directories list the files given up on and why, and where a root went silent. + These are paths, so they are on the loopback API (`index-status`) and in + nothing a member receives: a member sees the root paused. +- **One directory that cannot be listed costs that directory.** The walk used to + skip a refusal in silence and to end the whole root's scan on any other error. + ### 6.4 Uploads Five protections, and they are the substance: @@ -3705,7 +3752,7 @@ requirements, not compatibility notes. | **Case-insensitive, case-preserving** | The index needs a canonical identity and a **case-folding collision check** at scan time, reported to the operator rather than resolved silently. One directory indexed as two roots is the same problem one level up | | **Unicode normalization** | A name written on one platform in decomposed form and on another in composed form are different byte strings. **Normalize to NFC for identity and comparison; preserve the original bytes for display and for opening the file** | | **Reserved names and characters** | A group indexed on one platform can hold names another cannot create. The client sanitises on save **and says so**; the upload allowlist is the intersection across platforms, or some files are simply undownloadable | -| **Path length limits** | Use extended-length paths on Windows, in the node and the client alike | +| **Path length limits** | Windows refuses a path of 260 characters or more unless `LongPathsEnabled` is set, which it is not by default and which takes administrator rights to change. **The node hands every call that touches a group's disk the extended form** (`paths.long_path`, `\\?\`) and keeps the plain path in the index, the hash cache and the log: a directory is listed under its long form and its entries named from the plain one (`indexer.walk_files`, `wire._walk_dirs`). ffmpeg extends a long path itself and is given the plain one. `test_long_paths.py` holds indexing, serving, uploads, folders and enrichment with the setting off. What a client saves is not measured | | **Coarse timestamps, local time** | mtime alone is not a change detector. Size plus mtime with tolerance, and rehash when in doubt | | **Watcher reliability** | Change notification drops events under load on Windows, and inotify is unreliable on a FUSE mount. **Periodic reconciliation is mandatory on both platforms** | | **No symlinks, no POSIX permissions** | Simplifications: nothing to defend against, and the node runs as the user anyway | @@ -4461,6 +4508,8 @@ process runs it — `systemctl --user` on Linux, Task Scheduler on Windows. | Item | Status | |---|---| +| **A disk that stops answering still stalls what is not the scan** | The indexer's own reads are bounded (§6.3). Serving a chunk, the availability check and the directory deletions go through the roots' I/O thread, which has no such limit, and the tag and photo readers leave a thread in the shared pool at each timeout | +| **A file that gave no answer is asked again at every start** | The list of unread files is memory. With three or more such files in a row the root is paused again three minutes into each start, until the operator resumes it or moves them | | **The desktop cast relay drops whole segments** | `cast-relay.js` drops a fragment once 8 MB wait for a receiver, and a fragment is a segment, 5 to 10 MB at a film's bitrate: the picture freezes for its length. Its backlog is bounded in fragments only. The Android relay spools a receiver's lead to disk and bounds its backlog in bytes (§11.4); the desktop has neither | | **The desktop cast relay takes the first chunk for the whole header** | The node's first chunk can be the ftyp alone, the moov in the next; the receiver then gives up. Fixed in the Android relay (§11.4), not in `cast-relay.js` | | **A cast restart may pull far ahead** | Seen once on an emulator with a synthetic film: after the restart's reinit the node reported `duration=None`, and the page pulled most of the film at network speed. Not reproduced on a real film; the suspicion is a read-ahead budget computed without a duration | diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md index 9d1749f..bab2074 100644 --- a/docs/USERGUIDE.md +++ b/docs/USERGUIDE.md @@ -702,6 +702,19 @@ 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. +**A disk that stops answering.** A failing drive, a network share that dropped +or a file a sync client has not downloaded can leave a read waiting for ever. +The node waits sixty seconds for an answer, then leaves that file and goes on. +After three such files in a row it takes the disk to be the problem and pauses +that directory by itself; the other directories go on being indexed, and what +was already indexed stays visible. On the node's own machine, the indexing band +names the file once the disk has been silent for ten seconds, and Group → +Settings shows **disk not answering** on the directory, where it stopped, and +the list **Files that could not be read** with the reason for each. Check the +disk, or move those files out, then click **Resume indexing**: the node gets +past the files it already knows give no answer, and tries them again when they +change or when it restarts. + 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. @@ -1037,6 +1050,12 @@ Choose a download folder in Settings → Downloads, which switches the browser t writing straight to disk, and retry. The desktop application does not have this class of problem. +**Indexing stops on one directory, or some of its files never appear.** +Group → Settings, on the node's own machine: a directory marked **disk not +answering** was paused by the node, and **Files that could not be read** lists +what it gave up on and why. See Indexing in section 6. The node's log names +the same files (`No answer from the disk`, `Cannot index`). + **Nothing appears in Videos, Music or Photos.** No folders have been chosen for that application yet. Group → Settings. The tab says so rather than just looking empty, but it is easy to miss. diff --git a/docs/WINDOWS-PORT.md b/docs/WINDOWS-PORT.md index 840f22a..cd0c89b 100644 --- a/docs/WINDOWS-PORT.md +++ b/docs/WINDOWS-PORT.md @@ -729,8 +729,14 @@ real bug found while testing W4, not from the original plan. verify `safeStorage`/DPAPI works, verify Task Scheduler autostart. - **Electron:** visual check of the UI, native save dialog, node management panel (once W3 is done). -- **Long paths:** a library with a path exceeding 260 characters on NTFS - with long paths enabled. +- **Long paths:** done for the node (2026-10-14), with `LongPathsEnabled` + off, the default: `test_long_paths.py` indexes, serves, deletes, uploads + into, lists and enriches files whose path is past 260 characters, and + fails without the change. Before it, the walk could not list such a + directory and its files never reached the members. Checked by hand in the + installed app the same day, files at 312 and 380 characters: indexed, + listed, streamed, downloaded byte for byte, uploaded into, deleted; tags, + the cover beside the track and the video and photo thumbnails all there. - **Case collisions:** a group indexed on ext4 containing `Film.mkv` and `film.mkv`, opened from a Windows client. diff --git a/docs/generate_http_api.py b/docs/generate_http_api.py index d064519..cd99332 100644 --- a/docs/generate_http_api.py +++ b/docs/generate_http_api.py @@ -160,5 +160,6 @@ def render() -> str: if __name__ == "__main__": - OUT.write_text(render(), encoding="utf-8") + # Not `write_text`: on Windows it turns every line ending into CRLF. + OUT.write_bytes(render().encode("utf-8")) print(f"wrote {OUT}", file=sys.stderr) diff --git a/packages/meshbay-common/src/meshbay_common/paths.py b/packages/meshbay-common/src/meshbay_common/paths.py index 8be4680..059bf09 100644 --- a/packages/meshbay-common/src/meshbay_common/paths.py +++ b/packages/meshbay-common/src/meshbay_common/paths.py @@ -19,7 +19,7 @@ from __future__ import annotations import os import re import unicodedata -from pathlib import Path, PurePath +from pathlib import Path # Windows refuses these as a basename, with or without an extension: `AUX.txt` # is as impossible as `AUX`. A group indexed on Linux can hold them, and a @@ -39,6 +39,9 @@ _RESERVED_CHARS = set('<>:"/\\|?*') | {chr(c) for c in range(32)} | BIDI_CONTROL # Windows without long-path support. A deep media library reaches this. MAX_PATH_WINDOWS = 260 +# Creating a directory is refused twelve characters earlier (room for an 8.3 +# name), and listing one adds `\*`: the extended form starts there. +_EXTENDED_FROM = MAX_PATH_WINDOWS - 12 def nfc(text: str) -> str: @@ -102,19 +105,22 @@ def long_path(path: Path | str) -> str: """ A path string safe to hand to the OS. - On Windows, prefix with `\\\\?\\` so `MAX_PATH` does not truncate a deep - library. The prefix requires an absolute, already-normalized path, and it is - a no-op everywhere else. + On Windows, a path that reaches `MAX_PATH` is refused by every call unless + the system's `LongPathsEnabled` is set, which it is not by default and which + a program cannot set for itself. The extended form (`\\\\?\\`) is accepted + whatever the setting, so a long path gets it. The form skips the system's + normalisation, so the path is normalised here first. A no-op everywhere + else, and for a path short enough to need nothing. """ text = str(path) if os.name != "nt" or text.startswith("\\\\?\\"): return text - resolved = str(PurePath(text)) - if len(resolved) < MAX_PATH_WINDOWS and not text.startswith("\\\\"): + full = os.path.normpath(text) if os.path.isabs(text) else os.path.abspath(text) + if len(full) < _EXTENDED_FROM: return text - if text.startswith("\\\\"): # UNC: \\server\share → \\?\UNC\server\share - return "\\\\?\\UNC" + text[1:] - return "\\\\?\\" + text + if full.startswith("\\\\"): # UNC: \\server\share → \\?\UNC\server\share + return "\\\\?\\UNC" + full[1:] + return "\\\\?\\" + full def find_fold_collisions(names: list[str]) -> dict[str, list[str]]: diff --git a/packages/meshbay-common/tests/test_paths.py b/packages/meshbay-common/tests/test_paths.py index 012a32e..cbaf1f5 100644 --- a/packages/meshbay-common/tests/test_paths.py +++ b/packages/meshbay-common/tests/test_paths.py @@ -6,12 +6,15 @@ these are not compatibility niceties — they decide whether two index entries a one file, and whether a member can save what they downloaded. """ +import os + import pytest from meshbay_common.paths import ( find_fold_collisions, fold, fold_path, is_portable_name, + long_path, nfc, portable_name_problem, sanitize_for_download, @@ -126,3 +129,49 @@ def test_a_bidi_override_cannot_hide_an_extension(): disguised = "invoicefdp.exe" # displays as "invoiceexe.pdf" assert sanitize_for_download(disguised) == "invoice_fdp.exe" assert portable_name_problem(disguised) + + +# ── Long paths on Windows ──────────────────────────────────────────────────── + +on_windows = pytest.mark.skipif(os.name != "nt", reason="the extended form is Windows'") +BS = chr(92) +EXT = BS * 2 + "?" + BS +DEEP = BS.join(["C:", *["a folder name of some length"] * 9, "track.flac"]) + + +@on_windows +def test_a_short_path_is_left_as_it_is(): + assert long_path(f"C:{BS}Music{BS}track.flac") == f"C:{BS}Music{BS}track.flac" + + +@on_windows +def test_a_long_path_gets_the_extended_form(): + assert long_path(DEEP) == EXT + DEEP + + +@on_windows +def test_the_extended_form_is_normalised_first(): + """The system normalises nothing behind the prefix: a forward slash or a + `..` left in would name a file that does not exist.""" + messy = DEEP.replace(BS, "/").replace("/track.flac", "/x/../track.flac") + assert long_path(messy) == EXT + DEEP + + +@on_windows +def test_a_long_share_path_gets_the_unc_form(): + share = BS * 2 + "server" + BS + "share" + DEEP[2:] + assert long_path(share) == EXT + "UNC" + BS + share[2:] + + +@on_windows +def test_a_path_in_the_extended_form_is_left_alone(): + assert long_path(EXT + DEEP) == EXT + DEEP + + +@on_windows +def test_a_directory_gets_it_before_max_path(): + """Creating a directory is refused twelve characters before MAX_PATH.""" + at_248 = "C:" + BS + "d" * 245 + assert len(at_248) == 248 + assert long_path(at_248) == EXT + at_248 + assert long_path(at_248[:-1]) == at_248[:-1] 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 a666c29..8ed8eec 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js @@ -46,6 +46,9 @@ async function readLocalNode(node, groupId, attempts = NODE_CHECK_ATTEMPTS, // ── Shared Directories Table ──────────────────────────────────────────── +// How often the table asks its node what it could not read. +const TROUBLE_POLL_MS = 10000; + /** * A group's root directories, and the operator's controls over them. * @@ -98,6 +101,10 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, // reason nothing happened. const [msg, setMsg] = useState(null); const [indexProgress, setIndexProgress] = useState(null); + // What the node could not read and the folders it paused by itself, as its + // loopback API says them; `troubleTick` asks again after a change here. + const [trouble, setTrouble] = useState(null); + const [troubleTick, setTroubleTick] = useState(0); const say = useCallback((text) => setMsg(text ? { text, error: false } : null), []); const refuse = useCallback((text) => setMsg({ text, error: true }), []); @@ -161,6 +168,26 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, const canEdit = isLocal || overMnp || overLoopback; const canWiden = isLocal || onNodeMachine; + // Paths, so only where the loopback API answers: on the node's own machine. + // An operator elsewhere sees the folder paused, as any member does. + useEffect(() => { + if (!onNodeMachine) { setTrouble(null); return undefined; } + let stopped = false; + const read = async () => { + try { + const s = await platform.node.op('groupIndexStatus', { groupId }); + const next = { unread: s.unread || [], total: s.unread_total || 0, + stalled: s.stalled || [] }; + if (!stopped) { + setTrouble((prev) => JSON.stringify(prev) === JSON.stringify(next) ? prev : next); + } + } catch { /* busy, or mid-reload: the next poll asks again */ } + }; + read(); + const id = setInterval(read, TROUBLE_POLL_MS); + return () => { stopped = true; clearInterval(id); }; + }, [onNodeMachine, groupId, troubleTick]); + // Deliberately no index refresh after a root change. // // Adding a root makes the node reload, which rescans — minutes on a real @@ -182,7 +209,7 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, } catch (err) { refuse(platform.bridgeMessage(err)); return false; - } finally { setBusy(false); } + } finally { setBusy(false); setTroubleTick((n) => n + 1); } }, [onRootsChange, refuse]); const doUpdateRoot = useCallback(async (rootName, updates) => { @@ -296,6 +323,34 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, <p class="settings-hint" style="margin-top:8px"> ${t('settings_node.roots_local_only_hint')}</p>`; + // A folder the node paused because its disk stopped answering, and the + // files it gave up on. Under the table, with the paths: this is the list an + // operator used to need the node's log for. + const stalledAt = {}; + for (const s of (trouble && trouble.stalled) || []) { + if (displayRoots.some((r) => r.name === s.root && r.paused)) stalledAt[s.root] = s.path; + } + const troubleNote = (isLocal || !trouble) ? '' : html` + ${Object.entries(stalledAt).map(([name, file]) => html` + <p class="error-msg" role="status" style="margin-top:10px" key=${name}> + ${t('node.root_stalled_hint', { name, file })}</p>`)} + ${trouble.total > 0 && html` + <details class="unread-files"> + <summary>${t('node.unread_summary', { count: trouble.total })}</summary> + <ul> + ${trouble.unread.map((u) => html` + <li key=${u.path}> + <span class="unread-path">${u.path}</span> + <span class="unread-reason">${u.reason === 'no_answer' + ? t('node.unread_no_answer') : u.reason}</span> + </li>`)} + </ul> + ${trouble.total > trouble.unread.length && html` + <p class="settings-hint">${t('node.unread_more', + { count: trouble.total - trouble.unread.length })}</p>`} + </details>`} + `; + if (!displayRoots.length) { return html` <div class="shared-directories-table"> @@ -336,6 +391,9 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, <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>`} + ${!isLocal && stalledAt[r.name] && html` + <span class="node-root-badge node-root-badge-warn" + title=${stalledAt[r.name]}>${t('node.root_stalled')}</span>`} </td> ${hasPaths && html` <td class="sdt-col-path" title=${r.path || ''}>${r.path || ''}</td>`} @@ -390,6 +448,7 @@ function SharedDirectoriesTable({ roots, groupId, transport, signFn, `; })} </tbody> </table> + ${troubleNote} ${localOnlyHint} ${addControls} ${message} 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 ed81e7c..f0f9aa0 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 @@ -35,9 +35,18 @@ export function fromLoopback(g) { filesDone: g.files_done || 0, filesTotal: g.files_total || 0, queued: Array.isArray(g.queued) ? g.queued : [], + // What the disk has given no answer on, and for how long. A path, so the + // loopback answer only: a push carries counters and nothing else. + slowPath: g.slow_path || '', + slowSecs: g.slow_secs || 0, }; } +/** The last segment of a path, whichever way its separators lean. */ +export function baseName(path) { + return String(path || '').split(/[\\/]/).filter(Boolean).pop() || ''; +} + export function fromPush(groupId, s, roots) { const pos = Number.isInteger(s.root_pos) ? s.root_pos : -1; const known = pos >= 0 && Array.isArray(roots) && roots[pos]; @@ -52,6 +61,8 @@ export function fromPush(groupId, s, roots) { filesDone: s.files_done || 0, filesTotal: s.files_total || 0, queued: Number.isInteger(s.queued) ? s.queued : 0, + slowPath: '', + slowSecs: 0, }; } 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 47a5818..3a9a539 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/index-dock.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/index-dock.js @@ -7,6 +7,7 @@ import { useStickyBand } from './sticky.js'; import { Icon } from './icon.js'; import { fromLoopback, fromPush, mergeActivity, dockRows, hideRow, pauseRow, percent, queuedCount, + baseName, } from './index-dock-model.js'; // ── Indexing dock ─────────────────────────────────────────────────────────── @@ -117,9 +118,14 @@ function DockRow({ row, error, onOpen, onPause, onHide }) { : t(`indexdock.kind_${row.kind || 'scan'}`); const where = row.root ? `${row.groupName} › ${row.root}` : row.groupName; + // A scan that sits still says what it is sitting on. It used to show a bar + // that had stopped, and the name of a folder. + const slow = !done && !error && row.slowPath; let meta = ''; if (error) { meta = error; + } else if (slow) { + meta = t('indexdock.slow', { secs: number(row.slowSecs), file: baseName(row.slowPath) }); } else if (listing) { meta = t('indexdock.listing'); } else if (!done && !waiting) { @@ -144,7 +150,8 @@ function DockRow({ row, error, onOpen, onPause, 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 ${error ? 'index-dock-error' : ''}" + ${meta && html`<span class="index-dock-meta ${error || slow ? 'index-dock-error' : ''}" + title=${slow ? row.slowPath : null} role=${error ? 'alert' : null}>${meta}</span>`} </span> <span class="index-progress-bar index-dock-bar"> 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 afa1ba0..1fe2b36 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.open': "Einstellungen dieser Gruppe öffnen", 'indexdock.hide': "Bis zur nächsten Indizierung ausblenden", 'indexdock.pause': "Indexierung dieses Ordners pausieren", + 'indexdock.slow': "Seit {secs} s keine Antwort vom Datenträger: {file}", 'chat.load_older': '{n} ältere Nachrichten laden', 'chat.start_of_history': 'Anfang des Gesprächs', 'chat.today': 'Heute', @@ -1002,6 +1003,11 @@ export default { 'node.root_pause': "Indexierung pausieren", 'node.root_resume': "Indexierung fortsetzen", 'node.root_paused': "Indexierung pausiert", + 'node.root_stalled': "Datenträger antwortet nicht", + 'node.root_stalled_hint': "Der Node hat die Indexierung von {name} pausiert: Der Datenträger hat bei {file} nicht geantwortet. Prüfen Sie den Datenträger oder verschieben Sie diese Datei, und setzen Sie dann die Indexierung fort.", + 'node.unread_summary': "Dateien, die nicht gelesen werden konnten: {count}", + 'node.unread_no_answer': "keine Antwort vom Datenträger", + 'node.unread_more': "Und {count} weitere.", '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 b4346b9..642d3e2 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.open': "Open this group's settings", 'indexdock.hide': "Hide until the next indexing", 'indexdock.pause': "Pause indexing this folder", + 'indexdock.slow': "No answer from the disk for {secs} s: {file}", 'chat.load_older': 'Load {n} older messages', 'chat.start_of_history': 'Beginning of the conversation', 'chat.today': 'Today', @@ -1070,6 +1071,11 @@ export default { 'node.root_pause': "Pause indexing", 'node.root_resume': "Resume indexing", 'node.root_paused': "indexing paused", + 'node.root_stalled': "disk not answering", + 'node.root_stalled_hint': "The node paused indexing of {name}: its disk gave no answer at {file}. Check the disk, or move that file out, then resume indexing.", + 'node.unread_summary': "Files that could not be read: {count}", + 'node.unread_no_answer': "no answer from the disk", + 'node.unread_more': "And {count} more.", '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 9bb26e1..fe66af4 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.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", + 'indexdock.slow': "El disco no responde desde hace {secs} s: {file}", 'chat.load_older': 'Cargar {n} mensajes anteriores', 'chat.start_of_history': 'Inicio de la conversación', 'chat.today': 'Hoy', @@ -996,6 +997,11 @@ export default { 'node.root_pause': "Pausar la indexación", 'node.root_resume': "Reanudar la indexación", 'node.root_paused': "indexación en pausa", + 'node.root_stalled': "el disco no responde", + 'node.root_stalled_hint': "El nodo pausó la indexación de {name}: su disco no respondió en {file}. Revisa el disco, o saca ese archivo de la carpeta, y reanuda la indexación.", + 'node.unread_summary': "Archivos que no se pudieron leer: {count}", + 'node.unread_no_answer': "el disco no respondió", + 'node.unread_more': "Y {count} más.", '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 9f32a19..6296a8b 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.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", + 'indexdock.slow': "Pas de réponse du disque depuis {secs} s : {file}", 'chat.load_older': 'Charger {n} messages plus anciens', 'chat.start_of_history': 'Début de la conversation', 'chat.today': 'Aujourd’hui', @@ -1005,6 +1006,11 @@ export default { 'node.root_pause': "Mettre l'indexation en pause", 'node.root_resume': "Reprendre l'indexation", 'node.root_paused': "indexation en pause", + 'node.root_stalled': "le disque ne répond pas", + 'node.root_stalled_hint': "Le node a mis en pause l'indexation de {name} : son disque n'a pas répondu sur {file}. Vérifiez le disque, ou sortez ce fichier du dossier, puis reprenez l'indexation.", + 'node.unread_summary': "Fichiers qui n'ont pas pu être lus : {count}", + 'node.unread_no_answer': "pas de réponse du disque", + 'node.unread_more': "Et {count} autres.", '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 5578028..0751692 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.open': "Apri le impostazioni di questo gruppo", 'indexdock.hide': "Nascondi fino alla prossima indicizzazione", 'indexdock.pause': "Sospendi l'indicizzazione di questa cartella", + 'indexdock.slow': "Nessuna risposta dal disco da {secs} s: {file}", 'chat.load_older': 'Carica {n} messaggi precedenti', 'chat.start_of_history': 'Inizio della conversazione', 'chat.today': 'Oggi', @@ -1004,6 +1005,11 @@ export default { 'node.root_pause': "Sospendi l'indicizzazione", 'node.root_resume': "Riprendi l'indicizzazione", 'node.root_paused': "indicizzazione sospesa", + 'node.root_stalled': "il disco non risponde", + 'node.root_stalled_hint': "Il node ha sospeso l'indicizzazione di {name}: il disco non ha risposto su {file}. Controlla il disco, o sposta quel file altrove, poi riprendi l'indicizzazione.", + 'node.unread_summary': "File che non è stato possibile leggere: {count}", + 'node.unread_no_answer': "nessuna risposta dal disco", + 'node.unread_more': "E altri {count}.", '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 b5e091c..c3d4fa8 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.open': "このグループの設定を開く", 'indexdock.hide': "次のインデックスまで隠す", 'indexdock.pause': "このフォルダーのインデックス作成を一時停止", + 'indexdock.slow': "ディスクが {secs} 秒間応答していません: {file}", 'chat.load_older': '以前のメッセージを {n} 件読み込む', 'chat.start_of_history': '会話のはじまり', 'chat.today': '今日', @@ -990,6 +991,11 @@ export default { 'node.root_pause': "インデックス作成を一時停止", 'node.root_resume': "インデックス作成を再開", 'node.root_paused': "インデックス作成を一時停止中", + 'node.root_stalled': "ディスクが応答しません", + 'node.root_stalled_hint': "ノードが {name} のインデックス作成を一時停止しました。{file} でディスクが応答しませんでした。ディスクを確認するか、そのファイルを移動してから、インデックス作成を再開してください。", + 'node.unread_summary': "読み取れなかったファイル: {count}", + 'node.unread_no_answer': "ディスクからの応答なし", + 'node.unread_more': "ほか {count} 件。", '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 86e01a2..c02be99 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.open': "Instellingen van deze groep openen", 'indexdock.hide': "Verbergen tot de volgende indexering", 'indexdock.pause': "Indexeren van deze map pauzeren", + 'indexdock.slow': "Al {secs} s geen antwoord van de schijf: {file}", 'chat.load_older': '{n} oudere berichten laden', 'chat.start_of_history': 'Begin van het gesprek', 'chat.today': 'Vandaag', @@ -1006,6 +1007,11 @@ export default { 'node.root_pause': "Indexeren pauzeren", 'node.root_resume': "Indexeren hervatten", 'node.root_paused': "indexeren gepauzeerd", + 'node.root_stalled': "schijf antwoordt niet", + 'node.root_stalled_hint': "De node heeft het indexeren van {name} gepauzeerd: de schijf gaf geen antwoord bij {file}. Controleer de schijf, of verplaats dat bestand, en hervat daarna het indexeren.", + 'node.unread_summary': "Bestanden die niet gelezen konden worden: {count}", + 'node.unread_no_answer': "geen antwoord van de schijf", + 'node.unread_more': "En nog {count}.", '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 090be3f..97e8b53 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.open': "Otwórz ustawienia tej grupy", 'indexdock.hide': "Ukryj do następnego indeksowania", 'indexdock.pause': "Wstrzymaj indeksowanie tego folderu", + 'indexdock.slow': "Brak odpowiedzi dysku od {secs} s: {file}", 'chat.load_older': 'Wczytaj {n} starszych wiadomości', 'chat.start_of_history': 'Początek rozmowy', 'chat.today': 'Dziś', @@ -1030,6 +1031,11 @@ export default { 'node.root_pause': "Wstrzymaj indeksowanie", 'node.root_resume': "Wznów indeksowanie", 'node.root_paused': "indeksowanie wstrzymane", + 'node.root_stalled': "dysk nie odpowiada", + 'node.root_stalled_hint': "Węzeł wstrzymał indeksowanie {name}: dysk nie odpowiedział przy {file}. Sprawdź dysk albo przenieś ten plik, a potem wznów indeksowanie.", + 'node.unread_summary': "Pliki, których nie udało się odczytać: {count}", + 'node.unread_no_answer': "brak odpowiedzi dysku", + 'node.unread_more': "I jeszcze {count}.", '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 bfac600..8ff36a5 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.open': "Abrir as configurações deste grupo", 'indexdock.hide': "Ocultar até a próxima indexação", 'indexdock.pause': "Pausar a indexação desta pasta", + 'indexdock.slow': "Sem resposta do disco há {secs} s: {file}", 'chat.load_older': 'Carregar {n} mensagens anteriores', 'chat.start_of_history': 'Início da conversa', 'chat.today': 'Hoje', @@ -997,6 +998,11 @@ export default { 'node.root_pause': "Pausar a indexação", 'node.root_resume': "Retomar a indexação", 'node.root_paused': "indexação pausada", + 'node.root_stalled': "o disco não responde", + 'node.root_stalled_hint': "O node pausou a indexação de {name}: o disco não respondeu em {file}. Verifique o disco, ou tire esse arquivo da pasta, e retome a indexação.", + 'node.unread_summary': "Arquivos que não puderam ser lidos: {count}", + 'node.unread_no_answer': "sem resposta do disco", + 'node.unread_more': "E mais {count}.", '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 367efa0..d4ad64f 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.open': "打开此群组的设置", 'indexdock.hide': "隐藏，直到下一次索引", 'indexdock.pause': "暂停索引此文件夹", + 'indexdock.slow': "磁盘已 {secs} 秒没有响应：{file}", 'chat.load_older': '加载更早的 {n} 条消息', 'chat.start_of_history': '对话开头', 'chat.today': '今天', @@ -978,6 +979,11 @@ export default { 'node.root_pause': "暂停索引", 'node.root_resume': "继续索引", 'node.root_paused': "索引已暂停", + 'node.root_stalled': "磁盘没有响应", + 'node.root_stalled_hint': "节点已暂停索引 {name}：磁盘在 {file} 处没有响应。请检查磁盘，或将该文件移出，然后恢复索引。", + 'node.unread_summary': "无法读取的文件：{count}", + 'node.unread_no_answer': "磁盘没有响应", + 'node.unread_more': "还有 {count} 个。", '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 586f36d..a14bdc3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -3681,6 +3681,17 @@ h2 .gn-owner, h3 .gn-owner { font-size: 0.55em; } font-size: 0.8em; color: var(--text-dim); } +/* Files a scan could not read, with their paths: names somebody else chose. */ +.unread-files { margin-top: 10px; font-size: 0.9em; } +.unread-files summary { cursor: pointer; color: var(--warn, #d97706); } +.unread-files ul { + margin: 6px 0 0; + padding-left: 18px; + max-height: 240px; + overflow-y: auto; +} +.unread-files li { word-break: break-word; margin-bottom: 2px; } +.unread-files .unread-reason { color: var(--text-dim); margin-left: 6px; } .node-root-unavailable { opacity: 0.6; } .node-message { diff --git a/packages/meshbay-hub/tests/harness/photo_sync_probe.py b/packages/meshbay-hub/tests/harness/photo_sync_probe.py index 3fcce8a..122103a 100644 --- a/packages/meshbay-hub/tests/harness/photo_sync_probe.py +++ b/packages/meshbay-hub/tests/harness/photo_sync_probe.py @@ -143,9 +143,11 @@ window.meshbay = { <main class="main"><div id=root></div></main> <script type=module> import { html, render } from './vendor/htm-preact.js'; -import { initLocale } from './i18n.js'; +import { initLocale, setLocale } from './i18n.js'; import { startPhotoSync } from './photo-sync.js'; import { AndroidSyncPage } from './android-sync-page.js'; +// English, whatever the machine running this is set to. +setLocale('en'); await initLocale(); startPhotoSync({ hub: '', getUser: () => ({ username: 'bob', userId: 'u-bob', token: 't' }), diff --git a/packages/meshbay-hub/tests/test_indexing_dock.py b/packages/meshbay-hub/tests/test_indexing_dock.py index 1619b57..4510dfc 100644 --- a/packages/meshbay-hub/tests/test_indexing_dock.py +++ b/packages/meshbay-hub/tests/test_indexing_dock.py @@ -224,3 +224,61 @@ def test_the_dock_polls_the_node_wide_route_not_one_group(): source = DOCK.read_text(encoding="utf-8") assert "node.op('indexStatus')" in source assert "groupIndexStatus" not in source + + +# ── A disk that stops answering ───────────────────────────────────────────── + +@needs_node +def test_a_silent_disk_is_named_by_the_loopback_answer_only(tmp_path): + """A scan that sat still showed a bar that had stopped and a folder's name. + What it is waiting on is a path: the node's own machine, never a push.""" + paths = ["C:\\Music\\Some Album\\track 02.flac", "/srv/music/some album/track 02.flac", ""] + out = _run(tmp_path, f""" + const paths = {json.dumps(paths)}; + console.log(JSON.stringify({{ + local: m.fromLoopback({{ group_id: 'g', scanning: true, slow_path: paths[0], + slow_secs: 23 }}), + idle: m.fromLoopback({{ group_id: 'g', scanning: true }}), + push: m.fromPush('g', {{ scanning: true, slow_path: paths[0], slow_secs: 23 }}, []), + names: paths.map(m.baseName), + }})); + """) + assert (out["local"]["slowPath"], out["local"]["slowSecs"]) == (paths[0], 23) + assert (out["idle"]["slowPath"], out["idle"]["slowSecs"]) == ("", 0) + assert (out["push"]["slowPath"], out["push"]["slowSecs"]) == ("", 0), ( + "a member's connection must not be able to put a path on screen") + assert out["names"] == ["track 02.flac", "track 02.flac", ""], "either separator" + + +def test_the_row_says_what_the_disk_is_silent_on(): + dock = DOCK.read_text(encoding="utf-8") + row = dock[dock.index("function DockRow("):dock.index("export function IndexingDock(")] + assert ("t('indexdock.slow', { secs: number(row.slowSecs), " + "file: baseName(row.slowPath) })") in row + # Before the percentage: a bar that no longer moves is not the news. + assert row.index("else if (slow)") < row.index("else if (listing)") + assert "title=${slow ? row.slowPath : null}" in row, "the whole path, on hover" + + +def test_group_settings_list_what_the_node_could_not_read(): + """With paths, so asked of the loopback API and of nothing else: an operator + elsewhere sees the folder paused, as any member does.""" + settings = (STATIC / "group-settings.js").read_text(encoding="utf-8") + effect = settings[settings.index("if (!onNodeMachine) { setTrouble(null)"):] + effect = effect[:effect.index("}, [onNodeMachine, groupId, troubleTick]);")] + assert "platform.node.op('groupIndexStatus', { groupId })" in effect + assert "transport." not in effect + + note = settings[settings.index("const troubleNote ="):] + note = note[:note.index("if (!displayRoots.length)")] + assert "t('node.root_stalled_hint', { name, file })" in note + assert "t('node.unread_summary', { count: trouble.total })" in note + assert "u.reason === 'no_answer'" in note and "t('node.unread_no_answer')" in note + # Only for a folder that is still paused: resumed, the note has had its day. + assert "r.name === s.root && r.paused" in settings + assert "${troubleNote}" in settings[settings.index('<table class="shared-dirs-tbl">'):] + + css = (STATIC / "style.css").read_text(encoding="utf-8") + rule = css[css.index(".unread-files li {"):] + assert "word-break: break-word" in rule[:rule.index("}")], ( + "a path has no spaces to break at, and unpins the page on a phone") diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 022553e..557b523 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -39,7 +39,7 @@ from pathlib import Path import uvicorn from meshbay_common import MNP_VERSION from meshbay_common.background import spawn -from meshbay_common.paths import fold +from meshbay_common.paths import fold, long_path from meshbay_common.protocol import MNP from meshbay_node import uploads as uploads_mod @@ -1209,8 +1209,8 @@ class NodeDaemon(EnrichmentMixin): for path in uploads_mod.orphaned_parts( uploads_mod.find_parts(roots.roots), live, when): try: - size = path.stat().st_size - path.unlink() + size = os.stat(long_path(path)).st_size + os.unlink(long_path(path)) except OSError as exc: log.warning("Could not remove abandoned upload %s: %s", path.name, exc) diff --git a/packages/meshbay-node/src/meshbay_node/enrichment.py b/packages/meshbay-node/src/meshbay_node/enrichment.py index 19aafba..c09c5e5 100644 --- a/packages/meshbay-node/src/meshbay_node/enrichment.py +++ b/packages/meshbay-node/src/meshbay_node/enrichment.py @@ -2,8 +2,11 @@ thumbnail for Videos, tags and cover for Music, thumbnail and EXIF for Photos. NodeDaemon runs it; the enrichers themselves are in indexer/.""" +import os from dataclasses import replace +from meshbay_common.paths import long_path + from meshbay_node.indexer import DirectoryIndexer, GroupIndex from meshbay_node.roots import entry_abs_path @@ -67,7 +70,7 @@ class EnrichmentMixin: if not _under_any_directory(entry.path, video_dirs): continue file_path = entry_abs_path(indexer.roots, entry) - if not file_path or not file_path.exists(): + if not file_path or not os.path.exists(long_path(file_path)): continue self._enriched_attempted.add((indexer.group_id, entry.id)) @@ -153,7 +156,7 @@ class EnrichmentMixin: if owner is None: continue file_path = entry_abs_path(indexer.roots, entry) - if not file_path or not file_path.exists(): + if not file_path or not os.path.exists(long_path(file_path)): continue self._enriched_attempted.add((indexer.group_id, entry.id)) @@ -236,7 +239,7 @@ class EnrichmentMixin: if not _under_any_directory(entry.path, photo_dirs): continue file_path = entry_abs_path(indexer.roots, entry) - if not file_path or not file_path.exists(): + if not file_path or not os.path.exists(long_path(file_path)): continue self._enriched_attempted.add((indexer.group_id, entry.id)) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py index 9881298..53f1b27 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py @@ -13,14 +13,16 @@ the rest asynchronously and hands the result back via a callback. import asyncio import logging +import os from collections.abc import Awaitable, Callable from pathlib import Path import blake3 +from meshbay_common.paths import long_path from meshbay_common.protocol import IndexEntry from meshbay_node.indexer import title_parse -from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS +from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS, walk_files from meshbay_node.media_cache import MediaCache from meshbay_node.media_probe import probe_video @@ -102,7 +104,8 @@ def _title_from_siblings(file_path: Path) -> str | None: from another Special just like this one. """ try: - names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file()) + with os.scandir(long_path(file_path.parent)) as listing: + names = sorted(e.name for e in listing if e.is_file()) except OSError: return None checked = 0 @@ -142,8 +145,8 @@ def _synthetic_episode_number(file_path: Path, show_root: Path, season: int) -> """ try: candidates = sorted( - (p for p in show_root.rglob("*") - if p.is_file() and p.suffix.lower() in MEDIA_EXTENSIONS["video"]), + (p for p in walk_files(show_root) + if p.suffix.lower() in MEDIA_EXTENSIONS["video"]), key=lambda p: (str(p.parent), p.name), ) except OSError: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py index 397f760..ecc3b5e 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py @@ -60,11 +60,13 @@ image-file fallback for cover art, same as everything else. import asyncio import logging +import os import re from collections.abc import Awaitable, Callable from pathlib import Path import blake3 +from meshbay_common.paths import long_path from meshbay_common.protocol import IndexEntry from mutagen import File as MutagenFile from mutagen.asf import ASF @@ -155,8 +157,9 @@ _WMP_ALBUMART_GUID_RE = re.compile(r"^albumart_\{([0-9a-f-]{36})\}_(large|small) def _find_sibling_cover(folder: Path) -> Path | None: try: - candidates = [p for p in folder.iterdir() - if p.is_file() and p.suffix.lower() in _COVER_EXTS] + with os.scandir(long_path(folder)) as listing: + candidates = [folder / e.name for e in listing + if e.is_file() and Path(e.name).suffix.lower() in _COVER_EXTS] except OSError: return None if not candidates: @@ -240,12 +243,12 @@ def _read_format_specific_tags(path: Path) -> tuple[dict, float | None] | None: suffix = path.suffix.lower() if suffix == ".wma": try: - return _asf_tags_from_object(ASF(str(path))) + return _asf_tags_from_object(ASF(long_path(path))) except Exception: return {}, None if suffix == ".mpc": try: - return _musepack_tags_from_object(Musepack(str(path))) + return _musepack_tags_from_object(Musepack(long_path(path))) except Exception: return {}, None return None @@ -257,7 +260,7 @@ def _read_generic_tags_and_cover( tags: dict = {} duration: float | None = None try: - easy = MutagenFile(str(path), easy=True) + easy = MutagenFile(long_path(path), easy=True) except Exception: easy = None if easy is not None: @@ -277,7 +280,7 @@ def _read_generic_tags_and_cover( cover: bytes | None = None if not skip_cover: try: - raw = MutagenFile(str(path)) + raw = MutagenFile(long_path(path)) except Exception: raw = None if raw is not None: @@ -339,7 +342,8 @@ def _read_sibling_cover(path: Path) -> bytes | None: if sibling is None: return None try: - return sibling.read_bytes() + with open(long_path(sibling), "rb") as f: + return f.read() except OSError: return None diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py index 5cacef9..a9f85d0 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py @@ -26,6 +26,7 @@ from collections.abc import Awaitable, Callable from pathlib import Path import blake3 +from meshbay_common.paths import long_path from meshbay_common.protocol import IndexEntry from PIL import ExifTags, Image, ImageOps @@ -74,7 +75,7 @@ def _read_image(file_path: Path) -> tuple[bytes, int, int, int | None, str | Non *original* image's own dimensions — the thumbnail is a separate, resized copy, never what width/height describe. """ - with Image.open(file_path) as img: + with Image.open(long_path(file_path)) as img: taken_at = None camera = None try: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 4d68212..55ec56a 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -24,9 +24,12 @@ it is the only thing that recovers a missed event. import asyncio import logging +import os +import queue +import threading import time -from collections.abc import Awaitable, Callable -from concurrent.futures import ThreadPoolExecutor +from collections.abc import Awaitable, Callable, Iterable +from concurrent.futures import Executor, Future from dataclasses import dataclass from dataclasses import field as dataclass_field from pathlib import Path @@ -71,16 +74,171 @@ def _detect_type(path: Path) -> str: return "other" -def _is_indexable(path: Path) -> bool: - if not path.is_file(): - return False - name = path.name +def _is_indexable_name(name: str) -> bool: return ( not any(name.startswith(p) for p in EXCLUDED_PREFIXES) and not any(name.endswith(s) for s in EXCLUDED_SUFFIXES) ) +def _is_indexable(path: Path) -> bool: + return os.path.isfile(long_path(path)) and _is_indexable_name(path.name) + + +# ── A disk that stops answering ─────────────────────────────────────────────── +# +# A read on a failing drive, a share that dropped or a file a sync client has +# not fetched does not fail: it does not return. Nothing here can interrupt a +# thread blocked in the kernel, so the indexer measures instead how long the +# disk has said nothing, and past a limit leaves that thread where it is and +# goes on with another (`DirectoryIndexer._on_disk`). + +_serving = threading.local() + + +def _beat(where: Path | str | None = None) -> None: + """ + The disk answered: called by a blocking routine for each directory entry + listed and each piece of a file read, on the thread doing it. + + What is timed is silence, not the call. A root of a million files and a + file on a slow link both take as long as they take; neither goes a minute + without a single entry or a single megabyte. + """ + call = getattr(_serving, "call", None) + if call is not None: + call.answered = time.monotonic() + if where is not None: + call.where = str(where) + + +def _failed(reason: str) -> None: + """Why the routine running on this thread gave up on its file.""" + call = getattr(_serving, "call", None) + if call is not None: + call.failure = reason + + +class _DiskCall: + """One blocking routine handed to the disk thread.""" + + def __init__(self, fn, args) -> None: + self.fn = fn + self.args = args + self.future: Future = Future() + # When the disk last answered it: None until a thread takes it, so a + # call waiting behind another is not timed for the wait. + self.answered: float | None = None + self.where = "" + self.failure = "" + + def silent_for(self) -> float: + answered = self.answered + return 0.0 if answered is None else time.monotonic() - answered + + +class _DiskThread(Executor): + """ + The one thread an indexer reads the disk on, which can be left behind. + + One, deliberately: hashing two files at once buys nothing and can cost a + lot. Measured on a spinning USB drive, two interleaved multi-GB reads + seek-thrash against each other rather than go faster. It only takes the + blocking read and hash off the asyncio loop. + + `abandon()` is what a ThreadPoolExecutor has no word for: the thread stays + in the call it is stuck in, for as long as the kernel keeps it there, and a + new one takes the calls that were waiting behind. The threads are daemons + for the same reason, since the interpreter joins a pool's threads on its + way out and a node with one read that never returns could not stop. + """ + + def __init__(self, name: str = "indexer") -> None: + self._name = name + self._calls: queue.SimpleQueue = queue.SimpleQueue() + self._closed = False + self.left_behind = 0 + self._start() + + def _start(self) -> None: + threading.Thread(target=self._serve, args=(self._calls,), daemon=True, + name=f"{self._name}_{self.left_behind}").start() + + @staticmethod + def _serve(calls: queue.SimpleQueue) -> None: + while (call := calls.get()) is not None: + if not call.future.set_running_or_notify_cancel(): + continue + _serving.call = call + call.answered = time.monotonic() + try: + result = call.fn(*call.args) + except BaseException as e: + call.future.set_exception(e) + else: + call.future.set_result(result) + finally: + _serving.call = None + + def call(self, fn, /, *args) -> _DiskCall: + if self._closed: + raise RuntimeError("cannot schedule new futures after shutdown") + call = _DiskCall(fn, args) + self._calls.put(call) + return call + + def submit(self, fn, /, *args, **kwargs) -> Future: + return self.call(fn, *args).future + + def abandon(self) -> None: + """Leave the thread to the call it is in; a new one takes the rest.""" + stuck, self._calls = self._calls, queue.SimpleQueue() + try: + while True: + self._calls.put(stuck.get_nowait()) + except queue.Empty: + pass + stuck.put(None) # so it ends, if the call it is in ever does + self.left_behind += 1 + self._start() + + def shutdown(self, wait: bool = True, *, cancel_futures: bool = False) -> None: + self._closed = True + self._calls.put(None) + + +class DiskSilent(Exception): + """The disk gave no answer for `DirectoryIndexer.STALL_SECS`.""" + + def __init__(self, where: str) -> None: + super().__init__(where) + self.where = where + + +@dataclass +class UnreadFile: + """A file a scan met and could not index.""" + path: str + root: str + # NO_ANSWER, or what the system said when it refused. + reason: str + size: int + mtime: float + + +NO_ANSWER = "no_answer" + + +def _stop_observer(observer) -> None: + """Blocking, and on a disk that answers nothing possibly for good: run on + a thread of its own.""" + try: + observer.stop() + observer.join() + except Exception: + log.exception("Stopping a file watcher failed") + + # Found live: a 1256-byte ".mp3" with no audio stream at all, just an ID3 # tag — a truncated/corrupted rip, sitting between two good tracks of the # same album (docs/MESHBAY_DESIGN.md §9.8). A source this small claiming to @@ -95,7 +253,11 @@ def _is_indexable_size(path: Path, size: int) -> bool: return not (_detect_type(path) == "audio" and size < MIN_AUDIO_SIZE_BYTES) -_HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks +# What one read asks for. Each is one answer from the disk (`_beat`), so this is +# also the least a file must give per `STALL_SECS` to count as answering: at +# 1 MB that is 17 KB/s, below any link worth serving from, where 8 MB asked +# for 140 KB/s and would have given up on a slow share that was still reading. +_HASH_CHUNK = 1024 * 1024 # On a spinning disk the cost of a sample is its three seeks, not its bytes: # measured cold on USB, 45 MB took 516 ms a file, 9 MB 94 ms, 3 MB 79 ms. The @@ -111,6 +273,7 @@ def _feed(hasher, f, nbytes: int) -> None: remaining = nbytes while remaining > 0: chunk = f.read(min(_HASH_CHUNK, remaining)) + _beat() if not chunk: break hasher.update(chunk) @@ -121,6 +284,7 @@ def _partial_hash(file_path: Path, size: int) -> str: hasher = blake3.blake3() hasher.update(size.to_bytes(8, "little")) with open(long_path(file_path), "rb") as f: + _beat() _feed(hasher, f, _PARTIAL_HEAD) f.seek(size - _PARTIAL_TAIL) _feed(hasher, f, _PARTIAL_TAIL) @@ -184,10 +348,57 @@ def _walk_root(root: Root) -> list[Path]: every other thing the daemon is doing — WebRTC sessions, chat, the admin UI — for as long as it takes. """ - return [p for p in root.path.rglob("*") if p.is_file()] + return walk_files(root.path) + + +def walk_files(top: Path) -> list[Path]: + """ + Every file below `top`, links to files included, links to directories not + entered. Raises if `top` itself cannot be listed. Blocking. + + Each directory is listed under its long form and the paths returned are + built from the plain one, so a directory past Windows' MAX_PATH is listed + and everything downstream still sees, stores and compares plain paths. + """ + files: list[Path] = [] + pending = [top] + while pending: + directory = pending.pop() + _beat(directory) + try: + with os.scandir(long_path(directory)) as listing: + below: list[Path] = [] + for entry in listing: + _beat() + # The same two questions `rglob("*")` asked, which this + # replaces: a directory is entered unless it is a link, and + # a file counts whether or not it is one. + try: + if entry.is_dir(follow_symlinks=False): + below.append(directory / entry.name) + elif entry.is_file(): + files.append(directory / entry.name) + except OSError: + continue + except OSError as e: + if directory == top: + raise + # One directory that cannot be listed costs that directory. It + # used to be silent for a refusal and to end the whole walk for + # anything else. Said once: reconcile walks the root again every + # few minutes, and a drive's own system folders refuse every time. + if len(_unlistable) < 1000 and str(directory) not in _unlistable: + _unlistable.add(str(directory)) + log.warning("Cannot list %s: %s", directory, e.strerror or e) + continue + pending.extend(reversed(below)) + return files + +_unlistable: set[str] = set() -def _size_files(files: list[Path]) -> list[tuple[Path, int]]: + +def _size_files(files: list[Path]) -> list[tuple[Path, os.stat_result]]: """ Blocking: stat() every file from an already-completed walk — run via an executor for the same reason `_walk_root` is (its own docstring above). @@ -197,11 +408,17 @@ def _size_files(files: list[Path]) -> list[tuple[Path, int]]: the admin UI included, for as long as the stat() calls took — and did so *before* `_scan_root` had even set `progress.scanning`, so a consumer polling it saw "not scanning" the whole time real, blocking work was happening. + + The whole stat is returned and handed down to `_hash_or_cached`, which + needs the size and the date for the hash cache: asking again there would be + a disk call on the event loop for every file, and on a disk that stops + answering it is the loop, and with it the node, that would wait. """ - sized: list[tuple[Path, int]] = [] + sized: list[tuple[Path, os.stat_result]] = [] for p in files: + _beat(p.parent) try: - sized.append((p, p.stat().st_size)) + sized.append((p, os.stat(long_path(p)))) except OSError: continue return sized @@ -215,7 +432,7 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: if not _is_indexable(file_path): return None try: - stat = file_path.stat() + stat = os.stat(long_path(file_path)) if not _is_indexable_size(file_path, stat.st_size): return None if stat.st_size > _PARTIAL_THRESHOLD: @@ -224,7 +441,9 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: else: hasher = blake3.blake3() with open(long_path(file_path), "rb") as f: + _beat() while chunk := f.read(_HASH_CHUNK): + _beat() hasher.update(chunk) hex_hash = hasher.hexdigest() hv = 1 @@ -239,6 +458,7 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: ) except (OSError, PermissionError, ValueError) as e: log.warning("Cannot index %s: %s", file_path, e) + _failed(getattr(e, "strerror", None) or str(e)) return None @@ -273,6 +493,22 @@ class DirectoryIndexer: # it — several writes to the same file in quick succession (a slow copy # in several passes) collapse into one hash instead of one per write. DEFAULT_DEBOUNCE_SECS = 2.0 + # How long the disk may say nothing — not one directory entry, not one + # megabyte of a file — before the indexer stops waiting for it. A healthy + # read answers in milliseconds and a drive woken from sleep in fifteen + # seconds or so; nothing that is going to answer takes a minute. + STALL_SECS = 60.0 + # From here the silence is logged and shown to the operator, with the + # path: the one thing they could not find out while a scan sat still. + SLOW_SECS = 10.0 + DISK_POLL_SECS = 1.0 + # This many files in a row that gave no answer, with none read in between, + # and the fault is the root's and no longer a file's: it is paused. Going + # on would cost a minute a file, a thread left behind each time, and weeks + # on a music library. + STALL_LIMIT = 3 + # Files remembered as unread, for the operator's list. + UNREAD_KEPT = 500 def __init__( self, @@ -309,15 +545,21 @@ class DirectoryIndexer: self._index = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek) self._index.roots = roots.describe() - # One worker, deliberately, not a real pool: hashing two files at once - # buys nothing here and can cost a lot. It only offloads the blocking - # read+hash off the asyncio loop; it was never used for concurrency — - # every call site awaits one run_in_executor before starting the next - # (see _hash_or_cached below) — and measured on a spinning USB drive, - # two interleaved multi-GB reads would seek-thrash against each other - # rather than go faster. Left at 1 so the number does not promise a - # concurrency this code never provided. - self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="indexer") + # One thread, see `_DiskThread`. Every blocking call goes to it through + # `_on_disk`, which is where a disk that stops answering is noticed. + self._executor = _DiskThread("indexer") + self._disk_calls: dict[_DiskCall, str] = {} + # Files met and not indexed, by path. One that gave no answer is not + # read again until it changes: reconcile meets it at every pass, and + # would otherwise wait a minute on it each time. + self.unread: dict[str, UnreadFile] = {} + self._unread_dropped = 0 + # How many files in a row gave no answer, by folded root name. + self._silent_run: dict[str, int] = {} + # Roots this indexer paused by itself, and where the disk went silent. + # Not persisted: a restart tries the root again, which is also what + # brings back one whose share or drive has returned. + self.stalled: dict[str, str] = {} self._observer: Observer | None = None self._loop: asyncio.AbstractEventLoop | None = None self._reconciler: asyncio.Task | None = None @@ -385,6 +627,97 @@ class DirectoryIndexer: best = root return best + # ── A disk that stops answering ─────────────────────────────────────────── + + async def _on_disk(self, fn, *args, what: Path | str) -> _DiskCall: + """ + Run one blocking disk routine on the indexer's thread and wait for it, + for as long as the disk keeps answering. + + Raises DiskSilent once it has said nothing for `STALL_SECS`. The thread + is then left in that call, and the calls behind it go to a new one. + The call is returned rather than its result so the caller can read why + a routine gave up on a file (`failure`). + """ + call = self._executor.call(fn, *args) + waiting = asyncio.wrap_future(call.future) + self._disk_calls[call] = str(what) + said = False + try: + while True: + done, _ = await asyncio.wait({waiting}, timeout=self.DISK_POLL_SECS) + if done: + waiting.result() + return call + silent = call.silent_for() + if silent >= self.STALL_SECS: + where = call.where or str(what) + log.warning("No answer from the disk for %d s: giving up on %s", + silent, where) + self._executor.abandon() + raise DiskSilent(where) + if silent >= self.SLOW_SECS and not said: + said = True + log.warning("No answer from the disk for %d s, still waiting on %s", + silent, call.where or what) + finally: + self._disk_calls.pop(call, None) + if not waiting.done(): + waiting.cancel() + + def disk_wait(self) -> tuple[str, float] | None: + """What the disk has been silent on for `SLOW_SECS` or more, and for + how long. For the operator's own screen: it is a path.""" + for call, what in self._disk_calls.items(): + silent = call.silent_for() + if silent >= self.SLOW_SECS: + return call.where or what, silent + return None + + def _note_unread(self, root: Root, file_path: Path, st, reason: str) -> None: + key = str(file_path) + if key not in self.unread and len(self.unread) >= self.UNREAD_KEPT: + self._unread_dropped += 1 + return + self.unread[key] = UnreadFile(path=key, root=root.name, reason=reason, + size=st.st_size, mtime=st.st_mtime) + + def unread_count(self) -> int: + return len(self.unread) + self._unread_dropped + + def _forget_root_trouble(self, root: Root) -> None: + """A root resumed, replugged or removed starts again with a clean slate.""" + self.stalled.pop(root.name, None) + self._silent_run.pop(root.folded, None) + + async def _stall_root(self, root: Root, where: str) -> None: + """ + Pause a root whose disk stopped answering, and say so. + + The operator's own pause, set by the node: the walk under way lists + the rest from the hash cache, the watcher and reconcile leave the root + alone, what is indexed stays served, and Resume starts it again. + """ + live = self._live(root) + if live is None or live.paused: + return + live.paused = True + self.stalled[live.name] = where + self._silent_run.pop(live.folded, None) + self.progress.queued[:] = [n for n in self.progress.queued + if fold(n) != live.folded] + log.warning("Root %r paused by the node: its disk gave no answer (%s). " + "Nothing more is indexed on it until it is resumed", + live.name, where) + # Not `_restart_observer`: it joins the old watcher on this thread, and + # a watcher on the disk that just stopped answering may never let go. + old, self._observer = self._observer, None + if old is not None: + self._start_observer() + threading.Thread(target=_stop_observer, args=(old,), daemon=True, + name="indexer_unwatch").start() + await self.publish_roots() + # ── Initial scan ────────────────────────────────────────────────────────── async def initial_scan(self) -> None: @@ -431,7 +764,6 @@ class DirectoryIndexer: """ 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 @@ -449,23 +781,29 @@ class DirectoryIndexer: self._begin_job(root.name, root=root, kind=kind) try: try: - files = await loop.run_in_executor(self._executor, _walk_root, root) + files = (await self._on_disk(_walk_root, root, what=root.path) + ).future.result() + # Sizes up front, off the same listing that already walked the + # tree — the progress bar's denominator, not a second pass over + # the disk. Off-loop for the same reason the walk itself is + # (_size_files's own docstring: this used to be a synchronous + # loop right here, blocking the whole daemon for a large root). + sized = (await self._on_disk(_size_files, files, what=root.path) + ).future.result() + except DiskSilent as e: + # No list, so nothing to walk and nothing to list from the + # cache either (`_rescan_root` puts back what it had dropped). + await self._stall_root(root, e.where) + return 0 except OSError as e: log.warning("Cannot scan root %r: %s", root.name, e) return 0 - # Sizes up front, off the same listing that already walked the - # tree — the progress bar's denominator, not a second pass over - # the disk. Off-loop for the same reason the walk itself is - # (_size_files's own docstring: this used to be a synchronous - # loop right here, blocking the whole daemon for a large root). - sized = await loop.run_in_executor(self._executor, _size_files, files) - if not quiet: - self.progress.total_bytes = sum(size for _, size in sized) + self.progress.total_bytes = sum(st.st_size for _, st in sized) self.progress.files_total = len(sized) self.progress.current_dir = "" - for done, (file_path, size) in enumerate(sized): + for done, (file_path, st) 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: " @@ -480,10 +818,10 @@ class DirectoryIndexer: 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)) + entry = await (self._cached(root, file_path, st) if quiet + else self._hash_or_cached(root, file_path, st=st)) if not quiet: - self.progress.scanned_bytes += size + self.progress.scanned_bytes += st.st_size self.progress.files_done += 1 if entry: self._index.add_entry(entry) @@ -551,13 +889,15 @@ class DirectoryIndexer: if name in self.progress.queued: self.progress.queued.remove(name) - async def _cached(self, root: Root, file_path: Path) -> IndexEntry | None: + async def _cached(self, root: Root, file_path: Path, + st: os.stat_result | None = None) -> 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) + return await self._hash_or_cached(root, file_path, read=False, st=st) async def _hash_or_cached(self, root: Root, file_path: Path, *, - read: bool = True) -> IndexEntry | None: + read: bool = True, + st: os.stat_result | None = None) -> 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 @@ -566,12 +906,19 @@ class DirectoryIndexer: Whatever hash_version the cache holds is kept: TMDB matches, manual corrections, thumbnails and members' playlists are keyed by the id, so a file the node already knows never changes id because the scheme did. + + `st` is the file's stat when a walk already has it (`_size_files`), so + that a root's scan asks the disk nothing from the event loop. Without + it, for one file a watcher event names, it is asked for here. """ - if not _is_indexable(file_path): - return None - try: - st = file_path.stat() - except OSError: + if st is None: + if not _is_indexable(file_path): + return None + try: + st = os.stat(long_path(file_path)) + except OSError: + return None + elif not _is_indexable_name(file_path.name): return None if not _is_indexable_size(file_path, st.st_size): return None @@ -592,8 +939,26 @@ class DirectoryIndexer: if not read: return None - loop = asyncio.get_event_loop() - entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) + known = self.unread.get(str(file_path)) + if (known is not None and known.reason == NO_ANSWER + and (known.size, known.mtime) == (st.st_size, st.st_mtime)): + return None + try: + call = await self._on_disk(_scan_file, root, file_path, what=file_path) + except DiskSilent: + self._note_unread(root, file_path, st, NO_ANSWER) + run = self._silent_run.get(root.folded, 0) + 1 + self._silent_run[root.folded] = run + if run >= self.STALL_LIMIT: + await self._stall_root(root, str(file_path)) + return None + # The disk answered, whatever it said. + self._silent_run.pop(root.folded, None) + entry = call.future.result() + if entry is None and call.failure: + self._note_unread(root, file_path, st, call.failure) + elif entry is not None: + self.unread.pop(str(file_path), None) if entry and self._cache is not None: await self._cache.put(str(file_path), st.st_size, st.st_mtime, entry.id, entry.type, entry.added_at, @@ -635,7 +1000,7 @@ class DirectoryIndexer: if self._cache is None or not user_id: return try: - st = file_path.stat() + st = os.stat(long_path(file_path)) except OSError: # Gone between the rename and here. Nothing to attribute, and # nothing for anyone to delete either. @@ -742,8 +1107,17 @@ class DirectoryIndexer: root.name, len(dropped)) for entry in dropped: self._index.remove_entry(entry.id) + self._forget_root_trouble(root) + for path in [k for k, u in self.unread.items() + if fold(u.root) == root.folded]: + del self.unread[path] self.roots = roots + # The new set is built from the roster, which knows the operator's + # pauses and not this indexer's own. + for root in roots: + if root.name in self.stalled: + root.paused = True await off_disk(roots, roots.refresh_availability) added = [r for r in roots if r.folded not in old_names and r.available] @@ -942,14 +1316,21 @@ class DirectoryIndexer: entry it met and lose what the enrichment passes had put there. """ changed = False - loop = asyncio.get_event_loop() + unread = [path for path, u in self.unread.items() if fold(u.root) == root.folded] try: - on_disk, known = await loop.run_in_executor( - self._executor, self._sweep_scan_root, root) + on_disk, known, gone = (await self._on_disk( + self._sweep_scan_root, root, unread, what=root.path)).future.result() + except DiskSilent as e: + await self._stall_root(root, e.where) + return False except OSError as e: log.warning("Cannot reconcile root %r: %s", root.name, e) return False + # A file that could not be read and has since gone is no longer one. + for path in gone: + self.unread.pop(path, None) + 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 @@ -965,24 +1346,24 @@ class DirectoryIndexer: if not added_paths: return changed - 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)) + try: + added_sized = (await self._on_disk( + _size_files, list(added_paths), what=root.path)).future.result() + except DiskSilent as e: + await self._stall_root(root, e.where) + return changed - self._begin_job("", sum(size for _, size in added_sized), root=root, + self._begin_job("", sum(st.st_size for _, st in added_sized), root=root, kind=kind, files_total=len(added_sized)) try: - for done, (added, size) in enumerate(added_sized): + for done, (added, st) 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 + entry = await self._hash_or_cached(root, added, st=st) + self.progress.scanned_bytes += st.st_size self.progress.files_done += 1 if not entry: continue @@ -1044,7 +1425,18 @@ class DirectoryIndexer: """ carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)} self._drop_root_entries(root) + # It went away and came back: what it refused before is asked again. + self._forget_root_trouble(root) + for path in [k for k, u in self.unread.items() if fold(u.root) == root.folded]: + del self.unread[path] count = await self._scan_root(root, kind="rescan") + if not count and root.name in self.stalled: + # The disk went silent before anything was read: nothing was + # learnt about the root, so nothing it held is forgotten. + for old in carried.values(): + self._index.add_entry(old) + self.rescanned_ids.discard(old.id) + return 0 for entry in self._entries_under(root): old = carried.get((entry.id, entry.name, entry.path)) if old is None: @@ -1093,20 +1485,37 @@ class DirectoryIndexer: except OSError: return None - def _sweep_scan_root(self, root: Root) -> tuple[set[Path], dict[Path, str]]: + def _sweep_scan_root(self, root: Root, unread: Iterable[str] = (), + ) -> tuple[set[Path], dict[Path, str], list[str]]: """ Blocking: the two disk-touching pieces of one reconcile pass for a root, bundled so both run together in the executor rather than in the asyncio loop — the tree walk, and resolving the real path of every entry already known under this root (one syscall each). + + Also which of the `unread` paths are no longer on the disk, since + telling takes the same resolution. """ - on_disk = {p.resolve() for p in root.path.rglob("*") if _is_indexable(p)} + on_disk = set() + for p in _walk_root(root): + if _is_indexable_name(p.name): + _beat(p.parent) + on_disk.add(p.resolve()) known: dict[Path, str] = {} for entry in self._entries_under(root): + _beat() abs_path = self._entry_path(root, entry) if abs_path: known[abs_path] = entry.id - return on_disk, known + gone = [] + for path in unread: + _beat() + try: + if Path(path).resolve() not in on_disk: + gone.append(path) + except OSError: + continue + return on_disk, known, gone def _restart_observer(self) -> None: """Re-schedule watches after roots appeared or disappeared.""" @@ -1168,7 +1577,7 @@ class DirectoryIndexer: return True for path, size, mtime in known: try: - st = Path(path).stat() + st = os.stat(long_path(path)) except OSError: continue if st.st_size == size and (mtime is None or st.st_mtime == mtime): @@ -1276,6 +1685,9 @@ class DirectoryIndexer: if root is None or not root.paused: return root.paused = False + # Files that gave no answer stay in `unread` and are not read again by + # the walk this starts: it has to get past them to reach the rest. + self._forget_root_trouble(root) log.info("Root %r resumed", root.name) self._restart_observer() await self.publish_roots() @@ -1335,7 +1747,7 @@ class DirectoryIndexer: # bookkeeping still has to reach the `fire()` that actually runs. if not deleted and key not in self._burst_sizes: try: - size = file_path.stat().st_size + size = os.stat(long_path(file_path)).st_size except OSError: size = 0 if self._burst_inflight <= 0: @@ -1390,7 +1802,8 @@ class DirectoryIndexer: covers this directory too. """ try: - return not dir_path.exists() and dir_path.parent.is_dir() + return (not os.path.exists(long_path(dir_path)) + and os.path.isdir(long_path(dir_path.parent))) except OSError: return False @@ -1417,6 +1830,9 @@ class DirectoryIndexer: await self.on_change(self) return + if deleted: + self.unread.pop(str(file_path), None) + if not root.available or root.paused: return diff --git a/packages/meshbay-node/src/meshbay_node/ops/files.py b/packages/meshbay-node/src/meshbay_node/ops/files.py index 4609627..a448103 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/files.py +++ b/packages/meshbay-node/src/meshbay_node/ops/files.py @@ -4,8 +4,11 @@ from __future__ import annotations import asyncio import logging +import os from pathlib import Path +from meshbay_common.paths import long_path + from meshbay_node.ops.core import OpError, _group_ctx log = logging.getLogger("meshbay_node.ops") @@ -41,7 +44,7 @@ async def delete_file(state: dict, group_id: str, file_id: str) -> dict: f"not readable right now — the file is frozen, not gone", status=409) try: - path.unlink() + os.unlink(long_path(path)) except FileNotFoundError: # Already gone from disk; drop the stale entry rather than refusing. log.warning("Index named a file that is not on disk: %s", path) @@ -104,7 +107,7 @@ async def prune_index_cache(state: dict) -> dict: return True if any(not r.available for r in owning): return False - return not path.exists() + return not os.path.exists(long_path(path)) paths = await cache.all_paths() stale = await asyncio.to_thread(lambda: [p for p in paths if _is_stale(p)]) diff --git a/packages/meshbay-node/src/meshbay_node/ops/roots.py b/packages/meshbay-node/src/meshbay_node/ops/roots.py index 187e134..0cd2daa 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/roots.py +++ b/packages/meshbay-node/src/meshbay_node/ops/roots.py @@ -361,13 +361,16 @@ async def pause_root(state: dict, group_id: str, root_name: str) -> dict: 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: + indexer = state.get("indexers", {}).get(group_id) + # A root the node paused by itself, its disk having stopped answering, is + # paused on the indexer's own set and nowhere in the roster. + own = indexer.roots.by_name(root.name) if indexer else None + if not root.paused and not (own is not None and own.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 diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 61fdc9b..94560f5 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -36,7 +36,7 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path -from meshbay_common.paths import fold, portable_name_problem +from meshbay_common.paths import fold, long_path, portable_name_problem log = logging.getLogger(__name__) @@ -78,7 +78,7 @@ def _free_name(directory: Path, filename: str, or another upload (C5a). """ def free(name: str) -> bool: - return name not in taken and not (directory / name).exists() + return name not in taken and not os.path.exists(long_path(directory / name)) if free(filename): return filename @@ -109,17 +109,17 @@ def publish_upload(part: Path, directory: Path, stored_name: str, filename: str, for _ in range(8): target = directory / name try: - os.link(part, target) + os.link(long_path(part), long_path(target)) except FileExistsError: name = _free_name(directory, filename, taken) continue except OSError: - if target.exists(): + if os.path.exists(long_path(target)): name = _free_name(directory, filename, taken) continue - part.rename(target) + os.rename(long_path(part), long_path(target)) return name - part.unlink() + os.unlink(long_path(part)) return name raise FileExistsError(stored_name) @@ -195,7 +195,7 @@ class Root: def is_live(self) -> bool: """Readable right now. The question `available` caches.""" try: - return self.path.is_dir() + return os.path.isdir(long_path(self.path)) except OSError: return False diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 715448f..3974a2f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -48,6 +48,7 @@ from meshbay_common.handshake import ( quic_binding, verify_proof, ) +from meshbay_common.paths import long_path from meshbay_common.protocol import MNP, file_chunk_wire from meshbay_node.indexer import GroupIndex @@ -434,7 +435,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): if file_path is None: self._send(stream_id, {"type": "error", "detail": ROOT_NOT_SERVED}) return - if not file_path.exists(): + if not os.path.exists(long_path(file_path)): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return @@ -463,7 +464,7 @@ def _read_and_encrypt( file_id: str = "", ) -> dict: """Read one chunk off disk and encrypt it, in the one shape every transport uses.""" - with open(file_path, "rb") as f: + with open(long_path(file_path), "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py index e298942..386b81f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py @@ -1,8 +1,15 @@ -"""Blocking disk work the session hands to `off_disk`.""" +""" +Blocking disk work the session hands to `off_disk`. +Every call takes the path's long form (`long_path`): a file past Windows' +MAX_PATH is indexed, so it is served, deleted and written beside like any other. +""" + +import os import shutil from pathlib import Path +from meshbay_common.paths import long_path from meshbay_common.protocol import file_chunk_wire from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path @@ -22,7 +29,7 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: path = entry_abs_path(roots, entry) if path is None: return None, ROOT_NOT_SERVED - if not path.exists(): + if not os.path.exists(long_path(path)): return None, "File not on disk" return path, None @@ -37,15 +44,16 @@ def _mkdir_if_absent(target: Path) -> str | None: `mkdir` would raise where a refusal was meant. Blocking; called through `off_disk`. """ - if target.exists(): + if os.path.exists(long_path(target)): return "Already exists" - target.mkdir(parents=False) + os.mkdir(long_path(target)) return None def _is_empty_dir(target: Path) -> bool: """Blocking; called through `off_disk`.""" - return not any(target.iterdir()) + with os.scandir(long_path(target)) as listing: + return not any(listing) def _rmdir_if_empty(target: Path) -> bool: @@ -57,9 +65,10 @@ def _rmdir_if_empty(target: Path) -> bool: operator's browser, and a file can land in between. Two awaits here would reopen the same window one size smaller. Blocking; called through `off_disk`. """ - if any(target.iterdir()): - return False - target.rmdir() + with os.scandir(long_path(target)) as listing: + if any(listing): + return False + os.rmdir(long_path(target)) return True @@ -71,7 +80,7 @@ def _read_and_encrypt( file_id: str = "", ) -> dict: """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" - with open(file_path, "rb") as f: + with open(long_path(file_path), "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) @@ -80,10 +89,24 @@ def _read_and_encrypt( def _free_bytes(directory: Path) -> int: """Space left on the filesystem holding `directory`. Blocking; called through `off_disk`, since a `statvfs` wakes a sleeping disk like any stat.""" - return shutil.disk_usage(directory).free + return shutil.disk_usage(long_path(directory)).free def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" - with open(tmp_path, "wb" if first else "ab") as f: + with open(long_path(tmp_path), "wb" if first else "ab") as f: f.write(chunk_bytes) + + +def _unlink(path: Path, missing_ok: bool = False) -> None: + """Remove one file. Blocking; called through `off_disk`.""" + try: + os.unlink(long_path(path)) + except FileNotFoundError: + if not missing_ok: + raise + + +def _exists(path: Path) -> bool: + """Blocking; called through `off_disk`.""" + return os.path.exists(long_path(path)) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py index eafdaed..9c5fab5 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py @@ -17,6 +17,7 @@ from meshbay_node.transport.webrtc.disk import ( _mkdir_if_absent, _read_and_encrypt, _rmdir_if_empty, + _unlink, ) from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, LEASE_GRANTED, LEASE_QUEUED from meshbay_node.transport.wire import index_sync_message, list_dirs_off_loop @@ -439,7 +440,7 @@ class FilesMixin: self._send({"type": "error", "detail": ROOT_NOT_SERVED}) return if file_path is not None: - await off_disk(ctx["roots"], file_path.unlink) + await off_disk(ctx["roots"], _unlink, file_path) log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py index c052703..0d5bb0e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py @@ -17,7 +17,7 @@ from meshbay_node.roots import ( publish_upload, shell_active, ) -from meshbay_node.transport.webrtc.disk import _append_chunk, _free_bytes +from meshbay_node.transport.webrtc.disk import _append_chunk, _exists, _free_bytes, _unlink from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED log = logging.getLogger("meshbay_node.transport.webrtc_server") @@ -365,7 +365,7 @@ class UploadMixin: if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. - if await off_disk(roots, final_path.exists): + if await off_disk(roots, _exists, final_path): _refuse("File already exists", "already_exists") return # Asked once, before the first byte, from the size the client @@ -396,7 +396,7 @@ class UploadMixin: if state.bytes + len(chunk_bytes) > self._max_upload_bytes(): uploads.drop(user_id, rel_dir, filename) - await off_disk(roots, tmp_path.unlink, True) + await off_disk(roots, _unlink, tmp_path, True) _refuse("Upload exceeds size limit", "too_large") return @@ -410,7 +410,7 @@ class UploadMixin: # space that ran out. The client starts over from zero, which is # what it would have to do for a resume against a full disk anyway. uploads.drop(user_id, rel_dir, filename) - await off_disk(roots, tmp_path.unlink, True) + await off_disk(roots, _unlink, tmp_path, True) log.warning("Upload refused: the disk holding %s is full", rel_dir) self._audit("upload_refused", "disk_full") _refuse("The node's disk is full", "disk_full") diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py index 77c5d62..75d27c1 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/wire.py +++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py @@ -32,6 +32,7 @@ from pathlib import Path from meshbay_common import MNP_VERSION from meshbay_common.groupbox import PURPOSE_INDEX, seal +from meshbay_common.paths import long_path from meshbay_common.protocol import MNP, index_entry_wire from meshbay_node.roots import RootSet, off_disk @@ -75,22 +76,40 @@ def _dir_roots(roots: RootSet) -> list[tuple[str, Path, bool]]: def _walk_dirs(dir_roots: list[tuple[str, Path, bool]]) -> list[str]: - """Blocking. Directories only: `os.walk` reads each one once and never - stats a file, which `rglob("*")` and an `is_dir()` per entry did.""" + """Blocking. Directories only: each is read once and no file is stat'ed, + which `rglob("*")` and an `is_dir()` per entry did. What `os.walk` did, + except that each directory is listed under its long form: `os.walk` skipped + whatever lay past Windows' MAX_PATH, without a word.""" out: list[str] = [] for name, path, available in dir_roots: out.append(name) if not available: continue - for top, subdirs, _files in os.walk(path): - # Pruned in place: nothing under a hidden directory is listed either. - subdirs[:] = [d for d in subdirs if not d.startswith(".")] - rel = Path(top).relative_to(path) - for d in subdirs: - out.append(f"{name}/{(rel / d).as_posix()}") + pending = [path] + while pending: + top = pending.pop() + try: + with os.scandir(long_path(top)) as listing: + # Nothing under a hidden directory is listed either. + subdirs = [e for e in listing + if not e.name.startswith(".") and _is_dir(e)] + except OSError: + continue + rel = top.relative_to(path) + for entry in subdirs: + out.append(f"{name}/{(rel / entry.name).as_posix()}") + if not entry.is_symlink(): # listed, not entered, as os.walk did + pending.append(top / entry.name) return sorted(out)[:MAX_DIRS] +def _is_dir(entry: os.DirEntry) -> bool: + try: + return entry.is_dir() + except OSError: + return False + + def index_sync_message(index, roots: RootSet | None, hidden=(), dirs: list[str] | None = None) -> dict: """ diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 9933750..f5cd5d0 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -23,6 +23,9 @@ from meshbay_node import __version__, ops log = logging.getLogger(__name__) +# Unread files sent to the operator's page at once; `unread_total` says the rest. +UNREAD_LISTED = 100 + def _op(coro): """ @@ -490,17 +493,28 @@ def create_ui_app(state: dict) -> FastAPI: is only added to groups_ctx once that scan finishes (it is not yet authorized for member connections either way — see _reload_config) — this is precisely the window the wizard needs to watch. + + Also what the scan could not do, with paths, which is why it is here + and nowhere a member can ask: `unread`, the files it met and did not + index, each with the reason (`no_answer` when the disk said nothing); + and `stalled`, the folders the node paused by itself because their + disk stopped answering, each with where. """ indexer = state.get("indexers", {}).get(group_id) progress = indexer.progress if indexer else None if progress is None: return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0, - "current_dir": ""} + "current_dir": "", "unread": [], "unread_total": 0, "stalled": []} return { "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, "current_dir": progress.current_dir, + "unread": [{"root": u.root, "path": u.path, "reason": u.reason} + for u in list(indexer.unread.values())[:UNREAD_LISTED]], + "unread_total": indexer.unread_count(), + "stalled": [{"root": name, "path": where} + for name, where in indexer.stalled.items()], } @app.get("/api/index-status") @@ -512,12 +526,17 @@ def create_ui_app(state: dict) -> FastAPI: is on screen whatever page the operator is on, so it cannot ask per group. Names roots, like `current_dir` above: loopback only, the operator's own screen. Reads state["indexers"] for the same reason. + + `slow_path` and `slow_secs` say what the disk has given no answer on, + and for how long, once that passes ten seconds: a scan that sits still + names the file it is sitting on. """ config = state.get("config") names = {g.id: g.name for g in config.groups} if config else {} groups = [] for gid, indexer in list(state.get("indexers", {}).items()): p = indexer.progress + slow = indexer.disk_wait() groups.append({ "group_id": gid, "group_name": names.get(gid, gid[:8]), @@ -530,6 +549,10 @@ def create_ui_app(state: dict) -> FastAPI: "files_done": p.files_done, "files_total": p.files_total, "queued": list(p.queued), + "slow_path": slow[0] if slow else "", + "slow_secs": int(slow[1]) if slow else 0, + "unread": indexer.unread_count(), + "stalled": list(indexer.stalled), }) return {"groups": groups} diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py index c485e39..a0f65dc 100644 --- a/packages/meshbay-node/src/meshbay_node/uploads.py +++ b/packages/meshbay-node/src/meshbay_node/uploads.py @@ -24,6 +24,7 @@ build first. from __future__ import annotations +import os import re import secrets import time @@ -31,6 +32,10 @@ from collections.abc import Iterable from dataclasses import dataclass, field from pathlib import Path +from meshbay_common.paths import long_path + +from meshbay_node.indexer.indexer import walk_files + # What an unfinished upload is called on disk while it is being written. The # node has always used this; it is named here because the reaper below has to # recognise one, and a second spelling of it would be a bug nobody could see. @@ -208,14 +213,14 @@ def find_parts(roots: Iterable) -> list[tuple[Path, float]]: if not getattr(root, "available", False): continue try: - candidates = root.path.rglob(f"*{PART_SUFFIX}") + candidates = walk_files(root.path) except OSError: continue for path in candidates: + if not path.name.endswith(PART_SUFFIX): + continue try: - if not path.is_file(): - continue - found.append((path, path.stat().st_mtime)) + found.append((path, os.stat(long_path(path)).st_mtime)) except OSError: continue return found diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py index c0ab052..cacf936 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -260,7 +260,7 @@ def test_no_handler_touches_the_disk_on_the_loop(): # Written to block, and reached only through `off_disk`. on_the_disk_thread = {"_locate", "_append_chunk", "_read_and_encrypt", "_mkdir_if_absent", "_is_empty_dir", "_rmdir_if_empty", - "safe_subdir"} + "_unlink", "_exists", "safe_subdir"} # ffmpeg's own output goes through `_read_scratch_capped` and # `_discard_scratch` on a worker thread — `asyncio.to_thread` and not # `off_disk`, because a temp file is not a group root and has no platter to diff --git a/packages/meshbay-node/tests/test_index_jobs_are_described.py b/packages/meshbay-node/tests/test_index_jobs_are_described.py index dee6e39..1eb203e 100644 --- a/packages/meshbay-node/tests/test_index_jobs_are_described.py +++ b/packages/meshbay-node/tests/test_index_jobs_are_described.py @@ -55,11 +55,11 @@ class _Held(DirectoryIndexer): self.walked.append(root.name) return await super()._scan_root(root, *args, **kwargs) - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): if self.hold_under is not None and file_path.is_relative_to(self.hold_under): self.at_gate.set() await self.gate.wait() - return await super()._hash_or_cached(root, file_path) + return await super()._hash_or_cached(root, file_path, **kw) def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]: @@ -149,7 +149,7 @@ async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path): _, r1, r2 = _dirs(tmp_path) class _Broken(DirectoryIndexer): - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): raise RuntimeError("simulated failure mid-scan") idx = _Broken(roots=_set(r1, r2), group_id=GROUP, diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py index ecdc223..2fa8a5c 100644 --- a/packages/meshbay-node/tests/test_index_progress.py +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -14,7 +14,7 @@ from fastapi.testclient import TestClient from meshbay_node.config import Config, HubConfig, KeystoreConfig, NodeConfig from meshbay_node.daemon import NodeDaemon from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.indexer.indexer import IndexProgress +from meshbay_node.indexer.indexer import NO_ANSWER, IndexProgress, UnreadFile from meshbay_node.transport.webrtc_server import WebRTCPeerSession from meshbay_node.ui.app import create_ui_app @@ -64,19 +64,30 @@ def _ui_client(state: dict) -> TestClient: "indexes": {}, **state})) +def _indexer(progress: IndexProgress, *, unread=(), stalled=None, slow=None) -> MagicMock: + """What the routes read off an indexer, with nothing wrong unless said.""" + indexer = MagicMock() + indexer.progress = progress + indexer.unread = {u.path: u for u in unread} + indexer.unread_count.return_value = len(indexer.unread) + indexer.stalled = dict(stalled or {}) + indexer.disk_wait.return_value = slow + return indexer + + def test_index_status_route_idle_for_unknown_group(): client = _ui_client({"indexers": {}}) resp = client.get("/api/groups/unknown-group/index-status") assert resp.status_code == 200 assert resp.json() == {"scanning": False, "scanned_bytes": 0, - "total_bytes": 0, "current_dir": ""} + "total_bytes": 0, "current_dir": "", + "unread": [], "unread_total": 0, "stalled": []} def test_index_status_route_reflects_indexer_progress(): - fake_indexer = MagicMock() - fake_indexer.progress = IndexProgress( + fake_indexer = _indexer(IndexProgress( scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000, - current_dir="2024") + current_dir="2024")) client = _ui_client({"indexers": {"g" * 32: fake_indexer}}) resp = client.get(f"/api/groups/{'g' * 32}/index-status") @@ -84,20 +95,47 @@ def test_index_status_route_reflects_indexer_progress(): assert resp.json() == { "scanning": True, "scanned_bytes": 1_000_000, "total_bytes": 4_000_000_000, "current_dir": "2024", + "unread": [], "unread_total": 0, "stalled": [], } +def test_the_operator_is_shown_what_a_scan_could_not_read_and_where_it_stopped(): + """On the node's own machine only: these are paths. A scan that sat still + used to say nothing about the file it sat on, and a file it gave up on was + one line in a log the operator does not read.""" + silent = UnreadFile(path="/srv/music/a/one.flac", root="music", reason=NO_ANSWER, + size=1, mtime=1.0) + refused = UnreadFile(path="/srv/music/b/two.flac", root="music", + reason="Permission denied", size=1, mtime=1.0) + indexer = _indexer(IndexProgress(), unread=[silent, refused], + stalled={"music": "/srv/music/a/three.flac"}, + slow=("/srv/music/a/four.flac", 23.7)) + config = MagicMock() + config.groups = [] + client = _ui_client({"config": config, "indexers": {"g" * 32: indexer}}) + + one = client.get(f"/api/groups/{'g' * 32}/index-status").json() + assert one["unread"] == [ + {"root": "music", "path": "/srv/music/a/one.flac", "reason": "no_answer"}, + {"root": "music", "path": "/srv/music/b/two.flac", "reason": "Permission denied"}] + assert one["unread_total"] == 2 + assert one["stalled"] == [{"root": "music", "path": "/srv/music/a/three.flac"}] + + (band,) = client.get("/api/index-status").json()["groups"] + assert (band["slow_path"], band["slow_secs"]) == ("/srv/music/a/four.flac", 23) + assert (band["unread"], band["stalled"]) == (2, ["music"]) + + # ── /api/index-status (loopback, every group) ─────────────────────────────── def test_every_group_is_described_including_one_still_being_attached(): """The band reads one route for the whole node. A group in its initial scan is in state["indexers"] and not yet in groups_ctx, and must be there.""" - busy, idle = MagicMock(), MagicMock() - busy.progress = IndexProgress( + busy = _indexer(IndexProgress( scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024", root="results", root_pos=1, kind="scan", files_done=1, files_total=4, - queued=["archive"]) - idle.progress = IndexProgress() + queued=["archive"])) + idle = _indexer(IndexProgress()) config = MagicMock() named = MagicMock() named.id, named.name = "a" * 32, "outputs" @@ -112,6 +150,7 @@ def test_every_group_is_described_including_one_still_being_attached(): "kind": "scan", "root": "results", "current_dir": "2024", "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4, "queued": ["archive"], + "slow_path": "", "slow_secs": 0, "unread": 0, "stalled": [], } assert groups["b" * 32]["group_name"] == "b" * 8 assert groups["b" * 32]["scanning"] is False diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index fda40c8..5540be8 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -418,7 +418,14 @@ async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, assert indexer.progress.total_bytes == total_size assert indexer.progress.scanned_bytes == 0 - await asyncio.sleep(0.05) # past debounce_secs; lets all three fire and finish + # Past debounce_secs, until all three have fired and finished. Not a fixed + # 50 ms: a Windows timer ticks every 15.6 ms, the debounce alone can take + # two of them, and the test failed one run in five there on a count of + # milliseconds that says nothing about the indexer. + for _ in range(200): + if not indexer.progress.scanning: + break + await asyncio.sleep(0.01) assert indexer.progress.scanning is False, "must end idle, not stuck scanning" assert indexer.progress.scanned_bytes == total_size diff --git a/packages/meshbay-node/tests/test_indexer_silent_disk.py b/packages/meshbay-node/tests/test_indexer_silent_disk.py new file mode 100644 index 0000000..4073ac5 --- /dev/null +++ b/packages/meshbay-node/tests/test_indexer_silent_disk.py @@ -0,0 +1,485 @@ +""" +A disk that stops answering, and the scan that used to wait on it for ever. + +A read on a failing drive, a share that dropped or a file a sync client has not +fetched does not fail, it does not return. The hash ran on the indexer's one +thread with nothing to bound it, so one such file stopped the scan of its group +where it stood: every root behind it unread, the enrichment that follows a scan +never started, a pause unable to take effect (it is looked at between two +files), and nothing in the log to say which file, since a file is named there +only when it fails. + +What is timed is silence, not work: a listing of a million entries and a file +on a slow link both take as long as they take. The files below "never answer" +by waiting on an event the test sets when it ends. +""" + +import asyncio +import os +import threading +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.indexer import indexer as indexer_mod +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import NO_ANSWER, DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.roster import Roster + +pytestmark = pytest.mark.asyncio + +GID = "g" * 32 + + +@pytest.fixture(autouse=True) +def _short_patience(monkeypatch): + """Tenths of a second where the node waits a minute.""" + monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 0.3) + monkeypatch.setattr(DirectoryIndexer, "SLOW_SECS", 0.1) + monkeypatch.setattr(DirectoryIndexer, "DISK_POLL_SECS", 0.02) + + +@pytest.fixture +def never(): + """What a read that never returns waits on. Set at the end, so the threads + the tests leave behind do end.""" + event = threading.Event() + yield event + event.set() + + +def _folder(tmp_path: Path, name: str, files: list[str]) -> Path: + d = tmp_path / name + d.mkdir() + for f in files: + (d / f).write_bytes(f"{name}/{f}".encode()) + return d + + +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 _silent_files(monkeypatch, never: threading.Event, silent=lambda name: False) -> list[str]: + """Reads of the files `silent` names never return. Every read is recorded.""" + read: list[str] = [] + real = indexer_mod._scan_file + + def scan(root, path): + read.append(path.name) + if silent(path.name): + never.wait() + return None + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", scan) + return read + + +# ── One file ───────────────────────────────────────────────────────────────── + +async def test_a_read_that_never_returns_costs_its_file_not_the_scan( + tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv", "stuck.mkv", "c.mkv", "d.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "d.mkv"} + (unread,) = idx.unread.values() + assert (Path(unread.path).name, unread.root, unread.reason) == ( + "stuck.mkv", "Films", NO_ANSWER) + assert roots.roots[0].paused is False, "one file is not the root's fault" + assert idx.progress.scanning is False + assert idx.progress.files_done == 5, "the file given up on is done with, too" + + +async def test_reconcile_does_not_wait_on_that_file_again_until_it_changes( + tmp_path, monkeypatch, never): + """Reconcile indexes every file the index lacks, at every pass. Without a + memory of this one it would wait out the limit on it each time.""" + films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + answering = indexer_mod._scan_file + read = _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + await idx.initial_scan() + assert read.count("stuck.mkv") == 1 + + await idx.reconcile() + await idx.reconcile() + assert read.count("stuck.mkv") == 1, "asked the disk again for a file it left unanswered" + + monkeypatch.setattr(indexer_mod, "_scan_file", answering) # the disk is back... + await idx.reconcile() + assert "stuck.mkv" not in _names(idx), "...which nothing has told the node" + + (films / "stuck.mkv").write_bytes(b"written again, and longer") + await idx.reconcile() + assert "stuck.mkv" in _names(idx) + assert idx.unread == {} + + +async def test_a_file_the_system_refuses_is_listed_with_what_it_said( + tmp_path, monkeypatch): + films = _folder(tmp_path, "Films", ["a.mkv", "locked.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + + def refuse(path, *args, **kw): + # What a locked file does: it lists and stats, and refuses to open. + if Path(path).name == "locked.mkv": + raise PermissionError(13, "Permission denied") + return open(path, *args, **kw) + + monkeypatch.setattr(indexer_mod, "open", refuse, raising=False) + await idx.initial_scan() + + assert _names(idx) == {"a.mkv"} + (unread,) = idx.unread.values() + assert (Path(unread.path).name, unread.reason) == ("locked.mkv", "Permission denied") + + # A refusal costs nothing to ask again, and permissions change without the + # file changing: it is not remembered the way a silence is. + monkeypatch.delattr(indexer_mod, "open") + await idx.reconcile() + assert "locked.mkv" in _names(idx) + assert idx.unread == {} + + +async def test_a_file_that_has_gone_is_no_longer_unread(tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv", "stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: name == "stuck.mkv") + await idx.initial_scan() + assert len(idx.unread) == 1 + + (films / "stuck.mkv").unlink() + await idx.reconcile() + + assert idx.unread == {} + + +# ── What counts as silence ─────────────────────────────────────────────────── + +async def test_a_slow_file_that_keeps_arriving_is_not_given_up_on(tmp_path, monkeypatch): + """Twice the limit in all, never the limit without a megabyte.""" + films = _folder(tmp_path, "Films", ["slow.mkv"]) + root = RootSet.build([{"path": str(films)}]) + idx = _indexer(root) + real = indexer_mod._scan_file + + def trickle(root, path): + for _ in range(6): + threading.Event().wait(0.1) + indexer_mod._beat() + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", trickle) + await idx.initial_scan() + + assert _names(idx) == {"slow.mkv"} + assert idx._executor.left_behind == 0 + + +async def test_files_waiting_behind_a_slow_one_are_not_timed_for_the_wait( + tmp_path, monkeypatch): + """A season dropped into a watched folder is hashed one file at a time on + the one thread, each event its own task. Timed from when it was asked, the + last of them would be given up on for having queued.""" + names = [f"e{i}.mkv" for i in range(5)] + films = _folder(tmp_path, "Films", names) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + real = indexer_mod._scan_file + + def unhurried(root, path): + threading.Event().wait(0.15) + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", unhurried) + entries = await asyncio.gather(*( + idx._hash_or_cached(roots.roots[0], films / name) for name in names)) + + assert all(entries), "0.75 s of queue against a limit of 0.3 s" + assert idx.unread == {} and idx._executor.left_behind == 0 + + +# ── Three in a row ─────────────────────────────────────────────────────────── + +async def test_three_in_a_row_pause_the_root_and_the_next_root_is_indexed( + tmp_path, monkeypatch, never): + """A minute a file is weeks on a music library, with a thread left behind + each time. After three with none read in between it is the disk.""" + sick = _folder(tmp_path, "Sick", [f"s{i}.mkv" for i in range(6)]) + well = _folder(tmp_path, "Well", ["w0.mkv", "w1.mkv"]) + roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}]) + told = [] + + async def on_change(indexer): + told.append([r["paused"] for r in indexer.index.roots]) + + idx = _indexer(roots, on_change=on_change) + read = _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert sum(name.startswith("s") for name in read) == 3, read + assert roots.by_name("Sick").paused is True + assert Path(idx.stalled["Sick"]).parent == sick + assert _names(idx) == {"w0.mkv", "w1.mkv"}, "the root behind it waited for ever" + assert told and told[0] == [True, False], "members are told the root is paused" + assert idx.progress.scanning is False and idx.progress.queued == [] + + +async def test_a_file_read_in_between_starts_the_count_again(tmp_path, monkeypatch, never): + names = ["s0", "s1", "good", "s2", "s3", "s4"] + films = _folder(tmp_path, "Films", [n + ".mkv" for n in names]) + roots = RootSet.build([{"path": str(films)}]) + root = roots.roots[0] + idx = _indexer(roots) + _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + + for name in names[:5]: + await idx._hash_or_cached(root, films / f"{name}.mkv") + assert root.paused is False, "two, one read, two: never three in a row" + + await idx._hash_or_cached(root, films / "s4.mkv") + assert root.paused is True + + +async def test_resume_gets_past_the_files_that_gave_no_answer(tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + read = _silent_files(monkeypatch, never, lambda name: name.startswith("s")) + await idx.initial_scan() + assert roots.roots[0].paused is True + (films / "new.mkv").write_bytes(b"dropped in while it was paused") + + await idx.resume_root("Films") + await asyncio.gather(*idx._scan_tasks) + + assert _names(idx) == {"new.mkv"} + assert len(read) == 4, f"read again what gave no answer: {read}" + assert roots.roots[0].paused is False and idx.stalled == {} + assert len(idx.unread) == 3, "still listed for the operator" + + +async def test_a_reload_does_not_start_a_stalled_root_again(tmp_path, monkeypatch, never): + """A reload builds the set from the roster, which holds the operator's + pauses. This one is the node's own.""" + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + + again = RootSet.build([{"path": str(films)}]) + await idx.retarget(again) + + assert again.roots[0].paused is True + + +def _state(roots: RootSet, roster: Roster, idx: DirectoryIndexer) -> dict: + index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}}, + "indexers": {GID: idx}, + "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]), + "roster": roster, + "node_user_id": "operator", + } + + +async def test_the_operators_resume_reaches_a_root_the_node_paused( + tmp_path, monkeypatch, never): + """Paused on the indexer's set and nowhere in the roster; and after a + reload the set the operations look at may be another object.""" + films = _folder(tmp_path, "Films", ["s0.mkv", "s1.mkv", "s2.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + seen_by_ops = RootSet.build([{"path": str(films)}]) + out = await ops.resume_root(_state(seen_by_ops, roster, idx), GID, "Films") + await asyncio.gather(*idx._scan_tasks) + + assert out["status"] == "resumed" + assert idx.roots.roots[0].paused is False and idx.stalled == {} + finally: + await roster.close() + + +# ── The listing ────────────────────────────────────────────────────────────── + +async def test_a_listing_that_stops_answering_pauses_the_root(tmp_path, monkeypatch, never): + sick = _folder(tmp_path, "Sick", ["a.mkv"]) + well = _folder(tmp_path, "Well", ["w.mkv"]) + roots = RootSet.build([{"path": str(sick)}, {"path": str(well)}]) + idx = _indexer(roots) + real = indexer_mod._walk_root + + def walk(root): + if root.name == "Sick": + indexer_mod._beat(sick / "deep" / "down") + never.wait() + return real(root) + + monkeypatch.setattr(indexer_mod, "_walk_root", walk) + await asyncio.wait_for(idx.initial_scan(), timeout=10) + + assert roots.by_name("Sick").paused is True + assert idx.stalled == {"Sick": str(sick / "deep" / "down")}, "where it went silent" + assert _names(idx) == {"w.mkv"} + + +async def test_a_rescan_that_gets_no_listing_keeps_what_the_root_held( + tmp_path, monkeypatch, never): + """A rescan drops the root's entries before it walks. A walk that learns + nothing must not leave the root empty for everyone.""" + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + for entry in idx.index.entries: + entry.display_title = "Kept" + + monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait()) + await asyncio.wait_for(idx._rescan_root(roots.roots[0]), timeout=10) + + assert _names(idx) == {"a.mkv", "b.mkv"} + assert all(e.display_title == "Kept" for e in idx.index.entries) + assert idx.drain_rescanned_ids() == set(), "nothing was rebuilt, nothing to enrich again" + + +async def test_a_reconcile_that_gets_no_listing_pauses_and_removes_nothing( + tmp_path, monkeypatch, never): + films = _folder(tmp_path, "Films", ["a.mkv"]) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + + monkeypatch.setattr(indexer_mod, "_walk_root", lambda root: never.wait()) + await asyncio.wait_for(idx.reconcile(), timeout=10) + + assert roots.roots[0].paused is True + assert _names(idx) == {"a.mkv"} + + +def _listed_before(root_path: Path) -> set[Path]: + """The walk this replaced.""" + return {p for p in root_path.rglob("*") if p.is_file()} + + +async def test_the_walk_lists_what_it_always_listed(tmp_path): + root = _folder(tmp_path, "Root", ["top.mkv", ".hidden", "x.part"]) + (root / "a" / "b" / "c").mkdir(parents=True) + (root / "a" / "one.mkv").write_bytes(b"1") + (root / "a" / "b" / "c" / "deep.mkv").write_bytes(b"2") + (root / "empty").mkdir() + outside = _folder(tmp_path, "Outside", ["o.mkv"]) + try: + os.symlink(outside / "o.mkv", root / "link_to_file.mkv") + os.symlink(outside, root / "link_to_dir", target_is_directory=True) + except OSError: + pass # no right to make links here: the rest still holds + listed = indexer_mod._walk_root(RootSet.build([{"path": str(root)}]).roots[0]) + + assert set(listed) == _listed_before(root) + assert len(listed) == len(set(listed)) + + +async def test_a_directory_that_cannot_be_listed_costs_only_itself(tmp_path, monkeypatch): + """One refusal used to be silent, and any other error ended the walk of the + whole root: "Cannot scan root", and nothing indexed on it at all.""" + root = _folder(tmp_path, "Root", ["top.mkv"]) + for sub in ("ok", "broken"): + (root / sub).mkdir() + (root / sub / f"{sub}.mkv").write_bytes(sub.encode()) + real = os.scandir + + def scandir(path): + if Path(path).name == "broken": + raise OSError(5, "Input/output error") + return real(path) + + monkeypatch.setattr(indexer_mod.os, "scandir", scandir) + idx = _indexer(RootSet.build([{"path": str(root)}])) + await idx.initial_scan() + + assert _names(idx) == {"top.mkv", "ok.mkv"} + + +# ── Said, and survivable ───────────────────────────────────────────────────── + +async def test_the_operator_is_told_which_file_the_disk_is_silent_on( + tmp_path, monkeypatch, never): + monkeypatch.setattr(DirectoryIndexer, "STALL_SECS", 30) + films = _folder(tmp_path, "Films", ["stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + assert idx.disk_wait() is None + + scan = asyncio.create_task(idx.initial_scan()) + try: + for _ in range(200): + if idx.disk_wait(): + break + await asyncio.sleep(0.02) + path, seconds = idx.disk_wait() + assert Path(path) == films / "stuck.mkv" and seconds >= 0.1 + finally: + never.set() + await asyncio.wait_for(scan, timeout=10) + assert idx.disk_wait() is None + + +async def test_the_node_can_stop_with_a_read_still_stuck(tmp_path, monkeypatch, never): + """The interpreter joins the threads of a pool on its way out. One read + that never returned kept the node from stopping at all.""" + films = _folder(tmp_path, "Films", ["stuck.mkv"]) + idx = _indexer(RootSet.build([{"path": str(films)}])) + _silent_files(monkeypatch, never, lambda name: True) + await idx.initial_scan() + assert idx._executor.left_behind == 1 + + await asyncio.wait_for(idx.stop(), timeout=5) + + stuck = [t for t in threading.enumerate() if t.name.startswith("indexer_")] + assert stuck and all(t.daemon for t in stuck) + + +async def test_a_scan_asks_the_disk_nothing_from_the_event_loop(tmp_path, monkeypatch): + """The hash was off the loop and the stat before it was not. On a disk that + stops answering that stat is where the loop would wait, and with it every + connection of the node: no limit on the indexer's thread helps a call made + somewhere else.""" + films = _folder(tmp_path, "Films", ["a.mkv", "b.mkv"]) + (films / "sub").mkdir() + (films / "sub" / "c.mkv").write_bytes(b"c") + idx = _indexer(RootSet.build([{"path": str(films)}])) + asked: set[str] = set() + real = Path.stat + + def stat(self, **kw): + if films in self.parents: + asked.add(threading.current_thread().name.split("_")[0]) + return real(self, **kw) + + monkeypatch.setattr(Path, "stat", stat) + await idx.initial_scan() + (films / "new.mkv").write_bytes(b"missed by the watcher") + await idx.reconcile() + + assert _names(idx) == {"a.mkv", "b.mkv", "c.mkv", "new.mkv"} + assert asked == {"indexer"}, f"a stat ran on: {asked}" diff --git a/packages/meshbay-node/tests/test_long_paths.py b/packages/meshbay-node/tests/test_long_paths.py new file mode 100644 index 0000000..9f7876a --- /dev/null +++ b/packages/meshbay-node/tests/test_long_paths.py @@ -0,0 +1,400 @@ +""" +Files whose full path is longer than Windows' MAX_PATH (260 characters). + +A deep library reaches it without trying: "Artist / Album (year, edition) / +01 - a long track title.flac" below a folder that is already deep. With the +system's default setting (`LongPathsEnabled` off) every call given a plain path +that long refuses it, whatever the program declares, and such a file was +invisible to the members: the walk could not list its directory, and said so +only in the log. The node now hands the system the extended form (`\\\\?\\`) at +each call that touches a group's disk, which works whatever the setting, and +keeps the plain form everywhere else: in the index, the hash cache, the log. + +ffmpeg and ffprobe are given the plain form on purpose: they extend a long path +themselves, which the probe and thumbnail tests below hold. + +These run only where the limit exists: with the setting on, a plain path works +too and they would prove nothing. +""" + +import asyncio +import os +import shutil +import sys +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.protocol import file_chunk_plaintext +from meshbay_node import ops +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet, publish_upload, safe_subdir +from meshbay_node.transport.webrtc.disk import ( + _append_chunk, + _free_bytes, + _is_empty_dir, + _locate, + _mkdir_if_absent, + _read_and_encrypt, + _rmdir_if_empty, +) + + +def _limit_in_force() -> bool: + if sys.platform != "win32": + return False + import winreg + try: + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, + r"SYSTEM\CurrentControlSet\Control\FileSystem") as key: + return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0 + except OSError: + return True + + +pytestmark = pytest.mark.skipif( + not _limit_in_force(), reason="MAX_PATH binds only on Windows with LongPathsEnabled off") + +GID = "g" * 32 +SEGMENT = "A folder name as long as the ones a real library has" + + +def _x(path: Path) -> str: + """The extended form, for the test's own setup and checks.""" + return "\\\\?\\" + str(path) + + +def _write(path: Path, data: bytes) -> Path: + with open(_x(path), "wb") as f: + f.write(data) + return path + + +def _read(path: Path) -> bytes: + with open(_x(path), "rb") as f: + return f.read() + + +@pytest.fixture +def library(tmp_path): + """A root with one short file at the top and a directory past the limit. + Removed here: the plain-path cleanup pytest does would stop at it.""" + root = tmp_path / "Library" + deep = root + while len(str(deep)) < 280: + deep = deep / SEGMENT + os.makedirs(_x(deep)) + _write(root / "short.txt", b"near the top") + yield root, deep + shutil.rmtree(_x(root), ignore_errors=True) + + +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 _entry(idx: DirectoryIndexer, name: str): + (entry,) = [e for e in idx.index.entries if e.name == name] + return entry + + +def _ffmpeg() -> str: + exe = shutil.which("ffmpeg") + if exe is None: + pytest.skip("ffmpeg is not on PATH") + return exe + + +async def _make_media(args: list[str], out: Path) -> None: + proc = await asyncio.create_subprocess_exec( + _ffmpeg(), "-v", "error", "-y", *args, str(out), + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE) + _, err = await proc.communicate() + assert proc.returncode == 0, err.decode(errors="replace") + + +def test_the_fixture_is_past_the_limit(library): + """The premise: a plain path this long is refused on this machine.""" + _, deep = library + track = _write(deep / "track.mkv", b"deep down") + assert len(str(track)) > 260 + assert not os.path.exists(str(track)) + with pytest.raises(OSError): + os.scandir(str(deep)) + + +# ── Indexing ──────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_scan_indexes_a_file_past_the_limit(library): + root, deep = library + _write(deep / "track.mkv", b"deep down" * 100) + idx = _indexer(RootSet.build([{"path": str(root)}])) + + await idx.initial_scan() + + assert _names(idx) == {"short.txt", "track.mkv"} + assert _entry(idx, "track.mkv").size == 900 + assert idx.unread == {} + + +@pytest.mark.asyncio +async def test_reconcile_finds_a_new_one_and_leaves_a_known_one_alone(library): + """Reconcile compares the disk with the index by path. Were the two sides + to spell a long path differently, it would rewrite the entry at every + pass and push the change to every member.""" + root, deep = library + _write(deep / "track.mkv", b"deep down") + idx = _indexer(RootSet.build([{"path": str(root)}])) + await idx.initial_scan() + + _write(deep / "later.mkv", b"added while the node ran") + await idx.reconcile() + assert "later.mkv" in _names(idx) + + version = idx.index.version + await idx.reconcile() + assert idx.index.version == version, "a pass with nothing new changed the index" + + +@pytest.mark.asyncio +async def test_a_watcher_event_indexes_it(library): + root, deep = library + idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.01) + await idx.initial_scan() + idx._loop = asyncio.get_running_loop() + + _write(deep / "arrived.mkv", b"one more") + idx._schedule_update(deep / "arrived.mkv") + for _ in range(200): + if "arrived.mkv" in _names(idx): + break + await asyncio.sleep(0.01) + + assert "arrived.mkv" in _names(idx) + + +@pytest.mark.asyncio +async def test_the_real_watcher_sees_it(library): + root, deep = library + idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.05) + await idx.start() + try: + _write(deep / "watched.mkv", b"seen by the system's own notification") + for _ in range(100): + if "watched.mkv" in _names(idx): + break + await asyncio.sleep(0.05) + finally: + await idx.stop() + + assert "watched.mkv" in _names(idx) + + +@pytest.mark.asyncio +async def test_a_root_that_is_itself_past_the_limit(library): + _, deep = library + _write(deep / "track.mkv", b"deep down") + roots = RootSet.build([{"path": str(deep)}]) + assert roots.roots[0].is_live() + idx = _indexer(roots) + + await idx.initial_scan() + + assert _names(idx) == {"track.mkv"} + + +# ── Serving and deleting ──────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_it_is_served_to_a_member(library): + root, deep = library + data = os.urandom(5000) + _write(deep / "track.mkv", data) + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + entry = _entry(idx, "track.mkv") + gek = os.urandom(32) + + path, refusal = _locate(roots, entry) + assert refusal is None + msg = _read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id) + + assert file_chunk_plaintext(gek, msg) == data + + +@pytest.mark.asyncio +async def test_it_is_served_over_quic_too(library): + from meshbay_node.transport import quic_server + + root, deep = library + _write(deep / "track.mkv", b"over the other transport") + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + entry = _entry(idx, "track.mkv") + gek = os.urandom(32) + + path, refusal = _locate(roots, entry) + msg = quic_server._read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id) + + assert file_chunk_plaintext(gek, msg) == b"over the other transport" + + +@pytest.mark.asyncio +async def test_the_operator_can_delete_it(library): + root, deep = library + track = _write(deep / "track.mkv", b"to be deleted") + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + state = {"groups_ctx": {GID: {"index": idx.index, "roots": roots}}} + + await ops.delete_file(state, GID, _entry(idx, "track.mkv").id) + + assert not os.path.exists(_x(track)) + assert "track.mkv" not in _names(idx) + + +# ── Uploads and folders ───────────────────────────────────────────────────── + +def test_an_upload_lands_in_a_directory_past_the_limit(library): + root, deep = library + roots = RootSet.build([{"path": str(root), "writable": True}]) + rel = "/".join(["Library", *deep.relative_to(root).parts]) + target = safe_subdir(roots, rel) + assert target == deep + + assert _free_bytes(target) > 0 + part = target / "up.bin.0123abcd.part" + _append_chunk(part, b"first ", True) + _append_chunk(part, b"second", False) + assert publish_upload(part, target, "up.bin", "up.bin") == "up.bin" + assert _read(target / "up.bin") == b"first second" + + again = target / "up.bin.4567cdef.part" + _append_chunk(again, b"same name", True) + assert publish_upload(again, target, "up.bin", "up.bin") == "up (2).bin", \ + "a long path made the existing file look absent, and it was replaced" + assert _read(target / "up.bin") == b"first second" + + +def test_an_abandoned_upload_is_found(library): + from meshbay_node import uploads + + root, deep = library + roots = RootSet.build([{"path": str(root), "writable": True}]) + part = _write(deep / "film.mkv.89abcdef.part", b"half a film") + + found = [p for p, _ in uploads.find_parts(roots.roots)] + + assert found == [part] + + +def test_members_see_the_folders_past_the_limit(library): + """The folder tree a member browses and uploads into.""" + from meshbay_node.transport.wire import list_dirs + + root, deep = library + os.makedirs(_x(deep / "Deeper still")) + roots = RootSet.build([{"path": str(root)}]) + + dirs = list_dirs(roots) + + rel = "/".join(["Library", *deep.relative_to(root).parts]) + assert rel in dirs + assert f"{rel}/Deeper still" in dirs + + +def test_a_folder_is_created_and_removed_past_the_limit(library): + _, deep = library + folder = deep / "New folder" + assert _mkdir_if_absent(folder) is None + assert _mkdir_if_absent(folder) == "Already exists" + assert _is_empty_dir(folder) + assert _rmdir_if_empty(folder) + assert not os.path.exists(_x(folder)) + + +# ── Enrichment ────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +@pytest.mark.daemon_loop # a subprocess needs the proactor loop +async def test_audio_tags_and_the_cover_beside_it_are_read(library): + from meshbay_node.indexer.enrich_audio import _read_tags_and_cover + from PIL import Image + + _, deep = library + track = deep / "01 - A Long Track Title.flac" + await _make_media(["-f", "lavfi", "-i", "sine=duration=1", + "-metadata", "title=Some Song", "-metadata", "artist=Some Artist", + "-metadata", "album=Some Album"], track) + with open(_x(deep / "cover.jpg"), "wb") as f: + Image.new("RGB", (8, 8), "red").save(f, "JPEG") + + tags, duration, cover = await asyncio.to_thread(_read_tags_and_cover, track) + + assert tags["title"] == "Some Song" and tags["album"] == "Some Album" + assert duration == pytest.approx(1.0, abs=0.1) + assert cover == _read(deep / "cover.jpg") + + +@pytest.mark.asyncio +async def test_a_photo_is_read(library): + from meshbay_node.indexer.enrich_photo import _read_image + from PIL import Image + + _, deep = library + with open(_x(deep / "photo.jpg"), "wb") as f: + Image.new("RGB", (64, 48), "blue").save(f, "JPEG") + + thumb, width, height, _, _ = await asyncio.to_thread(_read_image, deep / "photo.jpg") + + assert (width, height) == (64, 48) and thumb + + +@pytest.mark.asyncio +@pytest.mark.daemon_loop # a subprocess needs the proactor loop +async def test_ffmpeg_probes_and_thumbnails_it_from_the_plain_path(library): + from meshbay_node.indexer.enrich import _make_thumbnail + from meshbay_node.media_probe import probe_video + + _, deep = library + clip = deep / "clip.mp4" + await _make_media(["-f", "lavfi", "-i", "testsrc=size=64x48:rate=5:duration=2", + "-pix_fmt", "yuv420p"], clip) + + probe = await probe_video(str(clip)) + thumb = await _make_thumbnail(clip, probe.duration) + + assert probe.duration == pytest.approx(2.0, abs=0.3) + assert thumb and thumb[:2] == b"\xff\xd8" + + +@pytest.mark.asyncio +async def test_a_show_folder_past_the_limit_numbers_its_episodes(library): + from meshbay_node.indexer.enrich import ( + _season_and_show_from_ancestors, + _synthetic_episode_number, + _title_from_siblings, + ) + + _, deep = library + season = deep / "Some Show" / "Season 1" + os.makedirs(_x(season)) + _write(season / "Some Show S01E01.mkv", b"") + _write(season / "Pilot without a number.mkv", b"") + _write(season / "Second without a number.mkv", b"") + later = season / "Second without a number.mkv" + + found = _season_and_show_from_ancestors(later) + assert found is not None + number, show = found + assert await asyncio.to_thread(_synthetic_episode_number, later, show, number) == 2 + assert await asyncio.to_thread(_title_from_siblings, later) == "Some Show" diff --git a/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py index 1f93ffa..056c0b9 100644 --- a/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py +++ b/packages/meshbay-node/tests/test_scan_progress_is_not_clobbered.py @@ -56,13 +56,13 @@ class _Held(DirectoryIndexer): self.rescanned: list[str] = [] self._seen = 0 - async def _hash_or_cached(self, root, file_path): + async def _hash_or_cached(self, root, file_path, **kw): if self.hold_under is not None and file_path.is_relative_to(self.hold_under): if self._seen == self.hold_after: self.at_gate.set() await self.gate.wait() self._seen += 1 - return await super()._hash_or_cached(root, file_path) + return await super()._hash_or_cached(root, file_path, **kw) async def _rescan_root(self, root): self.rescanned.append(root.name) |