diff --git a/leetcode_coach/leetcode_coach/relay/app.py b/leetcode_coach/leetcode_coach/relay/app.py index cbad8fe..53c24bc 100644 --- a/leetcode_coach/leetcode_coach/relay/app.py +++ b/leetcode_coach/leetcode_coach/relay/app.py @@ -104,6 +104,7 @@ class Relay: # pylint: disable=too-many-instance-attributes ) phone.attach() self.phone = phone + self._spawn("rtdb-write", lambda: phone.outbox.run(self.stop_event)) self._spawn("rtdb-listen", lambda: phone.listen(self.stop_event)) self._spawn( "rtdb-tend", lambda: phone.tend(self.stop_event, self.stop_event.wait) @@ -133,6 +134,8 @@ class Relay: # pylint: disable=too-many-instance-attributes """Stop taking requests, let the running one finish, close the port.""" self.stop_event.set() self.lane.stop() + if self.phone is not None: + self.phone.outbox.wake() if any(t.name == "http" for t in self._threads): self.server.shutdown() # blocks unless serve_forever runs self.server.server_close() diff --git a/leetcode_coach/leetcode_coach/relay/chat.py b/leetcode_coach/leetcode_coach/relay/chat.py index b49376f..e8f77b1 100644 --- a/leetcode_coach/leetcode_coach/relay/chat.py +++ b/leetcode_coach/leetcode_coach/relay/chat.py @@ -19,6 +19,7 @@ from __future__ import annotations import logging import os import re +import time from typing import TYPE_CHECKING, Any, Final import urllib.error @@ -33,6 +34,9 @@ if TYPE_CHECKING: _logger: Final = logging.getLogger(__name__) TURN_TIMEOUT_S: Final = 240.0 +REQUEST_CAP_S: Final = 280.0 +"""A whole request; a coach turn gets what is left of it (at most 240 s).""" +MIN_TURN_S: Final = 15.0 RESETS_RE: Final = re.compile(r"resets\s+([^\n]+?)\s*$", re.IGNORECASE | re.MULTILINE) TurnError = CoachError """A coach turn produced no reply (``model_error`` / ``claude_unavailable``).""" @@ -61,9 +65,20 @@ class CoachChat: """Bind the claude call and the local model (both injectable).""" self._ask = ask self._local_chat = local_chat + self.monotonic: Callable[[], float] = time.monotonic + self.deadline = float("inf") + """When the running request hits `REQUEST_CAP_S` (set by `Coach.handle`).""" + + def budget(self) -> float: + """Seconds this turn may take; `model_error` when too few are left.""" + left = self.deadline - self.monotonic() + if left < MIN_TURN_S: + msg = "The PC ran out of time for this turn; send it again." + raise TurnError(msg, code="model_error", retry=True) + return min(TURN_TIMEOUT_S, left) def _bounded(self, messages: list[dict[str, str]]) -> local_backend.LocalReply: - return self._local_chat(messages, timeout=TURN_TIMEOUT_S) + return self._local_chat(messages, timeout=self.budget()) def open(self, session: Session) -> str: """The coach's opening reply for a fresh session.""" @@ -79,7 +94,7 @@ class CoachChat: args = ["--model", model()] if resume and session.claude_id: args += ["--resume", session.claude_id] - reply = self._ask(args, text, timeout=TURN_TIMEOUT_S, env=_env(session)) + reply = self._ask(args, text, timeout=self.budget(), env=_env(session)) if claude_cli.is_limit_hit(reply): reason = str(reply.get("result") or "usage limit") _logger.warning("claude usage limit (%s): coaching on locally", reason) diff --git a/leetcode_coach/leetcode_coach/relay/coach.py b/leetcode_coach/leetcode_coach/relay/coach.py index 317a45f..932e11b 100644 --- a/leetcode_coach/leetcode_coach/relay/coach.py +++ b/leetcode_coach/leetcode_coach/relay/coach.py @@ -18,7 +18,7 @@ from typing import TYPE_CHECKING, Any, Final from leetcode_coach import answer_key, claude_cli, local_backend, toolbox from leetcode_coach.relay import config, grading, protocol, runcode, steps, turns -from leetcode_coach.relay.chat import CoachChat +from leetcode_coach.relay.chat import REQUEST_CAP_S, CoachChat from leetcode_coach.relay.errors import CoachError from leetcode_coach.relay.probe import ACCEPTED, AcceptedProbe from leetcode_coach.relay.session import Session, grade_line, restore, save @@ -170,6 +170,7 @@ class Coach: def handle(self, req: protocol.Request) -> dict[str, Any]: """Run one request; its result (section 2).""" with self.lock: + self.chat.deadline = self.chat.monotonic() + REQUEST_CAP_S try: data = self._dispatch(req) except CoachError as exc: diff --git a/leetcode_coach/leetcode_coach/relay/outbox.py b/leetcode_coach/leetcode_coach/relay/outbox.py new file mode 100644 index 0000000..59cbf54 --- /dev/null +++ b/leetcode_coach/leetcode_coach/relay/outbox.py @@ -0,0 +1,105 @@ +"""The phone mirror's writes, in order, on their own thread. + +A hanging Firebase must never stall the desktop: the lane's one worker only +*enqueues* (chat lines, ``state``, the heartbeat, a result) and this outbox +writes them. Order is the contract's: chat reset / new lines, then +``state``, then the heartbeat, then the request's result -- the phone +refetches chat when ``heartbeat.state_rev`` moves, so a heartbeat landing +before its chat lines would lose one. ``state`` and the heartbeat keep only +their latest value (an older one still waiting is dropped, the newer goes to +the back); results and chat lines are never dropped. A failed write stays at +the front and is retried after a backoff; nothing behind it overtakes it. +""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +import logging +import threading +from typing import TYPE_CHECKING, Final + +from leetcode_coach.relay.rtdb import RtdbError + +if TYPE_CHECKING: + from collections.abc import Callable + +_logger: Final = logging.getLogger(__name__) + +BACKOFF_S: Final = (1.0, 2.0, 5.0, 10.0, 30.0) +IDLE_S: Final = 0.5 + + +@dataclass(frozen=True) +class Write: + """One pending write: what it is (for the log) and how to do it.""" + + what: str + run: Callable[[], None] + latest: str | None = None + """A key whose older pending writes this one replaces (state, heartbeat).""" + + +class Outbox: + """An ordered queue of RTDB writes and the loop that drains it.""" + + def __init__(self) -> None: + """Start empty; `run` drains on a thread, `drain` in the caller.""" + self._cv = threading.Condition() + self._writes: deque[Write] = deque() + + def put( + self, what: str, run: Callable[[], None], latest: str | None = None + ) -> None: + """Queue one write (replacing a waiting one with the same `latest`).""" + with self._cv: + if latest is not None: + self._writes = deque(w for w in self._writes if w.latest != latest) + self._writes.append(Write(what, run, latest)) + self._cv.notify_all() + + def pending(self) -> list[str]: + """What is still waiting, in order.""" + with self._cv: + return [w.what for w in self._writes] + + def drain(self) -> bool: + """Write everything waiting; False at the first failure (it stays first).""" + while True: + with self._cv: + if not self._writes: + return True + write = self._writes.popleft() + try: + write.run() + except RtdbError as exc: + _logger.warning("RTDB %s failed: %s", write.what, exc) + with self._cv: + newer = write.latest is not None and any( + w.latest == write.latest for w in self._writes + ) + if not newer: + self._writes.appendleft(write) + return False + + def run( + self, stop: threading.Event, wait: Callable[[float], object] | None = None + ) -> None: + """Drain until `stop`, backing off after a failed write.""" + pause = stop.wait if wait is None else wait + failures = 0 + while not stop.is_set(): + with self._cv: + if not self._writes: + self._cv.wait(IDLE_S) + continue + if self.drain(): + failures = 0 + continue + pause(BACKOFF_S[min(failures, len(BACKOFF_S) - 1)]) + failures += 1 + + def wake(self) -> None: + """Let a waiting `run` notice its stop event at once.""" + with self._cv: + self._cv.notify_all() diff --git a/leetcode_coach/leetcode_coach/relay/phone.py b/leetcode_coach/leetcode_coach/relay/phone.py index 9711a7e..11869a0 100644 --- a/leetcode_coach/leetcode_coach/relay/phone.py +++ b/leetcode_coach/leetcode_coach/relay/phone.py @@ -8,8 +8,9 @@ on the shared lane. Results, ``state`` (a JSON string leaf) and the chat lines are written as they change; the heartbeat every `HEARTBEAT_S` and on every ``rev`` or busy change. ``requests/{rid}`` goes 24 h after its result. -Firebase being down never stops the relay: every write failure is logged -and the next change or beat tries again; the desktop keeps being served. +Writes only ever *enqueue* (`outbox`): the lane's worker never waits on +Firebase, so a hanging or down database never slows the desktop. A failed +write is logged and retried in order; the relay never exits over it. """ from __future__ import annotations @@ -22,6 +23,7 @@ from typing import TYPE_CHECKING, Any, Final from leetcode_coach.relay import protocol, sse from leetcode_coach.relay.done import DAY_MS +from leetcode_coach.relay.outbox import Outbox from leetcode_coach.relay.rtdb import SERVER_TIME, RtdbError if TYPE_CHECKING: @@ -74,19 +76,11 @@ class PhoneLink: self._clock = clock_ms self._beat_every = heartbeat_s self._rescan_every = rescan_s - self._write_lock = threading.Lock() + self.outbox = Outbox() + self._mirror_lock = threading.Lock() self._chat_sid: str | None = None self._sent: set[str] = set() - def _write(self, what: str, call: Callable[[], None]) -> bool: - with self._write_lock: - try: - call() - except RtdbError as exc: - _logger.warning("RTDB %s failed: %s", what, exc) - return False - return True - def attach(self) -> None: """Listen to the lane and the backend; then mirror everything once.""" self.lane.on_result.append(self.on_result) @@ -98,12 +92,14 @@ class PhoneLink: def beat(self) -> None: """PUT the heartbeat (``beat`` as a server timestamp).""" beat = {**self._heartbeat(), "beat": SERVER_TIME} - self._write("heartbeat", lambda: self.rtdb.put("heartbeat", beat)) + put = functools.partial(self.rtdb.put, "heartbeat", beat) + self.outbox.put("heartbeat", put, latest="heartbeat") def put_result(self, rid: str, result: dict[str, Any]) -> None: """``requests/{rid}/result`` as a JSON string leaf.""" text = json.dumps(result, ensure_ascii=False) - self._write("result", lambda: self.rtdb.put(f"requests/{rid}/result", text)) + put = functools.partial(self.rtdb.put, f"requests/{rid}/result", text) + self.outbox.put("result", put) def on_result(self, req: protocol.Request, result: dict[str, Any]) -> None: """A finished request: answer it if it came from the phone.""" @@ -111,21 +107,25 @@ class PhoneLink: self.put_result(req.rid, result) def on_state(self, state: dict[str, Any], session: Session | None) -> None: - """Mirror ``state``, the new chat lines and the heartbeat.""" + """Queue the chat lines, then ``state``, then the heartbeat.""" text = json.dumps(state, ensure_ascii=False) - self._write("state", lambda: self.rtdb.put("state", text)) - sid = None if session is None else session.sid - if sid != self._chat_sid and self._write( - "chat reset", lambda: self.rtdb.delete("chat") - ): - self._chat_sid, self._sent = sid, set() - if session is not None and sid == self._chat_sid: + with self._mirror_lock: + sid = None if session is None else session.sid + if sid != self._chat_sid: + self.outbox.put( + "chat reset", functools.partial(self.rtdb.delete, "chat") + ) + self._chat_sid, self._sent = sid, set() + lines = {} if session is None else dict(session.chat) new: dict[str, object] = { - k: v for k, v in session.chat.items() if k not in self._sent + k: v for k, v in lines.items() if k not in self._sent } - if new and self._write("chat", lambda: self.rtdb.patch("chat", new)): + if new: + self.outbox.put("chat", functools.partial(self.rtdb.patch, "chat", new)) self._sent |= set(new) - self.beat() + put = functools.partial(self.rtdb.put, "state", text) + self.outbox.put("state", put, latest="state") + self.beat() def consider(self, rid: str, node: object) -> None: """One ``requests/{rid}`` node: answer, queue, or leave it.""" @@ -183,7 +183,7 @@ class PhoneLink: done = self.lane.done.get(str(rid)) if done is None or now - done.at > DAY_MS: drop = functools.partial(self.rtdb.delete, f"requests/{rid}") - self._write("cleanup", drop) + self.outbox.put("cleanup", drop) def listen(self, stop: threading.Event) -> None: """The SSE loop; a cancelled stream leaves the rescans to carry on.""" diff --git a/leetcode_coach/leetcode_coach/tests/test_relay_outbox.py b/leetcode_coach/leetcode_coach/tests/test_relay_outbox.py new file mode 100644 index 0000000..43c84b7 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_relay_outbox.py @@ -0,0 +1,123 @@ +"""A hanging Firebase never stalls the desktop; writes land in contract order.""" + +from __future__ import annotations + +import threading +import time +from typing import TYPE_CHECKING + +from leetcode_coach.relay import outbox as outbox_mod +from leetcode_coach.relay.done import DoneMap +from leetcode_coach.relay.lane import Lane +from leetcode_coach.relay.phone import PhoneLink +from leetcode_coach.relay.rtdb import RtdbError +from leetcode_coach.tests._relay_direct import FakeAsk, make_coach, req + +if TYPE_CHECKING: + from pathlib import Path + + +class Stuck: + """An RTDB whose writes hang until released, recording their order.""" + + def __init__(self) -> None: + self.release = threading.Event() + self.calls: list[tuple[str, str]] = [] + + def _hang(self, method: str, path: str) -> None: + self.release.wait(10) + self.calls.append((method, path)) + + def put(self, path: str, _value: object) -> None: + self._hang("put", path) + + def patch(self, path: str, _value: object) -> None: + self._hang("patch", path) + + def delete(self, path: str) -> None: + self._hang("delete", path) + + +def test_the_desktop_is_served_while_rtdb_hangs(tmp_path: Path) -> None: + coach = make_coach(tmp_path, FakeAsk([{"result": "Hi?"}])) + lane = Lane( + coach.handle, DoneMap(tmp_path / "done.jsonl", lambda: 1), clock_ms=lambda: 1 + ) + rtdb = Stuck() + link = PhoneLink(rtdb, lane, coach, lambda: {"version": 1}, clock_ms=lambda: 1) + stop = threading.Event() + writer = threading.Thread(target=link.outbox.run, args=(stop,)) + writer.start() + link.attach() + t0 = time.monotonic() + started = coach.handle(req("session_start", slug="missing-number")) + phone_req = req("hint", transport="phone", session=started["data"]["session"]) + link.on_result(phone_req, {"ok": True}) + assert started["ok"] + assert time.monotonic() - t0 < 2 # never waited on the hanging writes + rtdb.release.set() + deadline = time.monotonic() + 5 + while link.outbox.pending() and time.monotonic() < deadline: + time.sleep(0.01) + stop.set() + link.outbox.wake() + writer.join(timeout=5) + tail = rtdb.calls[-5:] + assert tail == [ + ("delete", "chat"), + ("patch", "chat"), + ("put", "state"), + ("put", "heartbeat"), + ("put", f"requests/{phone_req.rid}/result"), + ] + + +def test_a_failing_write_backs_off_and_a_stale_one_is_dropped() -> None: + box = outbox_mod.Outbox() + tries: list[str] = [] + + def fail() -> None: + tries.append("state-1") + msg = "PUT state: HTTP 503" + raise RtdbError(msg) + + box.put("state", fail, latest="state") + assert not box.drain() + assert box.pending() == ["state"] + stop = threading.Event() + pauses: list[float] = [] + + def recovered() -> None: + tries.append("state-2") + stop.set() + + def pause(seconds: float) -> None: + pauses.append(seconds) + box.put("state", recovered, latest="state") + + box.run(stop, pause) + assert pauses == [1.0] + assert tries == ["state-1", "state-1", "state-2"] + assert box.pending() == [] + + def fail_then_newer() -> None: + box.put("state", lambda: tries.append("state-3"), latest="state") + msg = "PUT state: HTTP 503" + raise RtdbError(msg) + + box.put("state", fail_then_newer, latest="state") + assert not box.drain() + assert box.pending() == ["state"] # the failed, older one was dropped + assert box.drain() + assert tries[-1] == "state-3" + + +def test_an_idle_outbox_stops_on_wake() -> None: + box = outbox_mod.Outbox() + stop = threading.Event() + runner = threading.Thread(target=box.run, args=(stop,)) + runner.start() + stop.set() + box.wake() + runner.join(timeout=2) + assert not runner.is_alive() diff --git a/leetcode_coach/leetcode_coach/tests/test_relay_phone.py b/leetcode_coach/leetcode_coach/tests/test_relay_phone.py index c3e687a..ff5d38c 100644 --- a/leetcode_coach/leetcode_coach/tests/test_relay_phone.py +++ b/leetcode_coach/leetcode_coach/tests/test_relay_phone.py @@ -90,6 +90,7 @@ def test_events_at_every_depth_reach_the_lane(link: Any) -> None: made.consider(rid(), {"req": _req()}) limited = rid() made.consider(limited, {"req": _req()}) + assert made.outbox.drain() assert "rate_limited" in str(rec.calls[-1][2]) assert rec.calls[-1][1] == f"requests/{limited}/result" assert rec.paths("put").count("heartbeat") == 1 @@ -101,6 +102,7 @@ def test_results_go_only_to_the_phone(link: Any) -> None: made.on_result(desktop, {"ok": True}) phone_req = type("R", (), {"transport": "phone", "rid": rid()})() made.on_result(phone_req, {"ok": True}) + assert made.outbox.drain() puts = rec.paths("put") assert f"requests/{phone_req.rid}/result" in puts assert f"requests/{desktop.rid}/result" not in puts @@ -111,18 +113,26 @@ def test_write_failures_are_logged_and_retried( ) -> None: made, rec, _lane = link coach = made.coach + assert made.outbox.drain() + rec.calls.clear() rec.fail = {"delete"} with caplog.at_level(logging.WARNING): started(coach) + assert not made.outbox.drain() assert "RTDB chat reset failed" in caplog.text - assert rec.paths("patch") == [] - rec.fail = {"patch"} - made.on_state(coach.state(), coach.session) - assert made._sent == set() + assert rec.calls == [] # nothing overtook the failed reset + assert made.outbox.pending() == ["chat reset", "chat", "state", "heartbeat"] + made.on_state(coach.state(), coach.session) # newer state/beat replace waiting + assert made.outbox.pending() == ["chat reset", "chat", "state", "heartbeat"] rec.fail = set() - made.on_state(coach.state(), coach.session) - assert rec.calls[-2][0] == "patch" - assert json.loads(str(rec.calls[0][2]))["schema"] == 1 + assert made.outbox.drain() + assert [(m, p) for m, p, _v in rec.calls] == [ + ("delete", "chat"), + ("patch", "chat"), + ("put", "state"), + ("put", "heartbeat"), + ] + assert json.loads(str(rec.calls[2][2]))["schema"] == 1 def test_rescan_answers_and_cleans_up( @@ -149,6 +159,7 @@ def test_rescan_answers_and_cleans_up( "junk": 5, } made.rescan() + assert made.outbox.drain() assert sorted(rec.paths("delete")) == sorted( [f"requests/{old}", f"requests/{lost}"] ) diff --git a/leetcode_coach/leetcode_coach/tests/test_relay_units.py b/leetcode_coach/leetcode_coach/tests/test_relay_units.py index cc394bd..0dc8b97 100644 --- a/leetcode_coach/leetcode_coach/tests/test_relay_units.py +++ b/leetcode_coach/leetcode_coach/tests/test_relay_units.py @@ -169,3 +169,17 @@ def test_env_and_config_seams(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) - monkeypatch.delenv(config.PORT_ENV) assert config.state_dir() == config.DEFAULT_DIR assert config.port() == config.DEFAULT_PORT + + +def test_a_turn_gets_what_is_left_of_the_request(tmp_path: Path) -> None: + ask = FakeAsk([{"result": "Hi?"}, {"result": "Ok?"}]) + coach = make_coach(tmp_path, ask) + sid = started(coach) + assert ask.calls[0]["timeout"] == chat.TURN_TIMEOUT_S + ticks = iter([0.0, 100.0, 0.0, 270.0]) + coach.chat.monotonic = lambda: next(ticks) + assert coach.handle(req("chat", session=sid, text="so what then?"))["ok"] + assert ask.calls[-1]["timeout"] == 180.0 + late = coach.handle(req("chat", session=sid, text="and now?")) + assert late["code"] == "model_error" + assert late["retry"] is True