From 988c45a11bdd317f9fc8413b6857f7da551203d5 Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Thu, 20 Aug 2026 18:00:26 -0400 Subject: [PATCH] feat(setup): market data as a background job, so the last mechanical step can be a button MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This was the one step left in `NOT_AUTOMATED_YET`, with the reason recorded there: "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". This builds that somewhere. `keel/commands/jobs.py` is deliberately the smallest thing that works: 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 -- which look identical to a user watching a page that is not changing. A FAILED JOB STAYS VISIBLE. Not cleared on read, not cleared by time. The entire 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. The error recorded is 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. The progress is a bounded tail of the same lines the CLI prints. `run_fetch` needed no adapting at all: its `echo` parameter is already documented as the progress stream, and its `build_client` is already a lazy factory, so nothing constructs a broker until the fetch needs one. Blank separator lines are dropped -- in a bounded tail they would push real output out of view. The page auto-refreshes ONLY while something is running. 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 remains the only mechanism available to a page that ships no scripts. And the step's button disappears while its job runs: offering it again would invite a second start that the slot refuses anyway, which reads as the page ignoring the click. VERIFIED END TO END against a live server, on the realistic first-run mistake -- starting a fetch with no credentials configured: POST /setup/market_data -> 303, immediately job panel -> market_data · failed · AuthenticationError: Unauthenticated request to private endpoint. If you wish to access private endpoints, you must provide your API key and secret ... MISSING BTC-USD ONE_DAY refresh -> present while running, absent once finished That is the venue's own error, in front of the operator, with the progress that preceded it -- and the checklist directly above it already names the credential step. MYPY CAUGHT A BUG NO TEST WOULD HAVE. The fetch closure imported `keel.strategy.freshness`; the module is `keel.data.freshness`. Every import in that closure is lazy and runs only on the background thread, so the suite was green and the job would have died with a ModuleNotFoundError the first time anyone pressed the button. 4083 passed, 3 skipped (12 new). ruff clean repo-wide; mypy clean over keel + packages. `NOT_AUTOMATED_YET` is now empty, which is a fine value for it -- and it stays, because the next step someone decides not to automate should have somewhere to say why. Refs #437, #18. Co-Authored-By: Claude Opus 5 (1M context) --- keel/commands/jobs.py | 150 ++++++++++++++++++++++++++++++++++++ keel/commands/setup.py | 91 +++++++++++++++++++--- keel/web/render.py | 39 +++++++++- keel/web/server.py | 13 +++- tests/commands/test_jobs.py | 126 ++++++++++++++++++++++++++++++ tests/web/test_server.py | 97 +++++++++++++++++++++++ 6 files changed, 503 insertions(+), 13 deletions(-) create mode 100644 keel/commands/jobs.py create mode 100644 tests/commands/test_jobs.py 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()