1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
"""Blocking disk work the session hands to `off_disk`."""
from pathlib import Path
from meshbay_common.protocol import file_chunk_wire
from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path
from meshbay_node.transport.webrtc.limits import CHUNK_SIZE
def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]:
"""
Where an entry is, and whether it is readable — or the refusal to send.
Both halves are syscalls: `resolve()` walks the path and `exists()` stats
it, and a stat is what *wakes* a sleeping disk. Leaving either on the event
loop and offloading only the read would move the stall rather than remove
it, and the read would then find the disk already awake. Blocking; called
through `off_disk`.
"""
path = entry_abs_path(roots, entry)
if path is None:
return None, ROOT_NOT_SERVED
if not path.exists():
return None, "File not on disk"
return path, None
def _mkdir_if_absent(target: Path) -> str | None:
"""
Create a directory unless it is already there, or say why not.
Both in one call, not a check awaited and then an act: the disk thread is
one worker, so nothing can slip between them. Split across two awaits, two
members creating the same name would both find nothing there and the second
`mkdir` would raise where a refusal was meant. Blocking; called through
`off_disk`.
"""
if target.exists():
return "Already exists"
target.mkdir(parents=False)
return None
def _is_empty_dir(target: Path) -> bool:
"""Blocking; called through `off_disk`."""
return not any(target.iterdir())
def _rmdir_if_empty(target: Path) -> bool:
"""
Remove a directory if nothing is in it. False if something is.
The emptiness test and the removal are one call for the reason the caller
re-tests at all: the first test happened before a round trip to the
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()
return True
def _read_and_encrypt(
gek: bytes,
file_path: Path,
chunk_index: int,
file_hash: bytes,
file_id: str = "",
) -> dict:
"""Read one chunk off disk and encrypt it. Blocking; called through `off_disk`."""
with open(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)
|