Pipeline the firmware queue: concurrent compile + upload lanes

CLAUDE.md

@@ -281,8 +281,19 @@ against legacy behaviour before assuming the simpler version suffices.
   emitter and walks `models.*` to assert coverage. New events ship with a
   TypedDict from day one and a row in `_PAYLOAD_FACTORIES`. Full
   rationale in `docs/ARCHITECTURE.md` "Event bus → Typing event payloads".
-- **Persistent firmware queue.** One job runs at a time; queue + output
-  buffers survive restarts. See `controllers/firmware.py`.
+- **Persistent firmware queue — two concurrent lanes.** A CPU/compile lane
+  and a network/upload lane each run one job at a time, but run
+  concurrently, so a slow upload doesn't block the next compile (#3702).
+  `FirmwareState.compile_lane` / `upload_lane` (`controllers/firmware/
+  _state.py`); `lane_for(job)` routes UPLOAD to the upload lane, everything
+  else to compile. `firmware/install` enqueues a COMPILE job + a dependent
+  local UPLOAD job (`FirmwareJob.depends_on`); the upload is held off its
+  lane until the compile succeeds (`lifecycle.release_dependents`), and a
+  cancelled/failed compile cascades to cancel the held upload (so a
+  cancelled build never flashes). Remote install uses the same chain — a
+  remote compile materialises artifacts locally, then the local upload lane
+  flashes. Queue + output buffers survive restarts. See
+  `controllers/firmware/`.
 - **Component catalog is generated**, not hand-edited. Source is
   ESPHome's pre-built schema bundle (https://schema.esphome.io) plus
   narrow live `esphome` introspection for what the schema doesn't carry

docs/API.md

@@ -137,7 +137,7 @@ Connections that arrive on the trusted ingress site (HA add-on supervisor proxy)
 |---------|------|----------|-------------|
 | `firmware/compile` | `{configuration}` | `FirmwareJob` | Queue compile job |
 | `firmware/upload` | `{configuration, port?: ""}` | `FirmwareJob` | Queue upload of existing binary. `port` defaults to `""` (no `--device` arg — CLI auto-detects). Also accepts `"OTA"`, a serial path (`/dev/ttyUSB0`, `COM3`), or an explicit IP / hostname for "install to a specific address" — the address-cache shortcut is bypassed when a target is named directly. |
-| `firmware/install` | `{configuration, port?: "OTA" \| serial \| ip \| hostname, force_local?: bool}` | `FirmwareJob` | Queue compile + upload. `port` defaults to `"OTA"` (let the CLI resolve the configured host). Same `port` semantics as `firmware/upload` for non-default values. `force_local` defaults to `false`; when `true` the scheduler decision is bypassed and the install runs LOCAL regardless of paired build servers — used by the install dialog's "Build locally instead" override link to opt out of REMOTE routing for a single install. |
+| `firmware/install` | `{configuration, port?: "OTA" \| serial \| ip \| hostname, force_local?: bool}` | `FirmwareJob` (the COMPILE job) | Queue an install as a **two-job chain**: a `COMPILE` job + a dependent `UPLOAD` job (`FirmwareJob.depends_on` = the compile's `job_id`). Returns the COMPILE job; the UPLOAD renders as queued and starts only after the compile succeeds, on the **upload lane** so it doesn't block the next device's compile. A cancelled/failed compile cascades to cancel the held upload (a cancelled build never flashes). `port` (defaults `"OTA"`) lands on the UPLOAD job. `force_local=true` bypasses the scheduler (compile runs LOCAL). Remote installs use the same chain — the remote compile materialises artifacts locally, then the local upload lane flashes. |
 | `firmware/clean` | `{configuration}` | `FirmwareJob` | Queue build clean for one device |
 | `firmware/reset_build_env` | — | `FirmwareJob` | Queue full reset of `.esphome/` build dirs and PIO cache |
 | `firmware/compile_bulk` | `{configurations: string[]}` | `[FirmwareJob]` | Queue multiple compiles |

docs/ARCHITECTURE.md

@@ -196,7 +196,12 @@ firmware/install {configuration} → QUEUED → RUNNING → output... → COMPLE
                                      └──── persisted to disk ─────────────┘
 ```
 
-- One job runs at a time, others wait in queue
+- Two concurrent lanes — a compile lane (CPU) and an upload lane (network).
+  Each runs one job at a time, but the lanes run in parallel so a slow
+  network flash doesn't block the next device's compile (#3702). `install`
+  splits into a COMPILE job + a dependent local UPLOAD job (`depends_on`):
+  the upload is held until the compile succeeds, then runs on the upload
+  lane; a cancelled/failed compile cascades to cancel the held upload.
 - Output buffered in `FirmwareJob.output` — survives disconnect
 - `firmware/follow_job` sends history then streams live
 - Error detection scans output for failure patterns (not just exit code)

esphome_device_builder/controllers/firmware/_state.py

@@ -5,7 +5,21 @@
 import asyncio
 from dataclasses import dataclass, field
 
-from ...models import FirmwareJob
+from ...models import FirmwareJob, JobStatus, JobType
+
+
+@dataclass
+class Lane:
+    """One serial work lane: its FIFO queue + the job/subprocess on it now.
+
+    Two lanes run concurrently — a compile lane (CPU) and an upload lane
+    (network) — so a slow network flash doesn't block the next compile.
+    Within a lane work stays serialized (``current_job`` is the single slot).
+    """
+
+    queue: asyncio.Queue[FirmwareJob] = field(default_factory=asyncio.Queue)
+    current_job: FirmwareJob | None = None
+    current_process: asyncio.subprocess.Process | None = None
 
 
 @dataclass
@@ -18,10 +32,11 @@ class FirmwareState:
     # picks first. ``cli`` reads it to build the subprocess argv.
     esphome_cmd: list[str] = field(default_factory=list)
 
-    # Persistent single-job queue. Producer is ``_enqueue``;
-    # consumer is the runner loop. Survives restarts via the
-    # on-disk persistence layer.
-    queue: asyncio.Queue[FirmwareJob] = field(default_factory=asyncio.Queue)
+    # The two concurrent lanes. Producers enqueue onto the lane a job's
+    # type maps to (``_lane_for``); each lane has its own consumer task.
+    # Both survive restarts via the on-disk persistence layer.
+    compile_lane: Lane = field(default_factory=Lane)
+    upload_lane: Lane = field(default_factory=Lane)
 
     # Active + recent jobs keyed by ``job_id``. ``persistence``
     # reads / writes on every state transition; ``clean``,
@@ -30,12 +45,6 @@ class FirmwareState:
     # ``persistence._prune_history``.
     jobs: dict[str, FirmwareJob] = field(default_factory=dict)
 
-    # Single-job runner slot — ``None`` when idle. ``runner``
-    # reads / writes as the job lifecycle transitions through
-    # the queue.
-    current_job: FirmwareJob | None = None
-    current_process: asyncio.subprocess.Process | None = None
-
     # Job ids the user asked to cancel; the runner consults this
     # on subprocess exit to mark CANCELLED instead of FAILED.
     cancel_requested: set[str] = field(default_factory=set)
@@ -45,3 +54,18 @@ class FirmwareState:
     # frame unblocks instantly. The local subprocess path uses
     # SIGTERM instead and doesn't register here.
     cancel_events: dict[str, asyncio.Event] = field(default_factory=dict)
+
+    def lane_for(self, job: FirmwareJob) -> Lane:
+        """Return the lane *job* runs on: UPLOAD on the network lane, else the compile lane."""
+        return self.upload_lane if job.job_type is JobType.UPLOAD else self.compile_lane
+
+    def place_on_lane(self, job: FirmwareJob) -> None:
+        """Put *job* onto the lane its type maps to, ready for that lane's consumer."""
+        self.lane_for(job).queue.put_nowait(job)
+
+    def dependency_satisfied(self, job: FirmwareJob) -> bool:
+        """Return whether *job* has no prerequisite, or its prerequisite has completed."""
+        if not job.depends_on:
+            return True
+        prereq = self.jobs.get(job.depends_on)
+        return prereq is not None and prereq.status is JobStatus.COMPLETED

esphome_device_builder/controllers/firmware/bulk.py

@@ -8,6 +8,7 @@
 
 from ...helpers.api import CommandError
 from ...models import ErrorCode, FirmwareJob, JobType
+from . import factories
 from .helpers import _validate_port
 
 if TYPE_CHECKING:
@@ -140,13 +141,12 @@ async def install_bulk(
     for config in _configuration_order(controller, configurations):
         try:
             build_source = controller._resolve_install_source()
-            job = controller._create_job(
-                config,
-                JobType.INSTALL,
-                port=port,
-                build_source=build_source,
+            # A chain (COMPILE + dependent UPLOAD) per device, same as single
+            # install — so device B's compile pipelines against device A's
+            # upload instead of a fused job pinning both to the compile lane.
+            job = await factories.enqueue_install_chain(
+                controller, configuration=config, port=port, build_source=build_source
             )
-            await controller._enqueue(job)
         except CommandError as exc:
             if exc.code is ErrorCode.NO_COMPATIBLE_PEER:
                 raise

esphome_device_builder/controllers/firmware/cli.py

@@ -18,6 +18,7 @@
 from .helpers import _find_esptool_cmd
 
 if TYPE_CHECKING:
+    from ._state import Lane
     from .controller import FirmwareController
 
 _LOGGER = logging.getLogger(__name__)
@@ -100,7 +101,7 @@ def build_cache_args(controller: FirmwareController, job: FirmwareJob) -> list[s
     return controller._db.devices.get_ota_address_cache_args(job.configuration, port)
 
 
-async def verify_chip(controller: FirmwareController, job: FirmwareJob) -> None:
+async def verify_chip(controller: FirmwareController, job: FirmwareJob, lane: Lane) -> None:
     """
     Run ``esptool chip-id`` against *job*'s port and raise on mismatch.
 
@@ -125,6 +126,7 @@ async def verify_chip(controller: FirmwareController, job: FirmwareJob) -> None:
     expected_platform = storage.target_platform.lower()
 
     async with controller._tracked_subprocess(
+        lane,
         *_find_esptool_cmd(),
         "--port",
         job.port,

esphome_device_builder/controllers/firmware/controller.py

@@ -1,8 +1,9 @@
 """
 Firmware build queue + WS command surface.
 
-Owns the persistent single-job queue and the lifecycle event
-broadcasts; the bulk of each concern lives in sibling submodules
+Owns the persistent two-lane queue (a compile lane + an upload lane that
+run concurrently) and the lifecycle event broadcasts; the bulk of each
+concern lives in sibling submodules
 (``runner`` / ``factories`` / ``jobs`` / ``follow`` / ``clean`` /
 ``download`` / ``bulk`` / ``cli`` / ``persistence`` / ``lifecycle``).
 Public API is the ``@api_command``-decorated methods; everything
@@ -19,6 +20,7 @@
 from typing import TYPE_CHECKING, Any
 
 from ...helpers.api import CommandError, api_command
+from ...helpers.async_ import create_eager_task
 from ...models import (
     LOCAL_JOB_BUILD_SOURCE,
     ErrorCode,
@@ -31,7 +33,7 @@
 from . import bulk, cli, factories, follow, jobs, lifecycle, persistence, runner
 from . import clean as clean_mod
 from . import download as download_mod
-from ._state import FirmwareState
+from ._state import FirmwareState, Lane
 from .helpers import (
     _find_esphome_cmd,
     _validate_port,
@@ -47,11 +49,13 @@
 
 class FirmwareController:  # noqa: PLR0904 (grandfathered; new public methods need a refactor first)
     """
-    Manage firmware build jobs with a persistent queue.
+    Manage firmware build jobs with a persistent two-lane queue.
 
-    Only one job runs at a time. Jobs are persisted to disk so they
-    survive page refreshes and server restarts. Progress is broadcast
-    via the event bus to all connected clients.
+    A compile lane (CPU) and an upload lane (network) each run one job at a
+    time but run concurrently, so a slow upload doesn't block the next
+    compile. Jobs are persisted to disk so they survive page refreshes and
+    server restarts. Progress is broadcast via the event bus to all
+    connected clients.
     """
 
     def __init__(self, device_builder: DeviceBuilder) -> None:
@@ -74,20 +78,37 @@ def bus(self) -> EventBus:
     # Lifecycle
     # ------------------------------------------------------------------
 
-    def queue_status_snapshot(self) -> QueueStatus:
-        """Return a :class:`QueueStatus` snapshot of the firmware queue; sync, no I/O.
+    def lane_status(self, lane: Lane) -> QueueStatus:
+        """Return a :class:`QueueStatus` snapshot of one lane; sync, no I/O.
 
-        ``idle`` and ``running`` aren't redundant with each other:
-        ``running=False, queue_depth>0`` is the window between
-        ``_queue.put`` and the runner's ``_queue.get``, so a
-        scheduler reading only ``running`` would misclassify a
-        fully-loaded receiver as accepting more work.
+        ``idle`` and ``running`` aren't redundant: ``running=False,
+        queue_depth>0`` is the window between ``queue.put`` and the
+        runner's ``queue.get``, so a scheduler reading only ``running``
+        would misclassify a loaded lane as accepting more work.
         """
-        running = self.state.current_job is not None
-        queue_depth = self.state.queue.qsize()
+        running = lane.current_job is not None
+        queue_depth = lane.queue.qsize()
         idle = not running and queue_depth == 0
         return QueueStatus(idle=idle, running=running, queue_depth=queue_depth)
 
+    def compile_queue_status(self) -> QueueStatus:
+        """Compile-lane status — what a remote offloader keys on (a receiver only compiles).
+
+        An uploading receiver still has a free compile lane, so it must keep
+        advertising idle for delegated compiles rather than the aggregate.
+        """
+        return self.lane_status(self.state.compile_lane)
+
+    def queue_status_snapshot(self) -> QueueStatus:
+        """Aggregate snapshot across both lanes; sync, no I/O. Idle only when both are."""
+        compile_status = self.lane_status(self.state.compile_lane)
+        upload_status = self.lane_status(self.state.upload_lane)
+        return QueueStatus(
+            idle=compile_status.idle and upload_status.idle,
+            running=compile_status.running or upload_status.running,
+            queue_depth=compile_status.queue_depth + upload_status.queue_depth,
+        )
+
     async def start(self) -> None:
         """Start the queue processor and restore persisted jobs."""
         self.state.esphome_cmd = _find_esphome_cmd()
@@ -193,13 +214,14 @@ async def install(
         _validate_port(port)
         await self._validate_configuration_boundary(configuration)
         build_source = self._resolve_install_source(force_local=force_local)
-        job = self._create_job(
-            configuration,
-            JobType.INSTALL,
-            port=port,
-            build_source=build_source,
+        # Install is a compile + a dependent local upload. The compile (local
+        # or dispatched to a receiver) materialises the binary locally; the
+        # upload then flashes on the upload lane, freeing the compile lane to
+        # build the next device — so a slow flash never blocks a compile, and
+        # a remote receiver keeps compiling while we upload locally.
+        return await factories.enqueue_install_chain(
+            self, configuration=configuration, port=port, build_source=build_source
         )
-        return await self._enqueue(job)
 
     @api_command("firmware/rename")
     async def rename(self, *, configuration: str, new_name: str, **kwargs: Any) -> FirmwareJob:
@@ -343,18 +365,33 @@ async def download_token(self, *, configuration: str, file: str, **kwargs: Any)
     # ------------------------------------------------------------------
 
     async def _run_queue(self) -> None:
-        await runner.run_queue(self)
+        """Run both lane consumers concurrently; drain both on any exit.
 
-    async def _execute_job(self, job: FirmwareJob) -> None:
-        await runner.execute_job(self, job)
+        On shutdown-cancel or one lane raising, cancel and await both lane
+        tasks before the error propagates — else the sibling is left orphaned
+        mid-flight (subprocess not terminated, job not finalised).
+        """
+        lane_tasks = [
+            create_eager_task(runner.run_lane(self, self.state.compile_lane)),
+            create_eager_task(runner.run_lane(self, self.state.upload_lane)),
+        ]
+        try:
+            await asyncio.gather(*lane_tasks)
+        finally:
+            for task in lane_tasks:
+                task.cancel()
+            await asyncio.gather(*lane_tasks, return_exceptions=True)
+
+    async def _execute_job(self, job: FirmwareJob, lane: Lane) -> None:
+        await runner.execute_job(self, job, lane)
 
     async def _execute_remote_job(self, job: FirmwareJob) -> None:
         await runner.execute_remote_job(self, job)
 
     def _tracked_subprocess(
-        self, *args: Any, **kwargs: Any
+        self, lane: Lane, *args: Any, **kwargs: Any
     ) -> AbstractAsyncContextManager[asyncio.subprocess.Process]:
-        return runner.tracked_subprocess(self, *args, **kwargs)
+        return runner.tracked_subprocess(self, lane, *args, **kwargs)
 
     def _finalize_terminal(self, job: FirmwareJob, status: JobStatus) -> None:
         lifecycle.finalize_terminal(self, job, status)
@@ -365,11 +402,11 @@ def _finalize_cancelled(self, job: FirmwareJob) -> None:
     def _raise_if_cancelled(self, job: FirmwareJob, phase: str) -> None:
         lifecycle.raise_if_cancelled(self, job, phase)
 
-    async def _terminate_current_process(self) -> None:
-        await lifecycle.terminate_current_process(self)
+    async def _terminate_current_process(self, lane: Lane) -> None:
+        await lifecycle.terminate_current_process(self, lane)
 
-    async def _verify_chip(self, job: FirmwareJob) -> None:
-        await cli.verify_chip(self, job)
+    async def _verify_chip(self, job: FirmwareJob, lane: Lane) -> None:
+        await cli.verify_chip(self, job, lane)
 
     def _compose_subprocess_env(self, job: FirmwareJob) -> dict[str, str]:
         return cli.compose_subprocess_env(job)
@@ -465,8 +502,10 @@ async def _enqueue(self, job: FirmwareJob, *, supersede: bool = True) -> Firmwar
     def _check_rename_lock(self, job: FirmwareJob) -> None:
         factories.check_rename_lock(self, job)
 
-    async def _supersede_active_jobs(self, configuration: str, *, exclude_job_id: str) -> None:
-        await factories.supersede_active_jobs(self, configuration, exclude_job_id=exclude_job_id)
+    async def _supersede_active_jobs(
+        self, configuration: str, *, exclude_job_ids: set[str]
+    ) -> None:
+        await factories.supersede_active_jobs(self, configuration, exclude_job_ids=exclude_job_ids)
 
     def _prune_history(self) -> None:
         persistence.prune_history(self)