Zo doen wijdat in oost

tests/unit/services/nodes/adapters.py

@@ -0,0 +1,91 @@
+# retoor <retoor@molodetz.nl>
+
+import json
+
+import pytest
+
+from devplacepy.services.nodes.adapters.base import AdapterError, clean, tag_for
+from devplacepy.services.nodes.adapters.direct import DevTo, GitHub, HackerNews, Mastodon, StackExchange
+from devplacepy.services.nodes.adapters.reddit import Reddit
+from devplacepy.services.nodes.adapters.registry import ADAPTERS
+from devplacepy.services.nodes.spec import SPEC
+
+
+def body(value) -> bytes:
+    return json.dumps(value).encode()
+
+
+def test_clean_strips_tags_entities_and_caps():
+    assert clean("<p>a &amp; b</p>  c", 50) == "a & b c"
+    assert len(clean("x" * 900, 100)) == 100
+    assert clean(None, 10) == ""
+
+
+def test_tag_for_picks_longest_word():
+    assert tag_for("sqlite wal mode") == "sqlite"
+    assert tag_for("!!") == ""
+
+
+def test_hackernews_parse():
+    data = {
+        "hits": [
+            {"objectID": "7", "title": "T", "points": 5, "num_comments": 2, "author": "a", "created_at": "2026"},
+            {"objectID": "", "title": "no id"},
+            "junk",
+        ]
+    }
+    hits = HackerNews().parse(body(data), 8)
+    assert [item.url for item in hits] == ["https://news.ycombinator.com/item?id=7"]
+    assert hits[0].score == 5 and hits[0].via == "direct"
+
+
+def test_stackexchange_parse():
+    data = {"items": [{"title": "Q &amp; A", "link": "https://stackoverflow.com/q/1", "score": 3, "answer_count": 2, "tags": ["c"], "owner": {"display_name": "bob"}}]}
+    hit = StackExchange().parse(body(data), 8)[0]
+    assert hit.title == "Q & A" and hit.author == "bob" and hit.comments == 2
+
+
+def test_github_parse_requires_http_url():
+    data = {"items": [{"title": "I", "html_url": "javascript:alert(1)"}, {"title": "J", "html_url": "https://github.com/a/b/issues/1", "user": {"login": "u"}}]}
+    hits = GitHub().parse(body(data), 8)
+    assert [item.title for item in hits] == ["J"]
+
+
+def test_devto_and_mastodon_need_a_tag():
+    with pytest.raises(AdapterError) as error:
+        DevTo().request("!!", 5)
+    assert error.value.code == "no_tag"
+    with pytest.raises(AdapterError):
+        Mastodon().request("", 5)
+    assert "tag=sqlite" in DevTo().request("sqlite", 5)
+    assert "/tag/sqlite" in Mastodon().request("sqlite", 5)
+
+
+def test_mastodon_parse_strips_html():
+    data = [{"url": "https://m.example/1", "content": "<p>hello <b>world</b></p>", "account": {"acct": "x@y"}, "favourites_count": 2, "reblogs_count": 1}]
+    hit = Mastodon().parse(body(data), 5)[0]
+    assert hit.snippet == "hello world" and hit.score == 3
+
+
+def test_reddit_parse_and_via():
+    data = {"data": {"children": [{"data": {"title": "T", "permalink": "/r/x/comments/1/t/", "score": 9, "num_comments": 4, "author": "a", "selftext": "body"}}, {"data": {"title": "bad", "permalink": "http://evil"}}]}}
+    hits = Reddit().parse(body(data), 5)
+    assert len(hits) == 1 and hits[0].url == "https://www.reddit.com/r/x/comments/1/t/" and hits[0].via == "fetch"
+
+
+@pytest.mark.parametrize("adapter", list(ADAPTERS.values()))
+def test_forged_structure_is_rejected(adapter):
+    for forged in (b"not json", body("string"), body({"unexpected": 1}), body(42)):
+        with pytest.raises(AdapterError):
+            adapter.parse(forged, 5)
+
+
+def test_oversize_body_is_rejected():
+    with pytest.raises(AdapterError) as error:
+        HackerNews().parse(b" " * (2 * 1024 * 1024 + 1), 5)
+    assert error.value.code == "body_too_large"
+
+
+def test_node_adapters_are_exactly_the_spec_adapters():
+    node_backed = {name for name, adapter in ADAPTERS.items() if not adapter.direct}
+    assert node_backed == {item.adapter for item in SPEC.destinations}

tests/unit/services/nodes/cache.py

@@ -0,0 +1,109 @@
+# retoor <retoor@molodetz.nl>
+
+import random
+
+from devplacepy.services.nodes.cache import ResultCache, cache_key
+from devplacepy.services.nodes.spec import SPEC
+
+LIMITS = SPEC.limits
+
+
+class Clock:
+    def __init__(self):
+        self.now = 100.0
+
+    def __call__(self):
+        return self.now
+
+
+def make(**kwargs):
+    clock = Clock()
+    return ResultCache(clock=clock, **kwargs), clock
+
+
+def test_hit_and_miss_counters():
+    cache, _ = make()
+    assert cache.get("GET", "u") is None
+    assert cache.put("GET", "u", 200, {"a": "b"}, b"xyz", False)
+    hit = cache.get("GET", "u")
+    assert hit.body == b"xyz" and hit.headers == {"a": "b"}
+    assert (cache.hits, cache.misses) == (1, 1)
+
+
+def test_method_is_part_of_the_key():
+    assert cache_key("GET", "u") != cache_key("HEAD", "u")
+    assert cache_key("get", "u") == cache_key("GET", "u")
+
+
+def test_entries_expire_after_the_ttl():
+    cache, clock = make(ttl_seconds=10)
+    cache.put("GET", "u", 200, {}, b"x", False)
+    clock.now += 9.9
+    assert cache.get("GET", "u") is not None
+    clock.now += 0.2
+    assert cache.get("GET", "u") is None
+    assert len(cache) == 0 and cache.size_bytes == 0
+
+
+def test_ttl_is_clamped_to_the_spec_ceiling_and_zero_disables():
+    assert ResultCache(ttl_seconds=10**9).ttl_seconds == LIMITS.cache_ttl_seconds
+    off, _ = make(ttl_seconds=0)
+    assert off.put("GET", "u", 200, {}, b"x", False) is False
+
+
+def test_errors_and_truncated_bodies_are_never_cached():
+    cache, _ = make()
+    assert not cache.put("GET", "u", 500, {}, b"x", False)
+    assert not cache.put("GET", "u", 503, {}, b"x", False)
+    assert not cache.put("GET", "u", 200, {}, b"x", True)
+    assert cache.put("GET", "u", 404, {}, b"x", False)
+
+
+def test_an_oversized_entry_is_refused():
+    cache, _ = make(entry_max_bytes=10, max_bytes=100)
+    assert not cache.put("GET", "u", 200, {}, b"x" * 11, False)
+    assert cache.put("GET", "u", 200, {}, b"x" * 10, False)
+
+
+def test_the_entry_cap_cannot_exceed_the_spec_ceiling():
+    assert ResultCache(entry_max_bytes=10**12).entry_max_bytes == LIMITS.cache_entry_max_bytes
+
+
+def test_least_recently_used_goes_first():
+    cache, _ = make(max_bytes=30, entry_max_bytes=30)
+    for name in "abc":
+        cache.put("GET", name, 200, {}, b"x" * 10, False)
+    cache.get("GET", "a")
+    cache.put("GET", "d", 200, {}, b"x" * 10, False)
+    assert cache.get("GET", "b") is None
+    assert cache.get("GET", "a") is not None
+    assert cache.get("GET", "c") is not None
+    assert cache.get("GET", "d") is not None
+
+
+def test_replacing_an_entry_does_not_leak_bytes():
+    cache, _ = make()
+    cache.put("GET", "u", 200, {}, b"x" * 100, False)
+    cache.put("GET", "u", 200, {}, b"y" * 40, False)
+    assert cache.size_bytes == 40 and len(cache) == 1
+
+
+def test_size_never_exceeds_the_cap_under_a_random_workload():
+    cache, clock = make(max_bytes=1000, entry_max_bytes=400, ttl_seconds=50)
+    rng = random.Random(7)
+    for step in range(5000):
+        clock.now += rng.random()
+        key = f"k{rng.randint(0, 60)}"
+        if rng.random() < 0.7:
+            cache.put("GET", key, rng.choice([200, 404, 500]), {}, b"x" * rng.randint(0, 500), False)
+        else:
+            cache.get("GET", key)
+        assert 0 <= cache.size_bytes <= 1000
+        assert cache.size_bytes == sum(len(e.body) for e in cache._entries.values())
+
+
+def test_clear_empties_everything():
+    cache, _ = make()
+    cache.put("GET", "u", 200, {}, b"x", False)
+    cache.clear()
+    assert len(cache) == 0 and cache.size_bytes == 0

tests/unit/services/nodes/circuit.py

@@ -0,0 +1,97 @@
+# retoor <retoor@molodetz.nl>
+
+from devplacepy.services.nodes.circuit import CLOSED, HALF_OPEN, OPEN, CircuitBreaker
+
+
+class Clock:
+    def __init__(self):
+        self.now = 0.0
+
+    def __call__(self):
+        return self.now
+
+
+def make(failures=3, open_seconds=60):
+    clock = Clock()
+    return CircuitBreaker(failures, open_seconds, clock), clock
+
+
+def test_unknown_keys_are_closed():
+    breaker, _ = make()
+    assert breaker.state("x") == CLOSED and breaker.allows("x")
+
+
+def test_it_opens_at_the_threshold_and_not_before():
+    breaker, _ = make(failures=3)
+    breaker.failure("x")
+    breaker.failure("x")
+    assert breaker.allows("x")
+    breaker.failure("x")
+    assert breaker.state("x") == OPEN and not breaker.allows("x")
+
+
+def test_a_success_resets_the_streak():
+    breaker, _ = make(failures=3)
+    breaker.failure("x")
+    breaker.failure("x")
+    breaker.success("x")
+    breaker.failure("x")
+    breaker.failure("x")
+    assert breaker.allows("x")
+
+
+def test_it_goes_half_open_after_the_cooldown_and_admits_one_probe():
+    breaker, clock = make(failures=1, open_seconds=60)
+    breaker.failure("x")
+    clock.now = 59
+    assert breaker.state("x") == OPEN
+    clock.now = 60
+    assert breaker.state("x") == HALF_OPEN and breaker.allows("x")
+    breaker.begin("x")
+    assert not breaker.allows("x")
+
+
+def test_a_failed_probe_reopens_for_a_full_cooldown():
+    breaker, clock = make(failures=1, open_seconds=60)
+    breaker.failure("x")
+    clock.now = 70
+    breaker.begin("x")
+    breaker.failure("x")
+    assert breaker.state("x") == OPEN
+    clock.now = 129
+    assert breaker.state("x") == OPEN
+    clock.now = 131
+    assert breaker.state("x") == HALF_OPEN
+
+
+def test_a_successful_probe_closes_it():
+    breaker, clock = make(failures=1)
+    breaker.failure("x")
+    clock.now = 100
+    breaker.begin("x")
+    breaker.success("x")
+    assert breaker.state("x") == CLOSED and breaker.allows("x")
+
+
+def test_keys_are_independent():
+    breaker, _ = make(failures=1)
+    breaker.failure(("node", "a"))
+    assert not breaker.allows(("node", "a"))
+    assert breaker.allows(("node", "b"))
+    assert breaker.allows(("destination", "a"))
+
+
+def test_open_count_and_forget():
+    breaker, _ = make(failures=1)
+    breaker.failure("a")
+    breaker.failure("b")
+    assert breaker.open_count() == 2
+    breaker.forget("a")
+    assert breaker.open_count() == 1
+
+
+def test_a_zero_threshold_still_needs_one_failure():
+    breaker = CircuitBreaker(failures=0)
+    assert breaker.allows("x")
+    breaker.failure("x")
+    assert not breaker.allows("x")

tests/unit/services/nodes/dispatcher.py

@@ -0,0 +1,415 @@
+# retoor <retoor@molodetz.nl>
+
+import base64
+import json
+import random
+import time
+
+import pytest
+
+from devplacepy.database import get_table
+from devplacepy.database import nodes as store
+from devplacepy.services.nodes.broker import BodyError, JobBroker, clean_headers, decode_body
+from devplacepy.services.nodes.circuit import CircuitBreaker
+from devplacepy.services.nodes.dispatcher import JOB_HEADERS, NodeDispatcher
+from devplacepy.services.nodes.ledger import UsageLedger
+from devplacepy.services.nodes.policy import DestinationPolicy
+from devplacepy.services.nodes.registry import NodeCounters, NodeRegistry
+from devplacepy.services.nodes.scheduler import MEGABYTE, NodeScheduler
+from devplacepy.services.nodes.session import NodeSession
+from devplacepy.services.nodes.spec import SPEC
+from devplacepy.utils import generate_uid
+from tests.conftest import run_async
+from tests.unit.services.nodes.scheduler import Clock, node, settings
+
+LIMITS = SPEC.limits
+TARGET_URL = "https://www.reddit.com/r/python.json"
+
+
+class Rig:
+    def __init__(self):
+        self.clock = Clock()
+        self.counters = NodeCounters()
+        self.registry = NodeRegistry(self.counters)
+        self.breaker = CircuitBreaker(clock=self.clock)
+        self.scheduler = NodeScheduler(self.registry, self.breaker, self.clock, rng=random.Random(2))
+        self.broker = JobBroker()
+        self.dispatcher = NodeDispatcher(
+            self.registry, self.scheduler, self.broker, self.breaker, UsageLedger(), self.counters,
+            clock=self.clock,
+        )
+
+    def connect(self, **overrides):
+        owner = generate_uid()
+        device, _ = store.create_device(owner, "d", "native")
+        session = NodeSession(
+            node().websocket, {"uid": device["uid"], "user_uid": owner, "kind": "native"}
+        )
+        session.hello = {"modes": ["fetch"]}
+        session.settings = settings(**overrides)
+        session.roll_day()
+        self.dispatcher.attach(session)
+        self.registry.admit(session)
+        return session
+
+    def enqueue(self, url=TARGET_URL, caller="c1", kind="user", mode="any"):
+        target = DestinationPolicy().check(url, "GET", mode)
+        return self.broker.enqueue(kind, caller, target, mode)
+
+    def tick(self):
+        return run_async(self.dispatcher.tick())
+
+    def answer(self, session, uid, **fields):
+        frame = {"type": "job_result", "job_id": uid, "status": 200, "body_b64": base64.b64encode(b"hello").decode()}
+        frame.update(fields)
+        run_async(session._handle(frame))
+
+
+@pytest.fixture
+def rig(local_db):
+    get_table("node_requests").delete()
+    yield Rig()
+    get_table("node_requests").delete()
+
+
+def sent_jobs(session):
+    return [frame for frame in session.websocket.sent if frame["type"] == "job"]
+
+
+def test_a_queued_request_becomes_a_job_frame(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    assert rig.tick() == 1
+    job = sent_jobs(session)[0]
+    assert job == {"type": "job", "job_id": uid, "method": "GET", "url": TARGET_URL, "headers": JOB_HEADERS, "timeout": LIMITS.job_timeout_seconds}
+    assert store.get_request(uid)["status"] == "assigned"
+    assert store.get_request(uid)["device_uid"] == session.device_uid
+    assert uid in session.inflight
+
+
+def test_the_result_completes_the_request_and_is_metered(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    rig.answer(session, uid, headers={"Content-Type": "application/json", "Set-Cookie": "x=1"})
+    result = rig.broker.read(uid)
+    assert result.ok and result.http_status == 200 and result.body == b"hello"
+    assert result.headers == {"content-type": "application/json"}
+    assert uid not in session.inflight
+    today = store.usage_today(session.device_uid)
+    assert today["requests_ok"] == 1 and today["bytes_down"] == 5 and today["bytes_up"] > len(TARGET_URL)
+    assert store.get_donor(session.user_uid)["requests_ok"] == 1
+    assert session.used_today == today["bytes_up"] + today["bytes_down"]
+    assert rig.counters.snapshot()["jobs_done"] == 1
+
+
+def test_no_eligible_node_leaves_the_request_queued(rig):
+    uid = rig.enqueue()
+    assert rig.tick() == 0
+    assert store.get_request(uid)["status"] == "queued"
+    rig.connect(allowed_platforms=[])
+    assert rig.tick() == 0
+    assert store.get_request(uid)["status"] == "queued"
+
+
+def test_one_request_goes_to_exactly_one_node(rig):
+    sessions = [rig.connect() for _ in range(4)]
+    rig.enqueue()
+    assert rig.tick() == 1
+    assert sum(len(sent_jobs(item)) for item in sessions) == 1
+
+
+def test_a_caller_never_gets_its_own_node(rig):
+    session = rig.connect()
+    uid = rig.enqueue(caller=session.user_uid)
+    assert rig.tick() == 0
+    assert store.get_request(uid)["status"] == "queued"
+    guest = rig.enqueue(caller=session.user_uid, kind="guest")
+    assert rig.tick() == 1 and guest in session.inflight
+
+
+def test_a_policy_failure_at_dispatch_time_fails_the_request(rig):
+    rig.connect()
+    uid = store.enqueue_request("user", "c", "https://evil.example/", "GET", "x", "any")
+    rig.tick()
+    row = store.get_request(uid)
+    assert row["status"] == "failed" and row["error"] == "host_not_allowed"
+    assert rig.counters.snapshot()["jobs_refused"] == 1
+
+
+def test_a_job_error_fails_the_request_and_counts_against_the_node(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    run_async(session._handle({"type": "job_error", "job_id": uid, "code": "timeout"}))
+    assert store.get_request(uid)["error"] == "timeout"
+    assert store.usage_today(session.device_uid)["requests_failed"] == 1
+    assert rig.counters.snapshot()["jobs_failed"] == 1
+
+
+def test_five_node_failures_open_that_nodes_circuit(rig):
+    session = rig.connect()
+    for _ in range(LIMITS.breaker_failures):
+        uid = rig.enqueue()
+        session.host_started.clear()
+        rig.scheduler._destination_started.clear()
+        rig.tick()
+        run_async(session._handle({"type": "job_error", "job_id": uid, "code": "network"}))
+    follow = rig.enqueue()
+    session.host_started.clear()
+    rig.scheduler._destination_started.clear()
+    assert rig.tick() == 0
+    assert store.get_request(follow)["status"] == "queued"
+    rig.clock.now += LIMITS.breaker_open_seconds
+    assert rig.tick() == 1
+
+
+def test_donor_limit_codes_pause_the_node_without_blaming_it(rig):
+    session = rig.connect(bandwidth_mb_day=50)
+    uid = rig.enqueue()
+    rig.tick()
+    run_async(session._handle({"type": "job_error", "job_id": uid, "code": "bandwidth_exhausted"}))
+    assert session.used_today == 50 * MEGABYTE
+    assert rig.breaker.allows(("node", session.device_uid))
+    other = rig.connect()
+    rig.clock.now += 10
+    uid = rig.enqueue()
+    assert rig.tick() == 1
+    run_async(other._handle({"type": "job_error", "job_id": uid, "code": "outside_hours"}))
+    assert other.paused_until > rig.clock.now
+    assert rig.breaker.allows(("node", other.device_uid))
+
+
+def test_a_donor_refusal_of_an_allowed_destination_raises_a_flag(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    run_async(session._handle({"type": "job_error", "job_id": uid, "code": "host_not_allowed"}))
+    kinds = [flag["kind"] for flag in store.list_flags()]
+    assert "refusal_mismatch" in kinds
+    assert rig.breaker.allows(("node", session.device_uid))
+
+
+@pytest.mark.parametrize(
+    "fields,code",
+    [
+        ({"status": 99}, "bad_status"),
+        ({"status": 600}, "bad_status"),
+        ({"body_b64": "***not base64***"}, "bad_base64"),
+        ({"body_b64": "abc"}, "bad_base64"),
+    ],
+)
+def test_forged_results_are_integrity_failures(rig, fields, code):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    rig.answer(session, uid, **fields)
+    assert store.get_request(uid)["error"] == code
+    assert store.usage_today(session.device_uid)["integrity_failures"] == 1
+    assert rig.counters.snapshot()["integrity_failures"] == 1
+    assert any(flag["kind"] == "integrity" for flag in store.list_flags())
+    assert store.get_donor(session.user_uid)["requests_ok"] == 0
+
+
+def test_an_oversized_body_is_truncated_and_flagged(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    big = base64.b64encode(b"x" * (LIMITS.max_body_bytes + 10)).decode()
+    rig.answer(session, uid, body_b64=big)
+    result = rig.broker.read(uid)
+    assert result.truncated and len(result.body) == LIMITS.max_body_bytes
+    assert rig.counters.snapshot()["results_truncated"] == 1
+
+
+def test_a_donor_declared_truncation_is_kept(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    rig.answer(session, uid, truncated=True)
+    assert rig.broker.read(uid).truncated
+
+
+def test_a_result_for_an_unknown_job_is_a_protocol_error(rig):
+    session = rig.connect()
+    from devplacepy.services.nodes.frames import FrameError
+
+    with pytest.raises(FrameError):
+        rig.answer(session, "never-assigned")
+
+
+def test_a_job_cannot_be_answered_twice(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    rig.answer(session, uid)
+    from devplacepy.services.nodes.frames import FrameError
+
+    with pytest.raises(FrameError):
+        rig.answer(session, uid)
+    assert store.get_donor(session.user_uid)["requests_ok"] == 1
+
+
+def test_rate_limited_and_server_error_results_open_the_destination_circuit(rig):
+    for index in range(LIMITS.breaker_failures):
+        session = rig.connect()
+        uid = rig.enqueue()
+        rig.scheduler._destination_started.clear()
+        rig.tick()
+        rig.answer(session, uid, status=429 if index % 2 else 503)
+        assert rig.broker.read(uid).ok
+        rig.registry.release(session)
+    assert not rig.breaker.allows(("destination", "reddit.com"))
+    rig.connect()
+    rig.enqueue()
+    rig.scheduler._destination_started.clear()
+    assert rig.tick() == 0
+
+
+def test_a_healthy_result_closes_the_destination_circuit(rig):
+    rig.breaker.failure(("destination", "reddit.com"))
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    rig.answer(session, uid, status=200)
+    assert rig.breaker.state(("destination", "reddit.com")) == "closed"
+
+
+def test_a_job_that_never_answers_times_out(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    rig.clock.now += LIMITS.job_timeout_seconds + LIMITS.job_grace_seconds + 1
+    assert run_async(rig.dispatcher.sweep_inflight()) == 1
+    assert store.get_request(uid)["error"] == "timeout"
+    assert session.inflight == {}
+
+
+def test_a_disconnect_fails_every_inflight_job(rig):
+    session = rig.connect(max_concurrent=2)
+    uids = [rig.enqueue(), rig.enqueue()]
+    for _ in uids:
+        session.host_started.clear()
+        rig.scheduler._destination_started.clear()
+        assert rig.tick() == 1
+    assert len(session.inflight) == 2
+    assert rig.dispatcher.release_session(session) == 2
+    for uid in uids:
+        row = store.get_request(uid)
+        assert row["status"] == "failed" and row["error"] == "node_disconnected"
+    assert session.inflight == {}
+
+
+def test_release_session_marks_assigned_jobs_as_node_disconnected(rig):
+    session = rig.connect()
+    uid = rig.enqueue()
+    rig.tick()
+    assert rig.dispatcher.release_session(session) == 1
+    assert store.get_request(uid)["error"] == "node_disconnected"
+    assert rig.dispatcher.release_session(session) == 0
+
+
+def test_a_failed_send_releases_the_job(rig):
+    session = rig.connect()
+
+    async def broken(*args, **kwargs):
+        raise ConnectionError("gone")
+
+    session.send = broken
+    uid = rig.enqueue()
+    assert rig.tick() == 0
+    assert store.get_request(uid)["error"] == "send_failed"
+    assert session.inflight == {}
+
+
+def test_an_idle_tick_does_no_writes_and_no_work(rig):
+    rig.connect()
+    started = time.perf_counter()
+    for _ in range(50):
+        assert rig.tick() == 0
+    assert (time.perf_counter() - started) / 50 < 0.02
+
+
+def test_the_cap_refusing_the_ledger_pauses_the_node(rig):
+    session = rig.connect(bandwidth_mb_day=1)
+    store.save_node_settings(session.user_uid, {"enabled": True, "bandwidth_mb_day": 1})
+    session.used_today = 0
+    uid = rig.enqueue()
+    rig.tick()
+    store.record_job(session.device_uid, session.user_uid, ok=True, bytes_down=MEGABYTE, cap_bytes=MEGABYTE)
+    rig.answer(session, uid)
+    assert rig.broker.read(uid).ok
+    assert session.used_today >= MEGABYTE
+
+
+def test_every_assigned_job_is_answered_at_most_once_under_a_random_workload(rig):
+    rng = random.Random(11)
+    sessions = [rig.connect(max_concurrent=2) for _ in range(4)]
+    seen = set()
+    for _ in range(150):
+        rig.enqueue(caller=generate_uid())
+        rig.clock.now += 10
+        rig.scheduler._destination_started.clear()
+        rig.tick()
+        for session in sessions:
+            for job_id in list(session.inflight):
+                assert job_id not in seen
+                action = rng.random()
+                if action < 0.6:
+                    rig.answer(session, job_id)
+                    seen.add(job_id)
+                elif action < 0.8:
+                    run_async(session._handle({"type": "job_error", "job_id": job_id, "code": "network"}))
+                    seen.add(job_id)
+            assert len(session.inflight) <= session.max_concurrent
+    open_now = {job for item in sessions for job in item.inflight}
+    assert not (open_now & seen)
+
+
+def test_body_helpers():
+    assert decode_body(base64.b64encode(b"abc").decode()) == (b"abc", False)
+    with pytest.raises(BodyError):
+        decode_body("!!!")
+    assert clean_headers({"A": "1", "set-cookie": "s", "Authorization": "x", 5: "x", "b": 5}) == {"a": "1"}
+    assert clean_headers("nope") == {}
+    many = {f"h{i}": "v" for i in range(500)}
+    assert len(clean_headers(many)) == 64
+    assert len(next(iter(clean_headers({"k": "v" * 5000}).values()))) == 1024
+    assert json.loads(json.dumps(clean_headers({"A": "1"}))) == {"a": "1"}
+
+
+def test_a_saturated_destination_is_scanned_once_per_tick_not_once_per_request(rig, monkeypatch):
+    rig.connect(allowed_platforms=[])
+    for index in range(30):
+        rig.enqueue(caller=f"c{index}")
+    scans = []
+    original = rig.scheduler.candidates
+    monkeypatch.setattr(rig.scheduler, "candidates", lambda *a: scans.append(a) or original(*a))
+    assert rig.tick() == 0
+    assert len(scans) == 1
+
+
+def test_saturation_does_not_hide_other_destinations(rig):
+    session = rig.connect()
+    for index in range(3):
+        rig.enqueue(caller=f"c{index}")
+    rig.scheduler._destination_started["reddit.com"] = rig.clock.now
+    assert rig.tick() == 0
+    rig.clock.now += 10
+    assert rig.tick() == 1
+    assert len(session.inflight) == 1
+
+
+def test_the_owner_exclusion_does_not_saturate_the_destination(rig):
+    mine = rig.connect()
+    other = rig.connect()
+    first = rig.enqueue(caller=mine.user_uid)
+    second = rig.enqueue(caller=other.user_uid)
+    rig.clock.now += 10
+    rig.tick()
+    assert first in other.inflight
+    rig.clock.now += 10
+    rig.scheduler._destination_started.clear()
+    other.host_started.clear()
+    rig.tick()
+    assert second in mine.inflight

tests/unit/services/nodes/fabric.py

@@ -0,0 +1,231 @@
+# retoor <retoor@molodetz.nl>
+
+import json
+from datetime import datetime, timezone
+
+import pytest
+
+from devplacepy.database import get_table, set_setting
+from devplacepy.database import nodes as store
+from devplacepy.services.nodes.broker import BrokerResult, JobBroker
+from devplacepy.services.nodes.cache import ResultCache
+from devplacepy.services.nodes.fabric import (
+    FabricQuotaExceeded,
+    FabricRefused,
+    FabricUnavailable,
+    NodeFabric,
+)
+from devplacepy.services.nodes.spec import SPEC
+from devplacepy.utils import generate_uid
+from tests.conftest import run_async
+
+LIMITS = SPEC.limits
+URL = "https://www.reddit.com/r/python.json"
+
+
+class ScriptedBroker(JobBroker):
+    def __init__(self, outcomes):
+        self.outcomes = list(outcomes)
+        self.enqueued = 0
+
+    def enqueue(self, caller_kind, caller_id, target, mode):
+        self.enqueued += 1
+        return f"uid-{self.enqueued}"
+
+    async def wait(self, uid, timeout):
+        outcome = self.outcomes.pop(0)
+        if outcome == "ok":
+            return BrokerResult(uid, "done", 200, {"content-type": "text/plain"}, b"payload", False, "")
+        return BrokerResult(uid, "failed", 0, {}, b"", False, outcome)
+
+
+def heartbeat(connected):
+    table = get_table("service_state")
+    record = {
+        "name": "nodes",
+        "status": "running",
+        "heartbeat": datetime.now(timezone.utc).isoformat(),
+        "metrics": json.dumps({"stats": [{"label": "connected", "value": connected}]}),
+    }
+    existing = table.find_one(name="nodes")
+    if existing:
+        table.update({**record, "id": existing["id"]}, ["id"])
+    else:
+        table.insert(record)
+
+
+@pytest.fixture
+def online(local_db):
+    set_setting("nodes_enabled", "1")
+    set_setting("nodes_daily_quota_user", "1000")
+    set_setting("nodes_daily_quota_guest", "0")
+    heartbeat(3)
+    yield
+    set_setting("nodes_enabled", "0")
+    get_table("service_state").delete(name="nodes")
+
+
+def fetch(fabric, **kwargs):
+    kwargs.setdefault("caller_kind", "user")
+    kwargs.setdefault("caller_id", generate_uid())
+    return run_async(fabric.fetch(URL, **kwargs))
+
+
+def test_a_node_fetch_returns_the_body_and_caches_it(online):
+    broker = ScriptedBroker(["ok"])
+    fabric = NodeFabric(broker=broker)
+    first = fetch(fabric, caller_id="c")
+    assert (first.status, first.body, first.via) == (200, b"payload", "fetch")
+    second = fetch(fabric, caller_id="c")
+    assert second.via == "cache" and second.body == b"payload" and second.elapsed_ms == 0
+    assert broker.enqueued == 1
+
+
+def test_use_cache_false_always_asks_a_node(online):
+    broker = ScriptedBroker(["ok", "ok"])
+    fabric = NodeFabric(broker=broker)
+    fetch(fabric, caller_id="c", use_cache=False)
+    fetch(fabric, caller_id="c", use_cache=False)
+    assert broker.enqueued == 2
+
+
+def test_policy_refusals_surface_with_their_code(online):
+    fabric = NodeFabric(broker=ScriptedBroker([]))
+    for url, code in (("https://example.com/", "host_not_allowed"), ("https://reddit.com:22/", "port_not_allowed")):
+        with pytest.raises(FabricRefused) as info:
+            run_async(fabric.fetch(url, caller_kind="user", caller_id="c"))
+        assert info.value.code == code
+
+
+def test_the_kill_switch_and_missing_nodes_are_unavailable(online):
+    fabric = NodeFabric(broker=ScriptedBroker([]))
+    heartbeat(0)
+    with pytest.raises(FabricUnavailable) as info:
+        fetch(fabric)
+    assert info.value.code == "no_nodes"
+    heartbeat(2)
+    set_setting("nodes_enabled", "0")
+    with pytest.raises(FabricUnavailable) as info:
+        fetch(fabric)
+    assert info.value.code == "kill_switch"
+
+
+def test_a_stale_heartbeat_counts_as_no_nodes(online):
+    table = get_table("service_state")
+    row = table.find_one(name="nodes")
+    table.update({"id": row["id"], "heartbeat": "2000-01-01T00:00:00+00:00"}, ["id"])
+    assert NodeFabric(broker=ScriptedBroker([])).available().nodes_online == 0
+
+
+def test_no_service_row_means_no_nodes(online):
+    get_table("service_state").delete(name="nodes")
+    assert NodeFabric().available().nodes_online == 0
+
+
+def test_quota_is_enforced_before_any_node_is_touched(online):
+    set_setting("nodes_daily_quota_user", "2")
+    broker = ScriptedBroker(["ok"] * 5)
+    fabric = NodeFabric(broker=broker)
+    caller = generate_uid()
+    for _ in range(2):
+        fetch(fabric, caller_id=caller, use_cache=False)
+    with pytest.raises(FabricQuotaExceeded):
+        fetch(fabric, caller_id=caller, use_cache=False)
+    assert broker.enqueued == 2
+
+
+def test_guests_have_no_quota_by_default(online):
+    broker = ScriptedBroker(["ok"])
+    with pytest.raises(FabricQuotaExceeded):
+        fetch(NodeFabric(broker=broker), caller_kind="guest", caller_id="g")
+    assert broker.enqueued == 0
+
+
+def test_a_cache_hit_costs_no_quota(online):
+    set_setting("nodes_daily_quota_user", "1")
+    fabric = NodeFabric(broker=ScriptedBroker(["ok"]))
+    caller = generate_uid()
+    fetch(fabric, caller_id=caller)
+    for _ in range(5):
+        assert fetch(fabric, caller_id=caller).via == "cache"
+
+
+def test_a_retryable_failure_is_retried_once_on_another_node(online):
+    broker = ScriptedBroker(["node_disconnected", "ok"])
+    result = fetch(NodeFabric(broker=broker))
+    assert result.body == b"payload" and broker.enqueued == 2
+
+
+def test_the_retry_does_not_spend_a_second_quota_unit(online):
+    set_setting("nodes_daily_quota_user", "1")
+    caller = generate_uid()
+    fetch(NodeFabric(broker=ScriptedBroker(["timeout", "ok"])), caller_id=caller)
+    assert store.caller_requests_today("user", caller) == 1
+
+
+def test_retries_stop_after_the_attempt_cap(online):
+    broker = ScriptedBroker(["timeout"] * 5)
+    with pytest.raises(FabricUnavailable) as info:
+        fetch(NodeFabric(broker=broker))
+    assert info.value.code == "timeout" and broker.enqueued == 2
+
+
+def test_a_non_retryable_failure_is_not_retried(online):
+    broker = ScriptedBroker(["bad_status", "ok"])
+    with pytest.raises(FabricUnavailable):
+        fetch(NodeFabric(broker=broker))
+    assert broker.enqueued == 1
+
+
+def test_a_full_queue_is_a_clean_unavailable(online):
+    class Full(ScriptedBroker):
+        def enqueue(self, *args):
+            raise store.QueueFull("full")
+
+    with pytest.raises(FabricUnavailable) as info:
+        fetch(NodeFabric(broker=Full([])))
+    assert info.value.code == "queue_full"
+
+
+def test_the_wait_is_capped_at_the_spec_maximum(online):
+    seen = []
+
+    class Spy(ScriptedBroker):
+        async def wait(self, uid, timeout):
+            seen.append(timeout)
+            return await super().wait(uid, timeout)
+
+    fetch(NodeFabric(broker=Spy(["ok"])), timeout=10**6)
+    assert seen[0] <= LIMITS.max_wait_seconds
+
+
+def test_the_cache_ttl_follows_the_admin_setting_up_to_the_ceiling(online):
+    cache = ResultCache()
+    fabric = NodeFabric(broker=ScriptedBroker(["ok"]), cache=cache)
+    set_setting("nodes_cache_ttl_seconds", "0")
+    fetch(fabric, caller_id="c")
+    assert len(cache) == 0
+    set_setting("nodes_cache_ttl_seconds", str(10**9))
+    fabric._sync_cache_ttl()
+    assert cache.ttl_seconds == LIMITS.cache_ttl_seconds
+    set_setting("nodes_cache_ttl_seconds", str(LIMITS.cache_ttl_seconds))
+
+
+def test_can_serve_needs_policy_switch_and_a_node(online):
+    fabric = NodeFabric(broker=ScriptedBroker([]))
+    assert fabric.can_serve(URL)
+    assert not fabric.can_serve("https://example.com/")
+    assert not fabric.can_serve(URL, "POST")
+    heartbeat(0)
+    assert not fabric.can_serve(URL)
+
+
+def test_the_real_broker_times_out_a_request_nobody_picks_up(online):
+    get_table("node_requests").delete()
+    fabric = NodeFabric()
+    with pytest.raises(FabricUnavailable) as info:
+        fetch(fabric, timeout=0.3)
+    assert info.value.code in ("caller_timeout", "failed")
+    rows = list(get_table("node_requests").find())
+    assert rows and all(row["status"] == "failed" for row in rows)
+    get_table("node_requests").delete()

tests/unit/services/nodes/ledger.py

@@ -0,0 +1,55 @@
+# retoor <retoor@molodetz.nl>
+
+from devplacepy.database import get_table
+from devplacepy.database import nodes as store
+from devplacepy.services.nodes import ledger as ledger_module
+from devplacepy.services.nodes.ledger import FIRST_BADGE, GIGABYTE, UsageLedger
+from devplacepy.utils import generate_uid, register_account
+
+
+def member(local_db):
+    name = "ledger" + generate_uid()[-10:]
+    uid, _, _ = register_account(name, f"{name}@t.dev", "secret123", age_band="adult", accepted_terms=True)
+    return uid
+
+
+def badges(uid):
+    return {row["badge_name"] for row in get_table("badges").find(user_uid=uid)}
+
+
+def test_first_success_awards_xp_and_the_first_badge(local_db):
+    uid = member(local_db)
+    before = get_table("users").find_one(uid=uid).get("xp") or 0
+    store.record_job(generate_uid(), uid, ok=True, bytes_up=10, bytes_down=100)
+    UsageLedger().reward(uid)
+    assert FIRST_BADGE in badges(uid)
+    assert (get_table("users").find_one(uid=uid).get("xp") or 0) > before
+
+
+def test_xp_stops_at_the_daily_cap_but_not_the_counters(local_db):
+    uid = member(local_db)
+    ledger = UsageLedger()
+    store.record_job(generate_uid(), uid, ok=True, bytes_up=1, bytes_down=1)
+    granted = 0
+    for _ in range(store.LIMITS.xp_daily_cap + 10):
+        granted += int(ledger.claim_xp(uid, ledger_module.XP_PER_JOB))
+    assert granted == store.LIMITS.xp_daily_cap
+
+
+def test_byte_badges_follow_lifetime_totals(local_db):
+    uid = member(local_db)
+    store.record_job(
+        generate_uid(), uid, ok=True, bytes_up=0, bytes_down=GIGABYTE, cap_bytes=2 * GIGABYTE
+    )
+    UsageLedger().reward(uid)
+    assert {"Gigabyte Donor"} <= badges(uid)
+    assert "10 GB Donor" not in badges(uid)
+
+
+def test_a_second_reward_call_does_not_duplicate_badges(local_db):
+    uid = member(local_db)
+    store.record_job(generate_uid(), uid, ok=True, bytes_up=1, bytes_down=1)
+    ledger = UsageLedger()
+    ledger.reward(uid)
+    ledger.reward(uid)
+    assert len(list(get_table("badges").find(user_uid=uid, badge_name=FIRST_BADGE))) == 1

tests/unit/services/nodes/policy.py

@@ -0,0 +1,143 @@
+# retoor <retoor@molodetz.nl>
+
+import pytest
+
+from devplacepy.services.nodes.policy import DestinationPolicy, PolicyError, is_public_address
+from devplacepy.services.nodes.spec import SPEC
+
+POLICY = DestinationPolicy()
+
+
+def code(url, method="GET", mode="any"):
+    with pytest.raises(PolicyError) as info:
+        POLICY.check(url, method, mode)
+    return info.value.code
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://reddit.com/r/python.json",
+        "https://www.reddit.com/r/python/top.json?t=week",
+        "http://old.reddit.com/",
+        "https://REDDIT.COM/",
+        "https://reddit.com./r/x",
+        "https://redd.it/abc",
+        "https://reddit.com:443/x",
+        "http://reddit.com:80/x",
+    ],
+)
+def test_allowlisted_urls_pass(url):
+    target = POLICY.check(url)
+    assert target.method == "GET" and target.adapter == "reddit"
+    assert target.port in (80, 443)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://evilreddit.com/",
+        "https://reddit.com.evil.net/",
+        "https://notreddit.com/",
+        "https://example.com/",
+        "https://localhost/",
+        "https://reddit.evil.com/",
+        "https://evil.com/reddit.com",
+    ],
+)
+def test_other_hosts_are_refused(url):
+    assert code(url) == "host_not_allowed"
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://127.0.0.1/",
+        "https://[::1]/",
+        "http://10.0.0.5/",
+        "https://169.254.169.254/latest/meta-data",
+        "https://0x7f.1/",
+        "https://2130706433/",
+    ],
+)
+def test_address_literals_are_refused(url):
+    assert code(url) in ("host_not_allowed", "bad_url")
+
+
+@pytest.mark.parametrize("port", [22, 25, 8080, 8443, 1, 65535])
+def test_only_web_ports(port):
+    assert code(f"https://reddit.com:{port}/") == "port_not_allowed"
+
+
+@pytest.mark.parametrize(
+    "url",
+    ["ftp://reddit.com/", "file:///etc/passwd", "javascript:alert(1)", "//reddit.com/x", "reddit.com/x", ""],
+)
+def test_other_schemes_and_shapes_are_bad_urls(url):
+    assert code(url) == "bad_url"
+
+
+def test_credentials_in_the_url_are_refused():
+    assert code("https://user:pw@reddit.com/") == "bad_url"
+    assert code("https://user@reddit.com/") == "bad_url"
+    assert code("https://evil.com@reddit.com.evil.com/") == "bad_url"
+    assert code("https://reddit.com@evil.com/") == "bad_url"
+
+
+def test_overlong_and_non_string_urls_are_refused():
+    assert code("https://reddit.com/" + "a" * SPEC.limits.url_max_chars) == "bad_url"
+    with pytest.raises(PolicyError):
+        POLICY.check(None)
+    with pytest.raises(PolicyError):
+        POLICY.check(["https://reddit.com/"])
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE", "PATCH", "CONNECT", ""])
+def test_only_declared_methods(method):
+    assert code("https://reddit.com/", method) == "method_not_allowed"
+
+
+def test_method_is_case_insensitive():
+    assert POLICY.check("https://reddit.com/", "get").method == "GET"
+
+
+def test_unknown_mode_is_refused():
+    assert code("https://reddit.com/", mode="raw") == "bad_mode"
+
+
+def test_a_redirect_off_the_allowlist_is_a_redirect_refusal():
+    with pytest.raises(PolicyError) as info:
+        POLICY.check_redirect("https://evil.example/")
+    assert info.value.code == "redirect_not_allowed"
+    assert POLICY.check_redirect("https://www.reddit.com/new").host == "www.reddit.com"
+
+
+@pytest.mark.parametrize(
+    "address,expected",
+    [
+        ("8.8.8.8", True),
+        ("1.1.1.1", True),
+        ("2606:4700:4700::1111", True),
+        ("127.0.0.1", False),
+        ("10.1.2.3", False),
+        ("172.16.0.1", False),
+        ("192.168.1.1", False),
+        ("169.254.1.1", False),
+        ("100.64.0.1", False),
+        ("100.127.255.254", False),
+        ("224.0.0.1", False),
+        ("0.0.0.0", False),
+        ("::1", False),
+        ("fe80::1", False),
+        ("fc00::1", False),
+        ("not an ip", False),
+    ],
+)
+def test_public_address_classification(address, expected):
+    assert is_public_address(address) is expected
+
+
+def test_every_destination_in_the_spec_passes_its_own_policy():
+    for destination in SPEC.destinations:
+        target = POLICY.check(f"https://{destination.suffix}/x")
+        assert target.destination is destination

tests/unit/services/nodes/scheduler.py

@@ -0,0 +1,283 @@
+# retoor <retoor@molodetz.nl>
+
+import random
+from collections import Counter
+from datetime import datetime, timezone
+
+import pytest
+
+from devplacepy.services.nodes.circuit import CircuitBreaker
+from devplacepy.services.nodes.policy import DestinationPolicy
+from devplacepy.services.nodes.registry import NodeCounters, NodeRegistry
+from devplacepy.services.nodes.scheduler import MEGABYTE, NodeScheduler, in_active_hours
+from devplacepy.services.nodes.session import Inflight, NodeSession
+from devplacepy.services.nodes.spec import SPEC
+from tests.unit.services.nodes.session import FakeSocket
+
+LIMITS = SPEC.limits
+TARGET = DestinationPolicy().check("https://www.reddit.com/r/python.json")
+
+
+class Clock:
+    def __init__(self):
+        self.now = 1000.0
+
+    def __call__(self):
+        return self.now
+
+
+def settings(**overrides):
+    base = {
+        "enabled": True,
+        "bandwidth_mb_day": 100,
+        "max_concurrent": 2,
+        "active_from": "",
+        "active_to": "",
+        "timezone": "UTC",
+        "allowed_platforms": ["reddit"],
+        "pause_on_battery": True,
+        "public_display": False,
+    }
+    base.update(overrides)
+    return base
+
+
+def node(uid="d1", user="u1", modes=("fetch",), **overrides):
+    session = NodeSession(FakeSocket(), {"uid": uid, "user_uid": user, "kind": "native"})
+    session.hello = {"modes": list(modes)}
+    session.settings = settings(**overrides)
+    session.roll_day()
+    return session
+
+
+@pytest.fixture
+def world(local_db):
+    clock = Clock()
+    registry = NodeRegistry(NodeCounters())
+    breaker = CircuitBreaker(clock=clock)
+    wall = {"now": datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)}
+    scheduler = NodeScheduler(registry, breaker, clock, lambda: wall["now"], random.Random(1))
+    scheduler.wall = wall
+    scheduler.clock = clock
+    return scheduler
+
+
+def add(world, **kwargs):
+    session = node(**kwargs)
+    world.registry.admit(session)
+    return session
+
+
+def test_a_healthy_node_is_picked(world):
+    session = add(world)
+    assert world.pick(TARGET) is session
+
+
+def test_no_nodes_means_no_pick(world):
+    assert world.pick(TARGET) is None
+
+
+def test_tunnel_only_jobs_are_not_dispatchable_yet(world):
+    add(world)
+    assert world.pick(TARGET, "tunnel") is None
+    assert world.pick(TARGET, "fetch") is not None
+    assert world.pick(TARGET, "any") is not None
+
+
+def test_a_node_must_speak_the_transport(world):
+    add(world, modes=("other",))
+    assert world.pick(TARGET) is None
+
+
+def test_a_tunnel_node_serves_tunnel_and_any_but_not_fetch(world):
+    session = add(world, modes=("tunnel",))
+    assert world.pick(TARGET, "tunnel") is session
+    assert world.pick(TARGET, "any") is session
+    assert world.pick(TARGET, "fetch") is None
+    assert world.transport_of(session, TARGET, "any") == "tunnel"
+
+
+def test_any_prefers_nodes_that_can_tunnel(world):
+    fetcher = add(world, uid="f1", user="u1", modes=("fetch",))
+    tunneler = add(world, uid="t1", user="u2", modes=("fetch", "tunnel"))
+    assert world.candidates(TARGET, "any") == [tunneler]
+    assert world.transport_of(fetcher, TARGET, "any") == "fetch"
+
+
+def test_platform_opt_out_is_honoured(world):
+    add(world, allowed_platforms=[])
+    assert world.pick(TARGET) is None
+
+
+def test_a_node_never_serves_its_own_owner(world):
+    add(world, user="alice")
+    assert world.pick(TARGET, caller_owner="alice") is None
+    assert world.pick(TARGET, caller_owner="bob") is not None
+
+
+def test_concurrency_is_capped_by_the_lower_of_server_and_donor(world):
+    session = add(world, max_concurrent=1)
+    session.inflight["j"] = Inflight("j", "h", "reddit.com", "GET", "u", 0, 0)
+    assert world.pick(TARGET) is None
+    session.inflight.clear()
+    assert session.max_concurrent == 1
+    session.settings["max_concurrent"] = 99
+    assert session.max_concurrent == LIMITS.max_concurrent_per_node
+    session.hello["limits"] = {"max_concurrent": 1}
+    assert session.max_concurrent == 1
+
+
+def test_the_donor_bandwidth_cap_stops_assignment(world):
+    session = add(world, bandwidth_mb_day=1)
+    session.used_today = MEGABYTE - 1
+    assert world.pick(TARGET) is session
+    session.used_today = MEGABYTE
+    assert world.pick(TARGET) is None
+
+
+def test_a_new_day_resets_the_used_bytes(world):
+    session = add(world, bandwidth_mb_day=1)
+    session.used_today = MEGABYTE
+    session.used_day = "2000-01-01"
+    assert world.pick(TARGET) is session
+    assert session.used_today == 0
+
+
+def test_a_node_waits_between_requests_to_the_same_destination(world):
+    session = add(world)
+    world.mark_started(session, TARGET)
+    world.clock.now += LIMITS.min_seconds_per_node_domain - 0.1
+    assert world.eligible(TARGET) == []
+    world.clock.now += 0.2
+    world.clock.now += 1.0 / TARGET.destination.max_rps
+    assert world.eligible(TARGET) == [session]
+
+
+def test_the_destination_has_a_global_rate_ceiling(world):
+    first, second = add(world, uid="a", user="ua"), add(world, uid="b", user="ub")
+    world.mark_started(first, TARGET)
+    world.clock.now += 1.0 / TARGET.destination.max_rps - 0.01
+    assert world.pick(TARGET) is None
+    world.clock.now += 0.02
+    assert world.pick(TARGET) in (first, second)
+
+
+def test_a_paused_node_is_skipped_until_the_pause_ends(world):
+    session = add(world)
+    session.paused_until = world.clock.now + 5
+    assert world.pick(TARGET) is None
+    world.clock.now += 5
+    assert world.pick(TARGET) is session
+
+
+def test_an_open_node_circuit_removes_only_that_node(world):
+    bad, good = add(world, uid="bad", user="u1"), add(world, uid="good", user="u2")
+    for _ in range(LIMITS.breaker_failures):
+        world.breaker.failure(("node", "bad"))
+    assert world.eligible(TARGET) == [good]
+    world.clock.now += LIMITS.breaker_open_seconds
+    assert set(world.eligible(TARGET)) == {bad, good}
+
+
+def test_an_open_destination_circuit_stops_everything(world):
+    add(world)
+    for _ in range(LIMITS.breaker_failures):
+        world.breaker.failure(("destination", TARGET.destination.suffix))
+    assert world.pick(TARGET) is None
+
+
+def test_least_loaded_then_least_recently_used_wins(world):
+    busy, idle_old, idle_new = add(world, uid="a", user="1"), add(world, uid="b", user="2"), add(world, uid="c", user="3")
+    busy.inflight["j"] = Inflight("j", "h", "reddit.com", "GET", "u", 0, 0)
+    idle_old.last_assigned = 10
+    idle_new.last_assigned = 20
+    assert world.pick(TARGET) is idle_old
+
+
+def test_selection_is_fair_across_equal_nodes(world):
+    sessions = [add(world, uid=f"d{i}", user=f"u{i}") for i in range(5)]
+    picks = Counter()
+    for _ in range(500):
+        chosen = world.pick(TARGET)
+        picks[chosen.device_uid] += 1
+        chosen.last_assigned = world.clock.now
+        world.clock.now += 0.001
+    assert set(picks) == {item.device_uid for item in sessions}
+    assert max(picks.values()) - min(picks.values()) <= 1
+
+
+@pytest.mark.parametrize(
+    "start,end,hour,minute,expected",
+    [
+        ("", "", 3, 0, True),
+        ("09:00", "17:00", 9, 0, True),
+        ("09:00", "17:00", 16, 59, True),
+        ("09:00", "17:00", 17, 0, False),
+        ("09:00", "17:00", 8, 59, False),
+        ("22:00", "06:00", 23, 0, True),
+        ("22:00", "06:00", 5, 59, True),
+        ("22:00", "06:00", 6, 0, False),
+        ("22:00", "06:00", 12, 0, False),
+        ("09:00", "09:00", 3, 0, True),
+        ("bogus", "17:00", 3, 0, True),
+    ],
+)
+def test_active_hours_windows(start, end, hour, minute, expected):
+    moment = datetime(2026, 1, 1, hour, minute, tzinfo=timezone.utc)
+    assert in_active_hours({"active_from": start, "active_to": end, "timezone": "UTC"}, moment) is expected
+
+
+def test_active_hours_use_the_donors_timezone():
+    moment = datetime(2026, 1, 1, 22, 0, tzinfo=timezone.utc)
+    amsterdam = {"active_from": "09:00", "active_to": "17:00", "timezone": "Europe/Amsterdam"}
+    assert in_active_hours(amsterdam, moment) is False
+    morning = datetime(2026, 1, 1, 8, 30, tzinfo=timezone.utc)
+    assert in_active_hours(amsterdam, morning) is True
+    broken = {"active_from": "09:00", "active_to": "17:00", "timezone": "Mars/Base"}
+    assert in_active_hours(broken, datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)) is True
+
+
+def test_a_node_outside_its_hours_is_skipped(world):
+    add(world, active_from="22:00", active_to="23:00")
+    assert world.pick(TARGET) is None
+    world.wall["now"] = datetime(2026, 1, 1, 22, 30, tzinfo=timezone.utc)
+    assert world.pick(TARGET) is not None
+
+
+def test_mark_started_records_both_clocks(world):
+    session = add(world)
+    world.mark_started(session, TARGET)
+    assert session.host_started[TARGET.destination.suffix] == world.clock.now
+    assert session.last_assigned == world.clock.now
+    assert world._destination_started[TARGET.destination.suffix] == world.clock.now
+
+
+def test_the_per_destination_maps_stay_bounded_by_the_allowlist(world):
+    session = add(world)
+    policy = DestinationPolicy()
+    for index in range(200):
+        world.mark_started(session, policy.check(f"https://sub{index}.reddit.com/"))
+    assert len(session.host_started) == 1
+    assert len(world._destination_started) == 1
+
+
+def test_window_checks_are_cached_per_minute_and_still_correct(world):
+    from devplacepy.services.nodes.scheduler import _window_open
+
+    _window_open.cache_clear()
+    window = {"active_from": "09:00", "active_to": "17:00", "timezone": "UTC"}
+    noon = datetime(2026, 1, 1, 12, 0, 30, tzinfo=timezone.utc)
+    for _ in range(100):
+        assert in_active_hours(window, noon) is True
+    assert _window_open.cache_info().misses == 1
+    night = datetime(2026, 1, 1, 3, 0, tzinfo=timezone.utc)
+    assert in_active_hours(window, night) is False
+
+
+def test_choose_applies_the_owner_exclusion_to_a_shared_candidate_list(world):
+    mine, theirs = add(world, uid="a", user="alice"), add(world, uid="b", user="bob")
+    found = world.candidates(TARGET)
+    assert world.choose(found, "alice") is theirs
+    assert world.choose(found, "bob") is mine
+    assert world.choose([mine], "alice") is None
+    assert world.choose([], "") is None

tests/unit/services/nodes/service.py

@@ -1,5 +1,6 @@
 # retoor <retoor@molodetz.nl>
 
+import asyncio
 import time
 
 import pytest
@@ -146,3 +147,32 @@ def test_metrics_report_connected_and_capacity(service):
 def test_the_service_has_the_kill_switch_as_a_config_field(service):
     keys = [field.key for field in service.all_fields()]
     assert "nodes_enabled" in keys and "nodes_max_total" in keys
+
+
+def test_changed_settings_are_pushed_to_the_node_once(service):
+    uid = member()
+    device, socket = connected(service, uid)
+    session = service.registry.get(device["uid"])
+    session.settings = {"stale": True}
+    assert reconcile(service) == 0
+    pushed = [frame for frame in socket.sent if frame["type"] == "settings_update"]
+    assert len(pushed) == 1 and pushed[0]["settings"]["enabled"] is True
+    assert session.settings == pushed[0]["settings"]
+    reconcile(service)
+    assert len([frame for frame in socket.sent if frame["type"] == "settings_update"]) == 1
+    save_node_settings(uid, {"bandwidth_mb_day": 9})
+    reconcile(service)
+    assert session.settings["bandwidth_mb_day"] == 9
+
+
+def test_the_dispatch_loop_starts_once_and_stops_with_the_service(service):
+    async def scenario():
+        service.ensure_dispatching()
+        first = service._dispatch_task
+        service.ensure_dispatching()
+        assert service._dispatch_task is first and not first.done()
+        await service.on_disable()
+        await asyncio.sleep(0)
+        assert service._dispatch_task is None and first.cancelled()
+
+    run_async(scenario())

tests/unit/services/nodes/session.py

@@ -93,15 +93,20 @@ def test_a_silent_client_times_out(local_db, monkeypatch):
 
 def test_usage_is_advisory_and_summed():
     session = session_for(FakeSocket())
-    session._handle({"type": "usage", "bytes_up": 5, "bytes_down": 7})
-    session._handle({"type": "usage", "bytes_up": -9, "bytes_down": 1})
+    run(session._handle({"type": "usage", "bytes_up": 5, "bytes_down": 7}))
+    run(session._handle({"type": "usage", "bytes_up": -9, "bytes_down": 1}))
     assert (session.reported_up, session.reported_down) == (5, 8)
 
 
-@pytest.mark.parametrize("kind", ["job_result", "job_error", "tunnel_data", "tunnel_close", "tunnel_opened"])
+@pytest.mark.parametrize("kind", ["job_result", "job_error"])
 def test_frames_without_an_assigned_job_are_protocol_errors(kind):
     with pytest.raises(FrameError):
-        session_for(FakeSocket())._handle({"type": kind})
+        run(session_for(FakeSocket())._handle({"type": kind, "job_id": "nope"}))
+
+
+@pytest.mark.parametrize("kind", ["tunnel_data", "tunnel_close", "tunnel_opened"])
+def test_late_tunnel_frames_for_a_closed_relay_are_ignored(kind):
+    run(session_for(FakeSocket())._handle({"type": kind, "conn_id": "gone"}))
 
 
 def test_missed_pings_evict_after_the_floor(monkeypatch):

tests/unit/services/nodes/social.py

@@ -0,0 +1,88 @@
+# retoor <retoor@molodetz.nl>
+
+import json
+
+from devplacepy.database import set_setting
+from devplacepy.services.nodes import social as social_module
+from devplacepy.services.nodes.adapters.base import AdapterError
+from devplacepy.services.nodes.fabric import FabricUnavailable
+from devplacepy.services.nodes.social import SocialSearch
+from tests.conftest import run_async
+
+HN = json.dumps({"hits": [{"objectID": "1", "title": "HN story", "points": 1}]}).encode()
+SE = json.dumps({"items": [{"title": "SO question", "link": "https://stackoverflow.com/q/1"}]}).encode()
+
+
+class StubFabric:
+    def __init__(self, error=None):
+        self.error = error
+        self.calls = 0
+
+    async def fetch(self, url, **kwargs):
+        self.calls += 1
+        if self.error:
+            raise self.error
+        raise AssertionError("not expected")
+
+
+def search(monkeypatch, getter, platforms, fabric=None):
+    monkeypatch.setattr(social_module, "direct_get", getter)
+    return run_async(
+        SocialSearch(fabric or StubFabric()).search(
+            "sqlite", caller_kind="user", caller_id="u1", platforms=platforms
+        )
+    )
+
+
+def test_hits_are_merged_and_failures_reported(monkeypatch):
+    async def getter(url):
+        if "algolia" in url:
+            return HN
+        if "stackexchange" in url:
+            return SE
+        raise AdapterError("upstream", "HTTP 500")
+
+    outcome = search(monkeypatch, getter, ["hackernews", "stackexchange", "github"])
+    assert {item.platform for item in outcome.hits} == {"hackernews", "stackexchange"}
+    assert outcome.unavailable == {"github": "upstream"}
+
+
+def test_unknown_platform_is_dropped(monkeypatch):
+    async def getter(url):
+        return HN
+
+    outcome = search(monkeypatch, getter, ["hackernews", "x", "instagram"])
+    assert [item.platform for item in outcome.hits] == ["hackernews"]
+    assert outcome.unavailable == {}
+
+
+def test_reddit_is_off_unless_routing_through_nodes_is_enabled(monkeypatch):
+    async def getter(url):
+        return HN
+
+    fabric = StubFabric()
+    outcome = search(monkeypatch, getter, ["reddit"], fabric)
+    assert outcome.unavailable == {"reddit": "nodes_not_used"}
+    assert fabric.calls == 0
+
+
+def test_reddit_reports_the_fabric_reason(monkeypatch):
+    async def getter(url):
+        return HN
+
+    set_setting("nodes_social_via_nodes", "1")
+    try:
+        outcome = search(monkeypatch, getter, ["reddit"], StubFabric(FabricUnavailable("no_nodes")))
+    finally:
+        set_setting("nodes_social_via_nodes", "0")
+    assert outcome.unavailable == {"reddit": "no_nodes"}
+
+
+def test_blank_query_and_limit_cap(monkeypatch):
+    async def getter(url):
+        return HN
+
+    monkeypatch.setattr(social_module, "direct_get", getter)
+    empty = run_async(SocialSearch(StubFabric()).search("  ", caller_kind="user", caller_id="u"))
+    assert empty.hits == []
+    assert social_module.MAX_LIMIT == 20

tests/unit/services/nodes/tunnel.py

@@ -0,0 +1,112 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import base64
+
+from devplacepy.services.nodes import tunnel as tunnel_module
+from devplacepy.services.nodes.policy import DestinationPolicy
+from devplacepy.services.nodes.tunnel import Relay, TunnelBridge
+from tests.conftest import run_async
+
+TARGET = DestinationPolicy().check("https://www.reddit.com/r/python.json")
+
+
+class Donor:
+    def __init__(self, bridge, accept=True, garbage=False):
+        self.bridge = bridge
+        self.accept = accept
+        self.garbage = garbage
+        self.tunnels = {}
+        self.sent = []
+
+    async def send(self, frame_type, **fields):
+        self.sent.append({"type": frame_type, **fields})
+        if frame_type == "tunnel_open":
+            self.bridge.on_frame(self, {"type": "tunnel_opened", "conn_id": fields["conn_id"], "ok": self.accept})
+        elif frame_type == "tunnel_data":
+            raw = base64.b64decode(fields["b64"])
+            reply = "***not base64***" if self.garbage else base64.b64encode(raw.upper()).decode()
+            self.bridge.on_frame(self, {"type": "tunnel_data", "conn_id": fields["conn_id"], "b64": reply})
+
+
+async def session_through(donor, connect_line, payload=b"hello"):
+    relay = Relay(donor, TARGET)
+    server = await asyncio.start_server(relay.handle, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+    writer.write(connect_line.encode() + b"\r\n\r\n")
+    await writer.drain()
+    status = await asyncio.wait_for(reader.readline(), 5)
+    echoed = b""
+    if b" 200 " in status:
+        await reader.readuntil(b"\r\n")
+        writer.write(payload)
+        await writer.drain()
+        try:
+            echoed = await asyncio.wait_for(reader.read(len(payload)), 5)
+        except asyncio.TimeoutError:
+            echoed = b""
+    writer.close()
+    server.close()
+    await asyncio.sleep(0.05)
+    return status, echoed, relay
+
+
+def go(donor, line, payload=b"hello"):
+    return run_async(session_through(donor, line, payload))
+
+
+def test_connect_to_the_target_relays_both_ways():
+    donor = Donor(TunnelBridge())
+    status, echoed, relay = go(donor, "CONNECT www.reddit.com:443 HTTP/1.1")
+    assert b" 200 " in status and echoed == b"HELLO"
+    assert [item["type"] for item in donor.sent][:2] == ["tunnel_open", "tunnel_data"]
+    assert donor.sent[0]["host"] == "www.reddit.com" and donor.sent[0]["port"] == 443
+    assert relay.up == 5 and relay.down == 5
+
+
+def test_any_other_host_or_port_is_refused_without_bothering_the_donor():
+    for line in ("CONNECT evil.example:443 HTTP/1.1", "CONNECT www.reddit.com:22 HTTP/1.1", "CONNECT 10.0.0.1:443 HTTP/1.1"):
+        donor = Donor(TunnelBridge())
+        status, _, relay = go(donor, line)
+        assert b" 403 " in status and donor.sent == []
+        assert relay.error == "host_not_allowed"
+
+
+def test_only_connect_is_accepted():
+    donor = Donor(TunnelBridge())
+    status, _, _ = go(donor, "GET http://www.reddit.com/ HTTP/1.1")
+    assert b" 405 " in status and donor.sent == []
+
+
+def test_a_donor_that_refuses_is_a_bad_gateway():
+    donor = Donor(TunnelBridge(), accept=False)
+    status, _, relay = go(donor, "CONNECT www.reddit.com:443 HTTP/1.1")
+    assert b" 502 " in status and relay.error == "tunnel_refused"
+
+
+def test_the_relay_has_a_byte_cap(monkeypatch):
+    monkeypatch.setattr(tunnel_module, "RELAY_CAP_BYTES", 8)
+    donor = Donor(TunnelBridge())
+    _, echoed, relay = go(donor, "CONNECT www.reddit.com:443 HTTP/1.1", b"0123456789abcdef")
+    assert relay.error == "body_too_large" and len(echoed) < 16
+
+
+def test_invalid_base64_from_the_donor_ends_the_tunnel():
+    donor = Donor(TunnelBridge(), garbage=True)
+    _, echoed, relay = go(donor, "CONNECT www.reddit.com:443 HTTP/1.1")
+    assert relay.error == "bad_base64" and echoed == b""
+
+
+def test_releasing_a_session_closes_its_relays():
+    bridge = TunnelBridge()
+    donor = Donor(bridge)
+    relay = Relay(donor, TARGET)
+    donor.tunnels[relay.conn_id] = relay
+    bridge.release_session(donor)
+    assert donor.tunnels == {} and relay.frames.get_nowait() is tunnel_module.CLOSED
+
+
+def test_frames_for_unknown_connections_are_not_consumed():
+    bridge = TunnelBridge()
+    assert bridge.on_frame(Donor(bridge), {"type": "tunnel_data", "conn_id": "nope", "b64": ""}) is False