diff --git a/keel/commands/jobs.py b/keel/commands/jobs.py new file mode 100644 index 0000000..2bb927a --- /dev/null +++ b/keel/commands/jobs.py @@ -0,0 +1,150 @@ +"""One background job at a time, so a setup step that takes minutes can still be a button (#437). + +The first market-data fetch runs for minutes across an allowlist. A request that blocks that long +is not a button -- the browser gives up, the user reloads, and a second fetch starts on top of the +first. So this exists, and it is deliberately the smallest thing that could work. + +**Exactly one slot.** Not a queue, not a pool. Two concurrent fetches write candles to one SQLite +database and race each other; a setup flow has no use for concurrency; and "is something running?" +with one answer is a question a page can render honestly. A second start is REFUSED and says so, +rather than being silently dropped or silently queued -- both of which look identical to a user +watching a page that is not changing. + +**The progress is a bounded tail.** `run_fetch` already emits the same lines the CLI prints, in +the same order, through its `echo` parameter -- that contract is why this module needs no +knowledge of fetching at all. Keeping the last `_MAX_LINES` of them bounds the memory a long run +can consume in a process that is also serving pages. + +**A failed job stays visible.** It is not cleared on read and not cleared by time: the whole point +of running something in the background is that nobody was watching when it broke, so the failure +has to still be there when they look. It is replaced only when the next job starts. + +**Nothing here decides what to run.** A caller passes a callable; this module owns the thread, the +slot, the buffer and the status. That keeps it out of the argument about what a setup flow is +allowed to do -- which is `keel/commands/setup.py`'s business, and pinned there. +""" + +from __future__ import annotations + +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass, field, replace + +#: The tail kept from a job's progress stream. A first fetch across a wide allowlist emits a line +#: per product per granularity; forty is enough to show what is happening now without holding a +#: whole run's output in a process that is also serving pages. +_MAX_LINES = 40 + +RUNNING = "running" +DONE = "done" +FAILED = "failed" + + +@dataclass(frozen=True) +class JobStatus: + key: str + state: str + started_ts: float + finished_ts: float | None = None + lines: tuple[str, ...] = () + #: Set only when `state == FAILED`. The exception's type and message, never a traceback: a + #: traceback in a browser page is a stack of file paths from someone else's machine. + error: str | None = None + + @property + def is_running(self) -> bool: + return self.state == RUNNING + + @property + def elapsed_sec(self) -> float: + return (self.finished_ts or time.time()) - self.started_ts + + +@dataclass +class _Slot: + lock: threading.Lock = field(default_factory=threading.Lock) + status: JobStatus | None = None + thread: threading.Thread | None = None + + +_slot = _Slot() + + +def status() -> JobStatus | None: + """The current or most recent job, or `None` when nothing has ever run.""" + with _slot.lock: + return _slot.status + + +def is_running() -> bool: + current = status() + return current is not None and current.is_running + + +def start(key: str, run: Callable[[Callable[[str], None]], None]) -> bool: + """Begin `run` in the background. `False` when a job is already running. + + `run` receives an `echo` callable and is expected to feed progress through it. Any exception + it raises is captured into the status rather than propagating: the thread that would receive + it belongs to nobody, and a background failure that only reaches stderr is a failure the + person who started it never sees. + """ + with _slot.lock: + if _slot.status is not None and _slot.status.is_running: + return False + _slot.status = JobStatus(key=key, state=RUNNING, started_ts=time.time()) + + def _append(line: str) -> None: + text = str(line).rstrip() + if not text: + return + with _slot.lock: + current = _slot.status + if current is None or current.key != key: + return # superseded by a later job; its lines are not ours to add to + _slot.status = replace(current, lines=(current.lines + (text,))[-_MAX_LINES:]) + + def _body() -> None: + try: + run(_append) + except Exception as exc: + with _slot.lock: + current = _slot.status + if current is not None and current.key == key: + _slot.status = replace( + current, + state=FAILED, + finished_ts=time.time(), + error=f"{type(exc).__name__}: {exc}", + ) + return + with _slot.lock: + current = _slot.status + if current is not None and current.key == key: + _slot.status = replace(current, state=DONE, finished_ts=time.time()) + + thread = threading.Thread(target=_body, name=f"keel-job-{key}", daemon=True) + with _slot.lock: + _slot.thread = thread + thread.start() + return True + + +def wait(timeout: float | None = None) -> JobStatus | None: + """Block until the running job finishes. For TESTS and for a caller that genuinely has + nothing else to do -- never for a request handler, which is the entire reason this module + exists.""" + with _slot.lock: + thread = _slot.thread + if thread is not None: + thread.join(timeout) + return status() + + +def reset() -> None: + """Forget the slot. For tests: a module-level slot that persisted between them would make one + test's job visible to the next.""" + with _slot.lock: + _slot.status = None + _slot.thread = None diff --git a/keel/commands/setup.py b/keel/commands/setup.py index e7402d4..856ad3f 100644 --- a/keel/commands/setup.py +++ b/keel/commands/setup.py @@ -831,6 +831,72 @@ def store_market_data_credential( return ActionResult("credentials", True, "saved to the OS keychain") +#: Years of history the first fetch ensures. The CLI's own `--years` default: a first run should +#: land in the same state a `keel fetch` would, not a thinner one that quietly changes what a +#: backtest is measured over. +FIRST_FETCH_YEARS = 5 + + +def fetch_market_data(config_path: Path, db_path: Path, _values: dict[str, str]) -> ActionResult: + """Start the first candle fetch IN THE BACKGROUND, and return immediately. + + This is the one action that cannot be synchronous. A first fetch runs for minutes across the + allowlist; a request that blocks that long is not a button -- the browser gives up, the user + reloads, and a second fetch starts on top of the first. `keel.commands.jobs` owns the single + slot that makes the second one a refusal instead. + + `run_fetch` needs no adapting: its `echo` parameter is documented as the progress stream and + emits the same lines the CLI prints, and its `build_client` is a lazy factory, so nothing + constructs a broker until the fetch actually needs one. + """ + from keel.commands import jobs + + if jobs.is_running(): + return ActionResult("market_data", False, "a job is already running") + + def _run(echo: Callable[[str], None]) -> None: + import time as _time + + from keel.commands._common import _build_broker + from keel.commands._products import parse_products_option + from keel.commands.fetch import run_fetch + from keel.config import load_config + from keel.data.db import connect + from keel.data.repository import Repository + from keel.data import freshness as freshness_mod + + config = load_config(str(config_path)) + products, _warnings = parse_products_option(None, config) + # Its own connection: this runs on a background thread, and a sqlite3 connection belongs + # to the thread that made it. + conn = connect(str(db_path)) + try: + result = run_fetch( + Repository(conn), + config, + lambda: _build_broker(config), + db_path=str(db_path), + products=products, + years=FIRST_FETCH_YEARS, + now_ts=int(_time.time()), + tolerance_bars=freshness_mod.DEFAULT_TOLERANCE_BARS, + echo=echo, + echo_err=echo, + ) + finally: + conn.close() + if result.error is not None: + # RAISED, so the job records it as a failure. `run_fetch` returns the error rather + # than raising because how a front-end fails is its business -- and this front-end + # fails by showing a failed job, which is what the operator needs to see. + raise RuntimeError(result.error) + + jobs.start("market_data", _run) + return ActionResult( + "market_data", True, "fetching in the background -- this page will show its progress" + ) + + #: THE closed set of steps a machine may perform on the operator's behalf. ACTIONS: tuple[Action, ...] = ( Action( @@ -871,19 +937,22 @@ def store_market_data_credential( ActionInput("CDP_API_SECRET", "CDP API secret", secret=True), ), ), + Action( + key="market_data", + title="Fetch market data", + detail=( + "Downloads candle history for every allowlisted product. This runs in the background " + "and takes minutes on a first run; the page shows its progress." + ), + run=fetch_market_data, + ), ) -#: Mechanical steps that are deliberately NOT offered as one-click actions, and why. Recorded as -#: data rather than omitted silently, so the gap is visible to the next person rather than -#: looking like an oversight. -NOT_AUTOMATED_YET: dict[str, str] = { - "market_data": ( - "The first fetch is a network call that can run for minutes across the allowlist. A " - "request that blocks that long is not a button, it is a background job with progress " - "and cancellation -- so it stays `keel fetch` until there is somewhere for such a job " - "to live." - ), -} +#: Mechanical steps deliberately NOT offered as one-click actions, and why. Recorded as data +#: rather than omitted silently, so a gap is visible to the next person rather than looking like +#: an oversight. Empty is a fine value: `market_data` lived here until `keel.commands.jobs` gave +#: a long-running step somewhere to live. +NOT_AUTOMATED_YET: dict[str, str] = {} def action_for(key: str) -> Action | None: diff --git a/keel/web/render.py b/keel/web/render.py index e71e446..4d921fd 100644 --- a/keel/web/render.py +++ b/keel/web/render.py @@ -89,6 +89,7 @@ footer { border-top: 1px solid var(--line); color: var(--muted); font-size: 0.8rem; padding: 1rem 1.25rem; } pre { white-space: pre-wrap; word-break: break-word; margin: 0; font-size: 0.85rem; } +.job pre { margin-top: 0.6rem; max-height: 22rem; overflow-y: auto; } form { margin: 0.5rem 0 0; } .field { display: flex; flex-direction: column; gap: 0.2rem; margin: 0.6rem 0; max-width: 26rem; } .field span { font-size: 0.8rem; color: var(--muted); } @@ -620,6 +621,33 @@ def render_venues(infos: Sequence[Any]) -> str: } +def _job_panel(job: Any) -> str: + """A running, finished or failed background job, as a panel. + + The progress lines are shown NEWEST LAST, unscrolled, exactly as the CLI prints them -- an + operator who has run `keel fetch` in a terminal should recognise what they are looking at + rather than have to learn a second vocabulary for the same thing. + + A failure stays on screen. The whole point of running something in the background is that + nobody was watching when it broke.""" + tone = {"running": "warn", "done": "good", "failed": "bad"}.get(job.state, "muted") + elapsed = f"{int(job.elapsed_sec)}s" + parts = [ + '
', + f'
{esc(job.key)}' + f'{esc(job.state)}
', + f'

{esc(elapsed)} elapsed

', + ] + if job.error: + parts.append(f'

{esc(job.error)}

') + if job.lines: + parts.append("
" + esc("\n".join(job.lines)) + "
") + elif job.is_running: + parts.append('

starting…

') + parts.append("
") + return "".join(parts) + + def render_setup( state: Any, *, @@ -627,6 +655,7 @@ def render_setup( not_automated: dict[str, str] | None = None, csrf: str = "", ran: str = "", + job: Any = None, ) -> str: """The first-run checklist, with a button for each MECHANICAL step and a command for every other one. @@ -658,6 +687,9 @@ def render_setup( "run looks like. Work down the list; the paper stage places no orders at all." "" ) + if job is not None: + parts.append(_job_panel(job)) + nxt = state.next_step if nxt is not None: parts.append( @@ -693,7 +725,12 @@ def render_setup( f'
{esc(item.detail)}
' ) action = by_key.get(item.step.key) - if item.blocking and action is not None and csrf: + # A running job owns its step: offering the button again would invite a second start + # that the job slot refuses anyway, which reads as the page ignoring the click. + running_here = job is not None and job.is_running and job.key == item.step.key + if running_here: + body += '
running — see the panel above
' + elif item.blocking and action is not None and csrf: body += _action_form(action, csrf) elif item.blocking: body += f"
{esc(item.step.how)}
" diff --git a/keel/web/server.py b/keel/web/server.py index 4aa7d38..582254b 100644 --- a/keel/web/server.py +++ b/keel/web/server.py @@ -64,6 +64,11 @@ #: an action key and a token. _MAX_FORM_BYTES = 8 * 1024 +#: How often the setup page reloads WHILE a background job runs. Shorter than the dashboards' +#: 15s: someone watching a fetch wants to see it moving, and the page is a few kilobytes of local +#: HTML. +_JOB_REFRESH_SEC = 5 + #: Journal rows rendered on the insights page. A cap, not a paginator: the page answers "how has #: this been going", and the full history is what `keel insights journal` is for. _JOURNAL_LIMIT = 50 @@ -153,8 +158,10 @@ def _deployment_state(cfg: ServeConfig) -> Any: def page_setup(cfg: ServeConfig, query: dict[str, list[str]]) -> tuple[str, str, int | None]: + from keel.commands import jobs from keel.commands.setup import ACTIONS, NOT_AUTOMATED_YET + job = jobs.status() return ( "Setup", render.render_setup( @@ -163,8 +170,12 @@ def page_setup(cfg: ServeConfig, query: dict[str, list[str]]) -> tuple[str, str, not_automated=NOT_AUTOMATED_YET, csrf=csrf_token(cfg.token), ran=(query.get("ran") or [""])[0], + job=job, ), - None, + # Auto-refresh ONLY while something is running. A finished page that kept reloading would + # fight a reader, and the zero-JS meta refresh is the only progress mechanism available + # to a page that ships no scripts. + _JOB_REFRESH_SEC if job is not None and job.is_running else None, ) diff --git a/tests/commands/test_jobs.py b/tests/commands/test_jobs.py new file mode 100644 index 0000000..512d0e6 --- /dev/null +++ b/tests/commands/test_jobs.py @@ -0,0 +1,126 @@ +"""One background job at a time (#437). + +The first market-data fetch runs for minutes across an allowlist. A request that blocks that long +is not a button -- the browser gives up, the user reloads, and a second fetch starts on top of the +first. These pin the properties that make the single slot safe rather than merely simple. +""" + +from __future__ import annotations + +import threading + +import pytest + +from keel.commands import jobs + + +@pytest.fixture(autouse=True) +def _clean_slot() -> None: + """A module-level slot that persisted between tests would make one test's job visible to the + next.""" + jobs.reset() + yield + jobs.wait(5) + jobs.reset() + + +def test_a_job_runs_and_reports_its_progress() -> None: + def work(echo): + echo("one") + echo("two") + + assert jobs.start("probe", work) is True + status = jobs.wait(5) + assert status is not None + assert status.state == jobs.DONE + assert status.lines == ("one", "two") + assert status.finished_ts is not None + + +def test_a_second_start_is_refused_while_one_runs() -> None: + """Not queued, not silently dropped: both look identical to a user watching a page that is + not changing. Two concurrent fetches also write candles to one SQLite database and race.""" + gate = threading.Event() + + def slow(echo): + echo("working") + gate.wait(5) + + assert jobs.start("first", slow) is True + assert jobs.is_running() + assert jobs.start("second", lambda echo: None) is False + assert jobs.status().key == "first" + gate.set() + assert jobs.wait(5).state == jobs.DONE + + +def test_a_failure_is_captured_and_stays_visible() -> None: + """The whole point of running something in the background is that nobody was watching when it + broke, so the failure has to still be there when they look. And it must not escape onto a + thread that belongs to nobody.""" + + def boom(echo): + echo("about to fail") + raise ValueError("the venue said no") + + jobs.start("failing", boom) + status = jobs.wait(5) + assert status.state == jobs.FAILED + assert status.error == "ValueError: the venue said no" + assert status.lines == ("about to fail",) + # Still there on a second read -- not cleared by being observed. + assert jobs.status().state == jobs.FAILED + + +def test_the_error_is_a_type_and_message_not_a_traceback() -> None: + """A traceback in a browser page is a stack of file paths from someone else's machine.""" + + def boom(_echo): + raise RuntimeError("plain message") + + jobs.start("failing", boom) + error = jobs.wait(5).error + assert error == "RuntimeError: plain message" + assert "Traceback" not in error + assert "/" not in error + + +def test_a_finished_job_lets_the_next_one_start() -> None: + jobs.start("first", lambda echo: echo("done")) + jobs.wait(5) + assert not jobs.is_running() + assert jobs.start("second", lambda echo: echo("also done")) is True + assert jobs.wait(5).key == "second" + + +def test_the_progress_tail_is_bounded() -> None: + """A long run must not hold its whole output in a process that is also serving pages.""" + + def chatty(echo): + for index in range(jobs._MAX_LINES * 3): + echo(f"line {index}") + + jobs.start("chatty", chatty) + status = jobs.wait(10) + assert len(status.lines) == jobs._MAX_LINES + # The TAIL, not the head: what is happening now is what a watcher needs. + assert status.lines[-1] == f"line {jobs._MAX_LINES * 3 - 1}" + + +def test_blank_progress_lines_are_dropped() -> None: + """`run_fetch` emits blank separator lines the CLI renders as spacing; in a bounded tail they + would push real output out of view.""" + + def work(echo): + echo("real") + echo("") + echo(" ") + echo("also real") + + jobs.start("spaced", work) + assert jobs.wait(5).lines == ("real", "also real") + + +def test_nothing_has_run_reads_as_none() -> None: + assert jobs.status() is None + assert jobs.is_running() is False diff --git a/tests/web/test_server.py b/tests/web/test_server.py index 8f81579..eb66692 100644 --- a/tests/web/test_server.py +++ b/tests/web/test_server.py @@ -819,3 +819,100 @@ def _capture(_config: object, _db: object, values: dict[str, str]) -> object: }, ) assert set(seen) == {"CDP_API_KEY", "CDP_API_SECRET"} + + +# -- the background job, on the page ----------------------------------------------------------- + + +def test_the_setup_page_refreshes_only_while_a_job_runs( + empty_machine: web_server.ServeConfig, monkeypatch: pytest.MonkeyPatch +) -> None: + """A finished page that kept reloading would fight a reader; a running one that did not + would be a progress display that never progresses. The zero-JS meta refresh is the only + mechanism available to a page that ships no scripts.""" + from keel.commands import jobs + + jobs.reset() + _status, _headers, idle = _request(empty_machine, "/setup", cookie=_session(empty_machine)) + assert 'http-equiv="refresh"' not in idle + + gate = threading.Event() + jobs.start("market_data", lambda echo: (echo("fetching BTC-USD"), gate.wait(5))) + try: + _status, _headers, running = _request( + empty_machine, "/setup", cookie=_session(empty_machine) + ) + assert 'http-equiv="refresh"' in running + assert "fetching BTC-USD" in running + assert "running" in running + finally: + gate.set() + jobs.wait(5) + + _status, _headers, finished = _request(empty_machine, "/setup", cookie=_session(empty_machine)) + assert 'http-equiv="refresh"' not in finished + assert "done" in finished + jobs.reset() + + +def test_a_failed_job_is_shown_on_the_page_and_stays( + empty_machine: web_server.ServeConfig, +) -> None: + """Nobody was watching when it broke. If the page did not still say so, nothing would.""" + from keel.commands import jobs + + jobs.reset() + + def boom(_echo): + raise RuntimeError("the venue said no") + + jobs.start("market_data", boom) + jobs.wait(5) + for _ in range(2): + _status, _headers, body = _request(empty_machine, "/setup", cookie=_session(empty_machine)) + assert "failed" in body + assert "the venue said no" in body + jobs.reset() + + +def test_starting_market_data_returns_immediately( + empty_machine: web_server.ServeConfig, monkeypatch: pytest.MonkeyPatch +) -> None: + """The property the whole module exists for: the POST must not wait for the fetch.""" + from keel.commands import jobs + from keel.commands import setup as setup_mod + + jobs.reset() + gate = threading.Event() + started = threading.Event() + + def _slow_action(_config, _db, _values): + jobs.start("market_data", lambda echo: (started.set(), gate.wait(5))) + return setup_mod.ActionResult("market_data", True, "started") + + monkeypatch.setattr( + setup_mod, + "ACTIONS", + tuple( + a + if a.key != "market_data" + else type(a)(a.key, a.title, a.detail, _slow_action, a.inputs) + for a in setup_mod.ACTIONS + ), + ) + try: + status, headers, _body = _request( + empty_machine, + "/setup/market_data", + method="POST", + cookie=_session(empty_machine), + form={"csrf": _csrf(empty_machine)}, + ) + assert status == 303 + assert headers["Location"] == "/setup?ran=market_data" + assert started.wait(5), "the job never started" + assert jobs.is_running(), "the request returned before the job finished, as intended" + finally: + gate.set() + jobs.wait(5) + jobs.reset()