From b5b1cd740be8806dd09c4abdf8784c25e24fb842 Mon Sep 17 00:00:00 2001 From: Brett Williams Date: Sun, 30 Aug 2026 21:59:46 -0500 Subject: [PATCH] Stream live worker progress via async build dispatch The agent's /upload and /checkout_git were synchronous, blocking until a build finished, so the controller could not relay live per-step progress. Add ?async=1 support: the agent saves/clones the source, starts the build in a background thread (holding build_lock), and returns 202 + {"id": ...} immediately. The controller dispatches async, then polls /progress/, appending each step/log line to the job log and SSE stream, and treats the build as done when progress clears (404) and the worker reports ready. --- DESIGN.md | 27 +++++++------ cross_py_builder/build_agent.py | 36 +++++++++++++++-- ctrl/scheduler.py | 71 +++++++++++++++++++++++++-------- 3 files changed, 102 insertions(+), 32 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index c037f29..232d90c 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -44,8 +44,8 @@ trusted-LAN posture as today). The worker endpoint contract used by the controll | Worker endpoint | Method | Purpose | |---|---|---| | `/status` | GET | readiness, os, cpu, agent_version, hostname, ip | -| `/upload` | POST | multipart `file` zip → starts build, returns job JSON | -| `/checkout_git` | POST | JSON `{"repo_url": ...}` → starts build | +| `/upload` | POST | multipart `file` zip → starts build, returns job JSON. `?async=1` returns `202` + `{"id": ...}` immediately, building in background | +| `/checkout_git` | POST | JSON `{"repo_url": ...}` → starts build. `?async=1` behaves like `/upload?async=1` | | `/download/` | GET | zip of `dist/` for a built job | | `/delete_cache` | GET | clear cached jobs | | `/healthz` | GET | liveness (already added) | @@ -108,17 +108,18 @@ A single background thread owns all dispatch work. - exclude the local/controller LXC if present, to avoid self-builds 3. Choose first match (optionally: prefer workers with more free disk — v1: first match). 4. Mark job `dispatching`; set `worker_host`. -5. Send to worker: - - upload → `POST /upload` with the stored file - - git → `POST /checkout_git` with `{repo_url}` - - on success, read the worker job JSON → store `worker_job_id`, `worker_url`, set `building` - - on hard failure, mark `failed` (§ failure handling below) -6. **Progress relay** — while the dispatch call is in flight (the worker is building), poll - `GET /progress/` on an interval and append each snapshot to the job log + - push it to connected SSE clients. This gives live per-step progress in the UI, enabled by - the small agent addition (§1). -7. On success (dispatch call returns): compute download URL `worker_url + /download/`, - fetch the zip, save to `builds//`, record artifact in the job row, set `done`. + 5. Send to worker (**async**): `POST /upload?async=1` (with the stored file) or + `POST /checkout_git?async=1` (with `{repo_url}`). The agent clones/saves the source, then + starts the build in a background thread (holding its `build_lock`) and returns `202` + immediately with `{"id": }`. On hard failure, mark `failed` (§ failure handling + below). Store `worker_job_id`, `worker_url`, set `building`. + 6. **Progress relay** — while the build runs, poll `GET /progress/` on an interval, + appending each `last_log_line`/step snapshot to the job log and pushing to connected SSE + clients → live per-step progress in the UI. The build is complete when `/progress` returns + `404` (progress cleared) and the worker's `/status` reports `ready`. (This relies on the small + agent additions in §1.) + 7. On completion: compute download URL `worker_url + /download/`, fetch the zip, + save to `builds//`, record artifact in the job row, set `done`. **Concurrency/limits:** scheduler processes one job at a time (workers can already build only one at a time — `409`). Multiple queued jobs simply wait. This makes NIC/disk behavior predictable diff --git a/cross_py_builder/build_agent.py b/cross_py_builder/build_agent.py index 4155f8f..c109c21 100755 --- a/cross_py_builder/build_agent.py +++ b/cross_py_builder/build_agent.py @@ -220,6 +220,7 @@ def checkout_project(): if not build_lock.acquire(blocking=False): return jsonify({'error': f"Another build is already running: {system_status['running_job']}"}), 409 + async_mode = _is_async_request() repo_url = request.json['repo_url'] print(f"\n========== Checking Out Git Project ==========") @@ -233,6 +234,9 @@ def checkout_project(): system_status['status'] = "cloning_repo" subprocess.check_call(['git', 'clone', repo_url, repo_dir]) system_status['status'] = "ready" + if async_mode: + _run_build_background(job_id, repo_dir, start_time) + return jsonify({"id": job_id, "async": True, "message": "Build started"}), 202 return install_and_build(repo_dir, job_id, start_time) except Exception as e: print(f"Error processing checkout: {e}") @@ -244,7 +248,26 @@ def checkout_project(): return jsonify({'error': 'Failed to clone repository'}), 500 return jsonify({'error': f"Uncaught error processing checkout: {e}"}), 500 finally: - build_lock.release() + if not async_mode: + build_lock.release() + +def _run_build_background(job_id, project_path, start_time): + """Run install_and_build in a background thread. Assumes build_lock is held; + the thread releases it when the build finishes.""" + def runner(): + try: + install_and_build(project_path, job_id, start_time) + except Exception as e: + print(f"Background build failed for {job_id}: {e}") + _clear_progress(job_id) + finally: + build_lock.release() + threading.Thread(target=runner, name=f"build-{job_id}", daemon=True).start() + + +def _is_async_request(): + return request.args.get("async") in ("1", "true", "yes", "on") + def safe_extract(zip_ref, dest_dir): """Extract a zipfile, rejecting members that would escape dest_dir.""" @@ -260,6 +283,8 @@ def upload_project(): if not build_lock.acquire(blocking=False): return jsonify({'error': f"Another build is already running: {system_status['running_job']}"}), 409 + async_mode = _is_async_request() + working_dir = None try: start_time = datetime.datetime.now() if 'file' not in request.files: @@ -283,15 +308,20 @@ def upload_project(): print(f"Extracting uploaded project zip...") safe_extract(zip_ref, working_dir) + if async_mode: + _run_build_background(job_id, working_dir, start_time) + return jsonify({"id": job_id, "async": True, "message": "Build started"}), 202 + return install_and_build(working_dir, job_id, start_time) except Exception as e: print(f"Uncaught error processing job: {e}") system_status['status'] = "ready" _clear_progress(job_id) - shutil.rmtree(working_dir, ignore_errors=True) + shutil.rmtree(working_dir, ignore_errors=True) if working_dir else None return jsonify({"error": f"Uncaught error processing job: {e}"}), 500 finally: - build_lock.release() + if not async_mode: + build_lock.release() def install_and_build(project_path, job_id, start_time): _set_step(job_id, "starting", project_path=project_path) diff --git a/ctrl/scheduler.py b/ctrl/scheduler.py index db1e40f..57dfb60 100644 --- a/ctrl/scheduler.py +++ b/ctrl/scheduler.py @@ -1,6 +1,7 @@ import datetime import os import threading +import time import requests @@ -97,21 +98,22 @@ class Scheduler: try: if job["source_type"] == "git": - worker_job = self._request_git(job, worker) + worker_job_id = self._request_git(job, worker) else: - worker_job = self._request_upload(job, worker) + worker_job_id = self._request_upload(job, worker) except Exception as e: db.update_job(job_id, status="failed", finished_at=datetime.datetime.now().isoformat(), error=str(e)) self._append(job_id, f"[scheduler] FAILED: {e}\n") return - worker_job_id = worker_job.get("id") worker_url = worker["url"] db.update_job(job_id, status="building", worker_job_id=worker_job_id, worker_url=worker_url) + self._append(job_id, f"[scheduler] Building on {worker_url} (worker job {worker_job_id})\n") - # download artifacts + # Poll progress until the worker reports the build finished. try: + self._wait_for_build(job, worker, worker_job_id) self._fetch_artifacts(job, worker, worker_job_id) db.update_job(job_id, status="done", finished_at=datetime.datetime.now().isoformat()) self._append(job_id, "[scheduler] Build complete, artifacts saved.\n") @@ -123,28 +125,65 @@ class Scheduler: self._publish(job["id"], "job", '"done"') def _request_git(self, job, worker): - self._append(job["id"], f"[scheduler] Cloning {job['source']} on {worker['url']}\n") - resp = requests.post(f"{worker['url']}/checkout_git", + job_id = job["id"] + self._append(job_id, f"[scheduler] Cloning {job['source']} on {worker['url']}\n") + resp = requests.post(f"{worker['url']}/checkout_git?async=1", json={"repo_url": job["source"]}, - timeout=settings.WORKER_BUILD_TIMEOUT, + timeout=settings.WORKER_STATUS_TIMEOUT, stream=False) - return self._worker_response(resp, "checkout") + data = self._worker_response(resp, "checkout") + if not data.get("id"): + raise RuntimeError("checkout did not return a job id") + return data["id"] def _request_upload(self, job, worker): + job_id = job["id"] path = self._source_path(job) if not path or not os.path.exists(path): raise FileNotFoundError(f"Source file not found: {path}") - self._append(job["id"], f"[scheduler] Uploading {os.path.basename(path)} to {worker['url']}\n") + self._append(job_id, f"[scheduler] Uploading {os.path.basename(path)} to {worker['url']}\n") with open(path, "rb") as f: - resp = requests.post(f"{worker['url']}/upload", + resp = requests.post(f"{worker['url']}/upload?async=1", files={"file": (os.path.basename(path), f, "application/zip")}, - timeout=settings.WORKER_BUILD_TIMEOUT, + timeout=settings.WORKER_STATUS_TIMEOUT, stream=False) - worker_job = self._worker_response(resp, "upload") - if not worker_job.get("id"): - # Agent may have started building before returning; try progress fallback. - pass - return worker_job + data = self._worker_response(resp, "upload") + if not data.get("id"): + raise RuntimeError("upload did not return a job id") + return data["id"] + + def _wait_for_build(self, job, worker, worker_job_id): + """Poll worker progress until the build completes (progress cleared + worker ready).""" + job_id = job["id"] + deadline = datetime.datetime.now() + datetime.timedelta(seconds=settings.WORKER_BUILD_TIMEOUT) + while datetime.datetime.now() < deadline: + progress = self._probe_progress(job, worker["url"], worker_job_id) + if progress is None: + status = wrk.probe_worker(worker["url"]) + if status and status.get("status") == "ready": + return + # progress gone but worker not ready: brief window, keep waiting + time.sleep(settings.SCHEDULER_INTERVAL) + continue + line = progress.get("last_log_line") or progress.get("step") or "" + if line: + self._append(job_id, f"[worker] {line}\n") + time.sleep(settings.SCHEDULER_INTERVAL) + raise RuntimeError(f"Worker did not finish in {settings.WORKER_BUILD_TIMEOUT}s") + + @staticmethod + def _probe_progress(job, worker_url, worker_job_id): + try: + resp = requests.get(f"{worker_url}/progress/{worker_job_id}", + timeout=settings.WORKER_STATUS_TIMEOUT) + except requests.RequestException: + return None + if resp.status_code == 404: + return None + try: + return resp.json() + except ValueError: + return None @staticmethod def _worker_response(resp, action):