diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/uploads.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/uploads.py | 190 |
1 files changed, 190 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py new file mode 100644 index 0000000..f8ae7f9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/uploads.py @@ -0,0 +1,190 @@ +""" +Partial uploads: the state that must outlive a connection, and the files that +must not outlive their upload. + +Two defects live here, and they are the same defect seen from two sides. + +An upload's progress was kept on the **session** — `WebRTCSession._uploads`, +keyed by `rel_dir/filename`. A dropped connection therefore lost it, and the +client's next chunk was refused with `not_started`: an upload interrupted at +99% could only be started again from zero. The state belongs to the group, not +to the connection that happened to carry it, and it is keyed by member as well, +because a shared directory means two people can be sending `IMG_1234.jpg` at +the same time and neither may inherit the other's position. + +And what the lost state left behind was a `.part` file that nothing would ever +finish, delete or even look at again. One abandoned upload of a film is a +gigabyte of somebody else's disk, kept for ever, invisible in the index because +`.part` is not an index entry. That is the leak this module reaps. + +Pure logic, no asyncio and no transport — the same shape as `transfers.py`, and +for the same reason: the rules are worth testing without a WebRTC connection to +build first. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterable + +# 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. +PART_SUFFIX = ".part" + +# How long a `.part` with no upload behind it is kept before it is deleted. +# +# Generous on purpose. The cost of waiting is disk; the cost of being wrong is +# deleting an upload somebody is still making, which is unrecoverable and looks +# to them like a transfer that failed for no reason. A day covers a laptop +# closed overnight, a phone that lost signal in a tunnel, and a client that +# reconnects on the next launch — all of which are resumable and none of which +# should be swept. +ORPHAN_AFTER_SECS = 24 * 3600 + + +@dataclass +class Partial: + """One upload in progress, as the node knows it between two chunks.""" + + stored_name: str + # The `.part` this upload is writing. Recorded rather than recomputed: the + # reaper compares paths, and a path rebuilt from a root name and a relative + # directory is a second implementation of something that must agree exactly + # with the first, for ever, or a live upload gets deleted. + part_path: Path | None = None + next_index: int = 0 + bytes: int = 0 + updated_at: float = field(default_factory=time.time) + + +class PartialUploads: + """ + Every upload this group has in flight, keyed by member. + + Held in the group context rather than on a session, so that a reconnecting + client finds its own upload exactly where it left it. The key is + `(user_id, rel_dir, filename)`: the directory and name alone would let one + member resume — or clobber the position of — another member's upload of the + same name, which a shared folder makes an ordinary occurrence rather than an + attack. + """ + + def __init__(self) -> None: + self._by_key: dict[tuple[str, str, str], Partial] = {} + + # ── the state itself ──────────────────────────────────────────────────── + + def start(self, user_id: str, rel_dir: str, filename: str, + stored_name: str, part_path: Path | None = None, + now: float | None = None) -> Partial: + """Begin (or begin again) an upload, discarding any earlier position.""" + state = Partial(stored_name=stored_name, part_path=part_path, + updated_at=time.time() if now is None else now) + self._by_key[(user_id, rel_dir, filename)] = state + return state + + def get(self, user_id: str, rel_dir: str, filename: str) -> Partial | None: + return self._by_key.get((user_id, rel_dir, filename)) + + def advance(self, user_id: str, rel_dir: str, filename: str, + chunk_index: int, nbytes: int, + now: float | None = None) -> Partial | None: + """Record one accepted chunk. Returns None if there is no such upload.""" + state = self._by_key.get((user_id, rel_dir, filename)) + if state is None: + return None + state.next_index = chunk_index + 1 + state.bytes += nbytes + # Touched on every chunk, because the reaper measures *silence*, not + # age: an upload that has been running for two days is not an orphan, + # and one that stopped two days ago is, whatever it started as. + state.updated_at = time.time() if now is None else now + return state + + def drop(self, user_id: str, rel_dir: str, filename: str) -> Partial | None: + return self._by_key.pop((user_id, rel_dir, filename), None) + + def __len__(self) -> int: + return len(self._by_key) + + # ── what the reaper must not touch ────────────────────────────────────── + + def live_paths(self) -> set[Path]: + """The `.part` files that still have an upload behind them. + + Deliberately without the member's identity: a file on disk has no owner, + and the only question the reaper asks is whether anyone is writing it. + """ + return {state.part_path for state in self._by_key.values() + if state.part_path is not None} + + +def orphaned_parts(candidates: Iterable[tuple[Path, float]], + live: set[Path], + now: float, + older_than: float = ORPHAN_AFTER_SECS) -> list[Path]: + """ + Which `.part` files may be deleted. + + `candidates` is `(path, mtime)` for every `.part` found under the group's + writable roots. A file is an orphan when **both** are true: no upload in + `live` is writing it, and nothing has been written to it for `older_than` + seconds. + + Both conditions are load-bearing. The first alone would delete an upload + that is mid-flight but whose state is held elsewhere; the second alone would + keep a file for a day after the upload that owned it was abandoned, which is + correct but is also the entire reason this is bounded rather than immediate. + + A file with a future mtime — a clock that went backwards, a filesystem with + a different idea of now — is left alone rather than treated as infinitely + old, because deleting is not reversible and a wrong clock is not evidence. + """ + doomed: list[Path] = [] + for path, mtime in candidates: + if path.suffix != PART_SUFFIX: + continue + if path in live: + continue + age = now - mtime + if age < older_than: + continue + doomed.append(path) + return doomed + + +def find_parts(roots: Iterable) -> list[tuple[Path, float]]: + """ + Every `.part` under these roots, with its modification time. + + Only writable, available roots are walked: a read-only root cannot have + received an upload, and an unavailable one is a disk that is not mounted — + walking it would find nothing and reporting nothing found there is how a + reaper deletes an entire library the day a drive is unplugged. (It cannot + here, since it only ever deletes what it finds, but the shape of that + mistake is worth refusing at the source.) + + Errors are swallowed per entry rather than per walk: one unreadable + subdirectory must not stop the rest from being tidied. + """ + found: list[tuple[Path, float]] = [] + for root in roots: + if not getattr(root, "writable", False): + continue + if not getattr(root, "available", False): + continue + try: + candidates = root.path.rglob(f"*{PART_SUFFIX}") + except OSError: + continue + for path in candidates: + try: + if not path.is_file(): + continue + found.append((path, path.stat().st_mtime)) + except OSError: + continue + return found |