feat(setup): market data as a background job, so the last mechanical step can be a button

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

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:

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,13 +621,41 @@ 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 = [
+        '<div class="card job">',
+        f'<div class="kv"><span class="k">{esc(job.key)}</span>'
+        f'<span class="v {tone}">{esc(job.state)}</span></div>',
+        f'<p class="note">{esc(elapsed)} elapsed</p>',
+    ]
+    if job.error:
+        parts.append(f'<p class="bad"><strong>{esc(job.error)}</strong></p>')
+    if job.lines:
+        parts.append("<pre>" + esc("\n".join(job.lines)) + "</pre>")
+    elif job.is_running:
+        parts.append('<p class="muted">starting…</p>')
+    parts.append("</div>")
+    return "".join(parts)
+
+
 def render_setup(
     state: Any,
     *,
     actions: Sequence[Any] = (),
     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.</span>"
             "</div>"
         )
+    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'<div class="muted">{esc(item.detail)}</div>'
             )
             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 += '<div class="muted">running — see the panel above</div>'
+            elif item.blocking and action is not None and csrf:
                 body += _action_form(action, csrf)
             elif item.blocking:
                 body += f"<div><code>{esc(item.step.how)}</code></div>"

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,
     )