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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
"""
Tasks nobody holds, across every package.
asyncio keeps only a **weak** reference to a task, so a coroutine started with
`asyncio.ensure_future(...)` whose result is thrown away can be collected while
it is still running. Nothing raises: the loop logs "Task was destroyed but it is
pending!" and whatever that task was doing simply stops.
The node already had a guard for this — written after an abandoned stream task
never reached `async with sem.__aexit__` and lost a transcode slot for good —
and it read one file, `webrtc_server.py`, because that is where the defect was
found. Meanwhile the hub answered `chat_notify` with a discarded
`ensure_future` (a notification for every member of a group, dropped whenever
the collector got there first), the node's indexer scheduled every real-time
index update the same way (a file copied in that never appears until a
reconciliation sweep an unbounded time later), and seventeen more sites did it
in `daemon.py`, `ops.py` and the loopback API — a reload, an enrichment pass, a
swarm registration.
So this one walks **every package's source**. A guard that stops at the edge of
the file where the bug was found is a guard against that bug, not against its
class.
"""
import ast
import asyncio
import logging
from pathlib import Path
import pytest
from meshbay_common import background
REPO = Path(__file__).resolve().parents[3]
PACKAGES = ("meshbay-common", "meshbay-hub", "meshbay-node")
SPAWNERS = {"ensure_future", "create_task"}
# `spawn()` is the implementation of the rule and is the one place allowed to
# call the primitive with nothing holding the result — that is what it is for.
ALLOWED = {Path("meshbay-common/src/meshbay_common/background.py")}
def _sources() -> list[Path]:
out: list[Path] = []
for pkg in PACKAGES:
src = REPO / "packages" / pkg / "src"
if src.is_dir():
out.extend(sorted(src.rglob("*.py")))
return out
def test_there_are_sources_to_check():
"""A rule applied to nothing passes forever."""
files = _sources()
assert len(files) > 20, f"only found {len(files)} source files — wrong root?"
def _discarded(tree: ast.AST) -> list[int]:
"""Lines where a task is started and its handle dropped on the floor.
An `Expr` statement is a call whose value goes nowhere; a `Lambda` whose
whole body is such a call is the same thing wearing a callback's clothes
(`signal.add_signal_handler(SIGHUP, lambda: ensure_future(reload()))`).
Anything assigned, returned, awaited or collected into a list is held by
its caller and is not this defect.
"""
found: list[int] = []
def is_spawn(node: ast.AST) -> bool:
return (isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr in SPAWNERS
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "asyncio")
for node in ast.walk(tree):
if isinstance(node, ast.Expr) and is_spawn(node.value):
found.append(node.lineno)
elif isinstance(node, ast.Lambda) and is_spawn(node.body):
found.append(node.lineno)
return found
def test_no_task_is_started_and_forgotten():
offenders: list[str] = []
for path in _sources():
rel = path.relative_to(REPO / "packages")
if rel in ALLOWED:
continue
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for line in _discarded(tree):
offenders.append(f"{rel}:{line}")
assert not offenders, (
"a task whose result is discarded can be collected mid-flight and the "
"work silently does not happen. Use meshbay_common.background.spawn(), "
"or keep the handle where the owner can also cancel it:\n "
+ "\n ".join(offenders))
# ── The helper itself ────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_spawn_holds_the_task_until_it_finishes():
started = asyncio.Event()
release = asyncio.Event()
async def work():
started.set()
await release.wait()
task = background.spawn(work())
await started.wait()
assert background.pending() == 1, "nothing is holding the task"
# The only strong reference the test itself holds, gone: if `spawn` is not
# keeping one, the task is now collectable.
del task
release.set()
await asyncio.sleep(0)
await asyncio.sleep(0)
assert background.pending() == 0, "the set grows for the life of the process"
@pytest.mark.asyncio
async def test_a_failure_is_logged_where_it_happened(caplog):
"""Nobody awaits this task, so the log is the only place it can surface."""
async def work():
raise RuntimeError("the background work failed")
with caplog.at_level(logging.ERROR, logger=background.log.name):
# Pinned rather than assumed: an earlier test in the run may have
# raised this logger's level or disabled it outright.
background.log.disabled = False
background.log.setLevel(logging.NOTSET)
task = background.spawn(work(), what="the work")
with pytest.raises(RuntimeError):
await task
assert any("the work" in r.getMessage() for r in caplog.records), caplog.text
@pytest.mark.asyncio
async def test_the_work_names_itself_without_a_label():
async def a_named_coroutine():
raise RuntimeError("boom")
task = background.spawn(a_named_coroutine())
with pytest.raises(RuntimeError):
await task
# The default label is the coroutine's own name, so no call site has to
# carry one for a log line to be readable.
assert "a_named_coroutine" in a_named_coroutine.__qualname__
|