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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
"""
A full disk is a stated refusal, not an exception.
Nothing on the upload path used to know about ENOSPC: a write that found no
room raised out of the handler, the catch-all answered "Request failed", and
the `.part` stayed behind holding the space that had run out. A client could
not tell that from any other fault, so the phone's nightly photo backup would
retry the same file every day. Now the node refuses with `disk_full` — up
front when the announced size cannot fit beside the reserve, and at the chunk
where a write actually fails — and keeps nothing it cannot finish.
"""
import errno
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.webrtc import upload_handlers
from meshbay_node.transport.webrtc.upload_handlers import DISK_RESERVE_BYTES
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from conftest import one_root, sealed_upload
GROUP = "g" * 32
def _peer(ctx: dict) -> WebRTCPeerSession:
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = ctx
session._group_id = GROUP
session._user_id = "user-1"
session._pk_user = ""
session.sent = []
session._send = session.sent.append
session.audited = []
session._audit = lambda *a, **k: session.audited.append(a)
return session
def _group_ctx(tmp_path) -> dict:
shared = tmp_path / "shared"
shared.mkdir(exist_ok=True)
return {"roots": one_root(shared),
"index": GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()),
"gek": generate_gek()}
def _codes(session):
return [m.get("code") for m in session.sent if m.get("type") == "error"]
def _free(monkeypatch, nbytes: int) -> None:
monkeypatch.setattr(upload_handlers, "_free_bytes", lambda _d: nbytes)
def _shared(ctx):
return ctx["roots"].roots[0].path
async def test_a_file_that_cannot_fit_is_refused_before_a_byte_is_written(tmp_path, monkeypatch):
ctx = _group_ctx(tmp_path)
peer = _peer(ctx)
_free(monkeypatch, DISK_RESERVE_BYTES + 10)
await peer._do_file_upload(sealed_upload(peer, filename="IMG_0001.jpg",
data=b"x" * 8, total_chunks=2))
assert _codes(peer) == ["disk_full"]
assert list(_shared(ctx).iterdir()) == [], "a refused upload left a file behind"
assert ("upload_refused", "disk_full") in peer.audited
async def test_the_reserve_is_never_handed_out(tmp_path, monkeypatch):
"""Exactly enough for the file, nothing for the reserve, is still a refusal:
a disk filled to the last byte breaks the node's own databases too."""
ctx = _group_ctx(tmp_path)
peer = _peer(ctx)
_free(monkeypatch, DISK_RESERVE_BYTES + 15)
await peer._do_file_upload(sealed_upload(peer, filename="a.jpg",
data=b"x" * 8, total_chunks=2))
assert _codes(peer) == ["disk_full"]
async def test_room_beside_the_reserve_is_accepted(tmp_path, monkeypatch):
ctx = _group_ctx(tmp_path)
peer = _peer(ctx)
_free(monkeypatch, DISK_RESERVE_BYTES + 16)
await peer._do_file_upload(sealed_upload(peer, filename="a.jpg",
data=b"x" * 8, total_chunks=2))
assert _codes(peer) == []
@pytest.mark.parametrize("err", [errno.ENOSPC, getattr(errno, "EDQUOT", errno.ENOSPC)])
async def test_a_write_that_finds_no_room_is_refused_and_cleaned_up(tmp_path, monkeypatch, err):
"""The check above can be passed by two uploads at once, or by a disk the
operator fills meanwhile. The write is the last word."""
ctx = _group_ctx(tmp_path)
peer = _peer(ctx)
_free(monkeypatch, 100 * DISK_RESERVE_BYTES)
await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"first",
chunk_index=0, total_chunks=2))
assert _codes(peer) == []
def no_room(*_a):
raise OSError(err, "No space left on device")
monkeypatch.setattr(upload_handlers, "_append_chunk", no_room)
await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"second",
chunk_index=1, total_chunks=2))
assert _codes(peer) == ["disk_full"]
assert list(_shared(ctx).iterdir()) == [], (
"the .part was kept, holding the very space that ran out")
assert ctx["partial_uploads"].get("user-1", "shared", "film.mkv") is None
async def test_another_write_failure_is_not_called_a_full_disk(tmp_path, monkeypatch):
ctx = _group_ctx(tmp_path)
peer = _peer(ctx)
_free(monkeypatch, 100 * DISK_RESERVE_BYTES)
def denied(*_a):
raise OSError(errno.EACCES, "Permission denied")
monkeypatch.setattr(upload_handlers, "_append_chunk", denied)
with pytest.raises(OSError):
await peer._do_file_upload(sealed_upload(peer, filename="a.jpg", data=b"x"))
async def test_the_real_free_space_is_what_is_read(tmp_path):
"""The un-patched path: a small file into a temp directory goes through."""
ctx = _group_ctx(tmp_path)
peer = _peer(ctx)
if upload_handlers._free_bytes(_shared(ctx)) < DISK_RESERVE_BYTES + 1024:
pytest.skip("this machine's temp filesystem is itself nearly full")
await peer._do_file_upload(sealed_upload(peer, filename="a.jpg", data=b"hello"))
assert _codes(peer) == []
assert (_shared(ctx) / "a.jpg").read_bytes() == b"hello"
|