diff --git a/config/runtime.exs b/config/runtime.exs
index 2fab9561..900df5b7 100644
--- a/config/runtime.exs
+++ b/config/runtime.exs
@@ -167,19 +167,29 @@ if config_env() == :prod do
       terrain: 3,
       iemre: 3,
       era5: 2,
-      # No global_limit: the Pro tracker leaked phantom slots across deploys
-      # and stalled the whole backfill. The rate_limit protects CDS and the
-      # per-pod local_limit scales naturally with replicas.
+      # Split-worker ERA5 pipeline:
       #
-      # Parallelism sizing: each job makes 2 CDS requests *serially* (single-
-      # level then pressure-level), so each active worker holds 1 CDS slot at
-      # a time. CDS-Beta's per-user ceiling is ~16 concurrent. local_limit 4
-      # × 3 pods = 12 concurrent, leaving headroom. rate_limit 30/hour stays
-      # above steady-state throughput so it doesn't re-become the bottleneck.
-      era5_batch: [
+      # `era5_submit` runs Era5SubmitWorker, which POSTs two CDS requests
+      # and writes an era5_cds_jobs row. Each job is ~1s of real work, so
+      # local_limit is the gate on how fast we burn CDS submission quota.
+      # rate_limit protects CDS from submit bursts; the per-pod ceiling
+      # (local_limit × replicas) determines peak concurrent in-flight.
+      era5_submit: [
         local_limit: 4,
         rate_limit: [allowed: 30, period: {1, :hour}]
       ],
+      # `era5_poll` runs Era5PollWorker, which does a tiny GET per run and
+      # either snoozes (releasing the slot) or streams the completed GRIB
+      # and inserts. Needs high local_limit because in-flight CDS jobs
+      # scale independently of concurrent poll work — a pod may have 50
+      # tile-months in flight but be doing ~1 actual poll GET at a time.
+      era5_poll: [
+        local_limit: 20
+      ],
+      # Legacy queue name retained briefly so any in-flight Era5MonthBatchWorker
+      # jobs from the pre-split deploy drain cleanly (the worker now just
+      # forwards to :era5_submit). Safe to delete after one rolling deploy.
+      era5_batch: 1,
       rtma: 2,
       backfill_enqueue: 1,
       admin: 1,
diff --git a/config/test.exs b/config/test.exs
index c31606bf..3ad77019 100644
--- a/config/test.exs
+++ b/config/test.exs
@@ -50,6 +50,7 @@ config :microwaveprop, Oban, testing: :inline
 config :microwaveprop, cache_contact_count: false
 config :microwaveprop, cache_contact_map: false
 config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Terrain.ElevationClient}, retry: false]
+config :microwaveprop, era5_req_options: [plug: {Req.Test, Microwaveprop.Weather.Era5Client}, retry: false]
 config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false]
 config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false]
 
diff --git a/lib/microwaveprop/weather/era5_batch_client.ex b/lib/microwaveprop/weather/era5_batch_client.ex
index 52c4dcf5..e1f6853a 100644
--- a/lib/microwaveprop/weather/era5_batch_client.ex
+++ b/lib/microwaveprop/weather/era5_batch_client.ex
@@ -101,13 +101,8 @@ defmodule Microwaveprop.Weather.Era5BatchClient do
 
     try do
       with :ok <- Task.await(single_task, :infinity),
-           :ok <- Task.await(pressure_task, :infinity),
-           {:ok, profiles} <- build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
-        inserted = bulk_insert_profiles(profiles)
-
-        Logger.info("ERA5Batch: stored #{inserted} profiles for #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}")
-
-        {:ok, inserted}
+           :ok <- Task.await(pressure_task, :infinity) do
+        decode_and_insert(year, month, tile_lat, tile_lon, single_path, pressure_path)
       end
     after
       Task.shutdown(single_task, :brutal_kill)
@@ -117,6 +112,31 @@ defmodule Microwaveprop.Weather.Era5BatchClient do
     end
   end
 
+  @doc """
+  Decodes a downloaded single-level + pressure-level GRIB2 pair from disk
+  and bulk-inserts the resulting month-tile profiles into `era5_profiles`.
+
+  Shared between the legacy inline `do_fetch_month/4` path and the split
+  `Era5PollWorker` path.
+  """
+  @spec decode_and_insert(
+          pos_integer(),
+          pos_integer(),
+          integer(),
+          integer(),
+          Path.t(),
+          Path.t()
+        ) :: {:ok, non_neg_integer()} | {:error, term()}
+  def decode_and_insert(year, month, tile_lat, tile_lon, single_path, pressure_path) do
+    with {:ok, profiles} <- build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
+      inserted = bulk_insert_profiles(profiles)
+
+      Logger.info("ERA5Batch: stored #{inserted} profiles for #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}")
+
+      {:ok, inserted}
+    end
+  end
+
   defp month_tile_cached?(year, month, tile_lat, tile_lon) do
     {first, last} = month_bounds(year, month)
 
diff --git a/lib/microwaveprop/weather/era5_cds_job.ex b/lib/microwaveprop/weather/era5_cds_job.ex
new file mode 100644
index 00000000..311204d0
--- /dev/null
+++ b/lib/microwaveprop/weather/era5_cds_job.ex
@@ -0,0 +1,44 @@
+defmodule Microwaveprop.Weather.Era5CdsJob do
+  @moduledoc """
+  Tracks an in-flight ERA5 CDS request submitted by `Era5SubmitWorker` and
+  awaiting download+decode by `Era5PollWorker`. Persisting the CDS job IDs
+  decouples the Oban worker lifecycle from the 30+ minute CDS queue time,
+  so a pod restart resumes polling instead of re-submitting.
+  """
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "era5_cds_jobs" do
+    field :year, :integer
+    field :month, :integer
+    field :tile_lat, :integer
+    field :tile_lon, :integer
+    field :single_job_id, :string
+    field :pressure_job_id, :string
+    field :submitted_at, :utc_datetime
+    field :last_polled_at, :utc_datetime
+    field :poll_count, :integer, default: 0
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @type t :: %__MODULE__{}
+
+  @required_fields ~w(year month tile_lat tile_lon single_job_id pressure_job_id submitted_at)a
+  @optional_fields ~w(last_polled_at poll_count)a
+
+  @spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
+  def changeset(job, attrs) do
+    job
+    |> cast(attrs, @required_fields ++ @optional_fields)
+    |> validate_required(@required_fields)
+    |> unique_constraint([:year, :month, :tile_lat, :tile_lon],
+      name: :era5_cds_jobs_year_month_tile_lat_tile_lon_index,
+      error_key: :tile_month
+    )
+  end
+end
diff --git a/lib/microwaveprop/weather/era5_client.ex b/lib/microwaveprop/weather/era5_client.ex
index 25017fec..dbb3a5b7 100644
--- a/lib/microwaveprop/weather/era5_client.ex
+++ b/lib/microwaveprop/weather/era5_client.ex
@@ -135,13 +135,94 @@ defmodule Microwaveprop.Weather.Era5Client do
   @spec submit_and_download_to_file(String.t(), map(), Path.t()) :: :ok | {:error, term()}
   def submit_and_download_to_file(dataset, request, path) do
     with {:ok, source, api_key} <- submit_and_poll(dataset, request) do
-      case source do
-        {:body, body} -> File.write(path, body)
-        {:url, href} -> download_href_to_file(href, api_key, path)
+      download_source_to_file(source, api_key, path)
+    end
+  end
+
+  @doc """
+  Submits a CDS job and returns `{:ok, job_id}` without waiting for it to
+  complete. Used by the split `Era5SubmitWorker` / `Era5PollWorker` flow so
+  Oban worker slots aren't pinned to the 30+ minute CDS queue time.
+  """
+  @spec submit_job(String.t(), map()) :: {:ok, String.t()} | {:error, String.t()}
+  def submit_job(dataset, request) do
+    api_key = api_key()
+
+    if is_nil(api_key) do
+      {:error, "ERA5_CDS_API_KEY not configured"}
+    else
+      submit_request(dataset, request, api_key)
+    end
+  end
+
+  @doc """
+  Checks the status of a previously-submitted CDS job and, if complete,
+  follows the results pointer to return the download source.
+
+  Returns:
+  - `:running` — job is queued or in progress; caller should retry later.
+  - `{:done, {:url, href} | {:body, binary}}` — ready; pass the source to
+    `download_source_to_file/3`.
+  - `{:failed, reason}` — CDS marked the job failed; don't retry.
+  - `{:error, reason}` — transient HTTP/network error; caller should retry.
+  """
+  @spec check_status(String.t()) ::
+          :running
+          | {:done, {:url, String.t()} | {:body, binary()}}
+          | {:failed, String.t()}
+          | {:error, String.t()}
+  def check_status(job_id) do
+    api_key = api_key()
+
+    if is_nil(api_key) do
+      {:error, "ERA5_CDS_API_KEY not configured"}
+    else
+      do_check_status(job_id, api_key)
+    end
+  end
+
+  @doc """
+  Deletes a completed CDS job from the server so it doesn't count against
+  our per-user job quota. Called by `Era5PollWorker` after a successful
+  download. Treats 200/202/204/404 as success — 404 is fine because the
+  job may already have been reaped by CDS's own cleanup.
+  """
+  @spec delete_job(String.t()) :: :ok | {:error, String.t()}
+  def delete_job(job_id) do
+    api_key = api_key()
+
+    if is_nil(api_key) do
+      {:error, "ERA5_CDS_API_KEY not configured"}
+    else
+      url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
+
+      case Req.delete(
+             url,
+             [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 30_000] ++ req_options()
+           ) do
+        {:ok, %{status: status}} when status in [200, 202, 204, 404] -> :ok
+        {:ok, %{status: status}} -> {:error, "ERA5 delete_job HTTP #{status}"}
+        {:error, reason} -> {:error, "ERA5 delete_job failed: #{inspect(reason)}"}
       end
     end
   end
 
+  @doc """
+  Writes a CDS result source to `path`. `{:body, binary}` is written as a
+  plain file; `{:url, href}` is streamed through Req's `into: File.stream!`
+  so peak heap stays O(chunk) rather than O(full response).
+  """
+  @spec download_source_to_file(
+          {:body, binary()} | {:url, String.t()},
+          String.t() | nil,
+          Path.t()
+        ) :: :ok | {:error, String.t()}
+  def download_source_to_file({:body, body}, _api_key, path), do: File.write(path, body)
+
+  def download_source_to_file({:url, href}, api_key, path) do
+    download_href_to_file(href, api_key, path)
+  end
+
   defp do_submit_and_download(dataset, request) do
     with {:ok, source, api_key} <- submit_and_poll(dataset, request) do
       case source do
@@ -164,13 +245,57 @@ defmodule Microwaveprop.Weather.Era5Client do
     end
   end
 
+  defp do_check_status(job_id, api_key) do
+    url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
+
+    url
+    |> Req.get([headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000] ++ req_options())
+    |> interpret_status_response(url, api_key)
+  end
+
+  defp interpret_status_response({:ok, %{status: 200, body: body}}, url, api_key) do
+    interpret_status_body(body, url, api_key)
+  end
+
+  defp interpret_status_response({:ok, %{status: status, body: body}}, _url, _api_key) do
+    {:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"}
+  end
+
+  defp interpret_status_response({:error, reason}, _url, _api_key) do
+    {:error, "ERA5 poll failed: #{inspect(reason)}"}
+  end
+
+  defp interpret_status_body(%{"status" => "successful"}, url, api_key) do
+    case fetch_results_source("#{url}/results", api_key) do
+      {:ok, source} -> {:done, source}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp interpret_status_body(%{"status" => status}, _url, _api_key)
+       when status in ["accepted", "running"] do
+    :running
+  end
+
+  defp interpret_status_body(%{"status" => "failed"} = body, _url, _api_key) do
+    reason = body["message"] || body["error"] || inspect(body)
+    {:failed, reason}
+  end
+
+  defp interpret_status_body(body, _url, _api_key) do
+    {:error, "ERA5 poll unknown status body: #{inspect(body)}"}
+  end
+
   defp submit_request(dataset, request, api_key) do
     url = "#{@cds_url}/retrieve/v1/processes/#{dataset}/execution"
 
-    case Req.post(url,
-           json: %{"inputs" => request},
-           headers: [{"PRIVATE-TOKEN", api_key}],
-           receive_timeout: 30_000
+    case Req.post(
+           url,
+           [
+             json: %{"inputs" => request},
+             headers: [{"PRIVATE-TOKEN", api_key}],
+             receive_timeout: 30_000
+           ] ++ req_options()
          ) do
       {:ok, %{status: status, body: body}} when status in [200, 201, 202] ->
         job_id = body["jobID"] || body["request_id"] || body["jobId"]
@@ -201,7 +326,10 @@ defmodule Microwaveprop.Weather.Era5Client do
   defp poll_until_complete(job_id, api_key, attempt) do
     url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
 
-    case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000) do
+    case Req.get(
+           url,
+           [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000] ++ req_options()
+         ) do
       {:ok, %{status: 200, body: %{"status" => "successful"}}} ->
         results_url = "#{url}/results"
         fetch_results_source(results_url, api_key)
@@ -226,7 +354,10 @@ defmodule Microwaveprop.Weather.Era5Client do
   end
 
   defp fetch_results_source(url, api_key) do
-    case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000) do
+    case Req.get(
+           url,
+           [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000] ++ req_options()
+         ) do
       {:ok, %{status: 200, body: %{"asset" => %{"value" => %{"href" => href}}}}} ->
         # New CDS returns a JSON with download link
         {:ok, {:url, href}}
@@ -253,7 +384,10 @@ defmodule Microwaveprop.Weather.Era5Client do
   end
 
   defp download_file(href, api_key) do
-    case Req.get(href, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000) do
+    case Req.get(
+           href,
+           [headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000] ++ req_options()
+         ) do
       {:ok, %{status: 200, body: body}} -> {:ok, body}
       {:ok, %{status: status}} -> {:error, "ERA5 file download HTTP #{status}"}
       {:error, reason} -> {:error, "ERA5 file download failed: #{inspect(reason)}"}
@@ -266,10 +400,13 @@ defmodule Microwaveprop.Weather.Era5Client do
   # removed on HTTP error so a failed download never leaves a partial tmp file.
   defp download_href_to_file(href, api_key, path) do
     result =
-      Req.get(href,
-        headers: [{"PRIVATE-TOKEN", api_key}],
-        receive_timeout: 300_000,
-        into: File.stream!(path)
+      Req.get(
+        href,
+        [
+          headers: [{"PRIVATE-TOKEN", api_key}],
+          receive_timeout: 300_000,
+          into: File.stream!(path)
+        ] ++ req_options()
       )
 
     case result do
@@ -286,6 +423,10 @@ defmodule Microwaveprop.Weather.Era5Client do
     end
   end
 
+  defp req_options do
+    Application.get_env(:microwaveprop, :era5_req_options, [])
+  end
+
   defp build_profile(lat, lon, valid_time, single_grib, pressure_grib) do
     # Decode GRIB2 data using existing infrastructure
     with {:ok, single_points} <- extract_point(single_grib, lat, lon),
diff --git a/lib/microwaveprop/workers/era5_month_batch_worker.ex b/lib/microwaveprop/workers/era5_month_batch_worker.ex
index 1e5f1454..c9a89ecb 100644
--- a/lib/microwaveprop/workers/era5_month_batch_worker.ex
+++ b/lib/microwaveprop/workers/era5_month_batch_worker.ex
@@ -1,29 +1,15 @@
 defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
   @moduledoc """
-  Fetches one month of ERA5 data for a 2° × 2° tile in a single CDS request
-  and bulk-inserts all resulting hourly profiles into `era5_profiles`.
+  Legacy entry point for the ERA5 month-tile backfill. **Forwards** to
+  `Era5SubmitWorker` in the new split-worker pipeline — kept in place so
+  already-enqueued callers (`Era5FetchWorker`, `mix era5.backfill`, any
+  in-flight Oban rows) don't need to be changed on the same deploy.
 
-  This exists because fetching ERA5 per point-hour is tragically slow: each
-  CDS request goes through a full submit → poll → assemble → download cycle
-  whose overhead dwarfs the data transfer. Grouping by month+tile turns
-  thousands of CDS jobs into a handful, and once a tile-month is cached in
-  the DB every downstream query for a point inside that window is a local
-  lookup.
-
-  Uniqueness on the full arg set means Oban collapses duplicate enqueues
-  for the same tile-month — multiple `Era5FetchWorker` requests for the
-  same region collapse into a single batch.
-
-  Runs on its own `:era5_batch` queue so its slow CDS poll cycles don't
-  starve the fast `Era5FetchWorker` router that also lives in the ERA5
-  namespace.
+  On the next deploy cycle, callers will be updated to enqueue
+  `Era5SubmitWorker` directly and this module can be removed.
   """
-  # The :era5_batch queue is rate-limited (see config/runtime.exs) to stay
-  # under CDS-Beta's per-user concurrent ceiling. Retrying 5×, combined with
-  # the generous backoff below, gives CDS up to a day to come back without
-  # discarding the tile and silently dropping the historical contact.
   use Oban.Worker,
-    queue: :era5_batch,
+    queue: :era5_submit,
     max_attempts: 5,
     unique: [
       period: :infinity,
@@ -31,42 +17,26 @@ defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
       keys: [:year, :month, :tile_lat, :tile_lon]
     ]
 
-  alias Microwaveprop.Weather.Era5BatchClient
+  alias Microwaveprop.Workers.Era5SubmitWorker
 
   require Logger
 
   @impl Oban.Worker
   def backoff(%Oban.Job{attempt: attempt}) do
-    # A full month fetch can sit in the CDS queue for ages; back off generously.
-    min(600 * Integer.pow(2, attempt - 1), _one_day = 86_400)
+    # Forwarding is cheap; retry fast on transient failure.
+    min(30 * Integer.pow(2, attempt - 1), 3_600)
   end
 
   @impl Oban.Worker
   def perform(%Oban.Job{args: args}) do
-    params = %{
-      year: fetch_int!(args, "year"),
-      month: fetch_int!(args, "month"),
-      tile_lat: fetch_int!(args, "tile_lat"),
-      tile_lon: fetch_int!(args, "tile_lon")
-    }
+    Logger.debug("Era5MonthBatchWorker forwarding to Era5SubmitWorker: #{inspect(args)}")
 
-    case Era5BatchClient.fetch_month_into_db(params) do
-      {:ok, _count} ->
-        :ok
-
-      {:error, reason} ->
-        Logger.warning(
-          "ERA5Batch: #{params.year}-#{params.month} tile #{params.tile_lat},#{params.tile_lon} failed: #{inspect(reason)}"
-        )
-
-        {:error, reason}
-    end
-  end
-
-  defp fetch_int!(args, key) do
-    case Map.fetch!(args, key) do
-      n when is_integer(n) -> n
-      n when is_binary(n) -> String.to_integer(n)
+    args
+    |> Era5SubmitWorker.new()
+    |> Oban.insert()
+    |> case do
+      {:ok, _job} -> :ok
+      {:error, reason} -> {:error, reason}
     end
   end
 end
diff --git a/lib/microwaveprop/workers/era5_poll_worker.ex b/lib/microwaveprop/workers/era5_poll_worker.ex
new file mode 100644
index 00000000..c117f8a9
--- /dev/null
+++ b/lib/microwaveprop/workers/era5_poll_worker.ex
@@ -0,0 +1,152 @@
+defmodule Microwaveprop.Workers.Era5PollWorker do
+  @moduledoc """
+  Second half of the split ERA5 backfill pipeline. Polls CDS for the two
+  jobs submitted by `Era5SubmitWorker`; on completion, streams both GRIB2
+  files to disk, decodes them, bulk-inserts the profiles, deletes the
+  `era5_cds_jobs` row, and asks CDS to delete the completed jobs so they
+  don't count against our per-user quota.
+
+  While either CDS job is still running the worker returns
+  `{:snooze, N}` — Oban re-schedules the same job without counting an
+  attempt and releases the worker slot immediately, so one pod can keep
+  dozens of tile-months in flight without pinning workers to
+  `Process.sleep`.
+
+  `max_attempts` is high because each attempt models a human-scale
+  failure (e.g. CDS reports the job failed), not a poll retry — snoozes
+  don't consume attempts.
+  """
+  use Oban.Worker,
+    queue: :era5_poll,
+    max_attempts: 10
+
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Weather.Era5BatchClient
+  alias Microwaveprop.Weather.Era5CdsJob
+  alias Microwaveprop.Weather.Era5Client
+
+  require Logger
+
+  # Re-check every 5 minutes. CDS month-tile jobs typically take 30-45 min,
+  # so 5 minutes strikes a balance between responsiveness and noise.
+  @snooze_seconds 5 * 60
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"era5_cds_job_id" => id}}) do
+    case Repo.get(Era5CdsJob, id) do
+      nil ->
+        Logger.debug("Era5Poll: row #{id} no longer exists — no-op")
+        :ok
+
+      %Era5CdsJob{} = row ->
+        handle_row(row)
+    end
+  end
+
+  defp handle_row(row) do
+    single = Era5Client.check_status(row.single_job_id)
+    pressure = Era5Client.check_status(row.pressure_job_id)
+
+    case {single, pressure} do
+      {{:done, single_source}, {:done, pressure_source}} ->
+        handle_both_done(row, single_source, pressure_source)
+
+      {{:failed, reason}, _} ->
+        fail_row(row, "single-level job failed: #{reason}")
+
+      {_, {:failed, reason}} ->
+        fail_row(row, "pressure-level job failed: #{reason}")
+
+      {{:error, reason}, _} ->
+        {:error, "single-level check_status error: #{reason}"}
+
+      {_, {:error, reason}} ->
+        {:error, "pressure-level check_status error: #{reason}"}
+
+      _both_running_or_mixed ->
+        bump_poll_count(row)
+        {:snooze, @snooze_seconds}
+    end
+  end
+
+  defp handle_both_done(row, single_source, pressure_source) do
+    tmp_dir = System.tmp_dir!()
+
+    tag =
+      "#{row.year}_#{pad(row.month)}_#{row.tile_lat}_#{row.tile_lon}_#{System.unique_integer([:positive])}"
+
+    single_path = Path.join(tmp_dir, "era5_single_#{tag}.grib2")
+    pressure_path = Path.join(tmp_dir, "era5_pressure_#{tag}.grib2")
+
+    try do
+      with :ok <-
+             Era5Client.download_source_to_file(single_source, api_key(), single_path),
+           :ok <-
+             Era5Client.download_source_to_file(pressure_source, api_key(), pressure_path),
+           {:ok, inserted} <-
+             Era5BatchClient.decode_and_insert(
+               row.year,
+               row.month,
+               row.tile_lat,
+               row.tile_lon,
+               single_path,
+               pressure_path
+             ) do
+        Repo.delete!(row)
+        delete_cds_jobs(row)
+
+        Logger.info(
+          "Era5Poll: stored #{inserted} profiles for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}"
+        )
+
+        {:ok, inserted}
+      else
+        {:error, reason} ->
+          Logger.warning(
+            "Era5Poll: download/decode failed for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}: #{inspect(reason)}"
+          )
+
+          {:error, reason}
+      end
+    after
+      File.rm(single_path)
+      File.rm(pressure_path)
+    end
+  end
+
+  defp fail_row(row, reason) do
+    Logger.warning(
+      "Era5Poll: CDS failure for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}: #{reason}"
+    )
+
+    delete_cds_jobs(row)
+    Repo.delete!(row)
+    {:error, reason}
+  end
+
+  defp delete_cds_jobs(row) do
+    # Best-effort — don't block progress on a CDS cleanup failure.
+    for job_id <- [row.single_job_id, row.pressure_job_id] do
+      case Era5Client.delete_job(job_id) do
+        :ok ->
+          :ok
+
+        {:error, reason} ->
+          Logger.warning("Era5Poll: failed to delete CDS job #{job_id}: #{reason}")
+          :ok
+      end
+    end
+  end
+
+  defp bump_poll_count(row) do
+    now = DateTime.truncate(DateTime.utc_now(), :second)
+
+    row
+    |> Era5CdsJob.changeset(%{poll_count: row.poll_count + 1, last_polled_at: now})
+    |> Repo.update!()
+  end
+
+  defp api_key, do: System.get_env("ERA5_CDS_API_KEY")
+
+  defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
+end
diff --git a/lib/microwaveprop/workers/era5_submit_worker.ex b/lib/microwaveprop/workers/era5_submit_worker.ex
new file mode 100644
index 00000000..3748d1cc
--- /dev/null
+++ b/lib/microwaveprop/workers/era5_submit_worker.ex
@@ -0,0 +1,207 @@
+defmodule Microwaveprop.Workers.Era5SubmitWorker do
+  @moduledoc """
+  First half of the split ERA5 backfill pipeline.
+
+  Submits the two CDS jobs for a month-tile (single-level + pressure-level)
+  in parallel, persists the returned CDS job IDs to `era5_cds_jobs`, and
+  enqueues an `Era5PollWorker` job scheduled a few minutes out to check for
+  completion.
+
+  The fast submit → enqueue-poll split means an Oban slot is held for ~1
+  second (the time to POST and write a row), not the 30-45 minutes CDS
+  takes to queue, assemble, and return a month-tile. A pod restart between
+  submit and poll does not lose the CDS work: the row in `era5_cds_jobs`
+  persists the job IDs and the poll worker picks up where we left off.
+
+  Idempotent: if the month-tile already has profiles (`era5_profiles`) or
+  an in-flight `era5_cds_jobs` row, no submit is made.
+  """
+  use Oban.Worker,
+    queue: :era5_submit,
+    max_attempts: 5,
+    unique: [
+      period: :infinity,
+      states: [:available, :scheduled, :executing, :retryable],
+      keys: [:year, :month, :tile_lat, :tile_lon]
+    ]
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Weather.Era5BatchClient
+  alias Microwaveprop.Weather.Era5CdsJob
+  alias Microwaveprop.Weather.Era5Client
+  alias Microwaveprop.Weather.Era5Profile
+  alias Microwaveprop.Workers.Era5PollWorker
+
+  require Logger
+
+  @poll_delay_seconds 5 * 60
+
+  @impl Oban.Worker
+  def backoff(%Oban.Job{attempt: attempt}) do
+    # The submit itself is cheap; retry fast so transient CDS/HTTP blips
+    # don't stall the tile for an hour waiting on exponential backoff.
+    min(30 * Integer.pow(2, attempt - 1), _one_hour = 3_600)
+  end
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: args}) do
+    year = fetch_int!(args, "year")
+    month = fetch_int!(args, "month")
+    tile_lat = fetch_int!(args, "tile_lat")
+    tile_lon = fetch_int!(args, "tile_lon")
+
+    cond do
+      month_tile_cached?(year, month, tile_lat, tile_lon) ->
+        Logger.info("Era5Submit: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already cached — skip")
+
+        {:ok, :cached}
+
+      existing = find_in_flight(year, month, tile_lat, tile_lon) ->
+        Logger.info(
+          "Era5Submit: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already submitted (cds_job_id=#{existing.id}) — skip"
+        )
+
+        {:ok, :already_submitted}
+
+      true ->
+        submit_and_persist(year, month, tile_lat, tile_lon)
+    end
+  end
+
+  defp submit_and_persist(year, month, tile_lat, tile_lon) do
+    days = Calendar.ISO.days_in_month(year, month)
+    area = tile_area(tile_lat, tile_lon)
+
+    # Run both submits in parallel — they're independent HTTP POSTs.
+    single_task =
+      Task.async(fn ->
+        Era5Client.submit_job("reanalysis-era5-single-levels", single_request(year, month, days, area))
+      end)
+
+    pressure_task =
+      Task.async(fn ->
+        Era5Client.submit_job("reanalysis-era5-pressure-levels", pressure_request(year, month, days, area))
+      end)
+
+    try do
+      with {:ok, single_id} <- Task.await(single_task, :infinity),
+           {:ok, pressure_id} <- Task.await(pressure_task, :infinity),
+           {:ok, row} <-
+             insert_cds_job_row(year, month, tile_lat, tile_lon, single_id, pressure_id) do
+        enqueue_poll(row)
+
+        Logger.info(
+          "Era5Submit: submitted #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} (single=#{single_id}, pressure=#{pressure_id})"
+        )
+
+        {:ok, row}
+      end
+    after
+      Task.shutdown(single_task, :brutal_kill)
+      Task.shutdown(pressure_task, :brutal_kill)
+    end
+  end
+
+  defp insert_cds_job_row(year, month, tile_lat, tile_lon, single_id, pressure_id) do
+    %Era5CdsJob{}
+    |> Era5CdsJob.changeset(%{
+      year: year,
+      month: month,
+      tile_lat: tile_lat,
+      tile_lon: tile_lon,
+      single_job_id: single_id,
+      pressure_job_id: pressure_id,
+      submitted_at: DateTime.truncate(DateTime.utc_now(), :second)
+    })
+    |> Repo.insert()
+    |> case do
+      {:ok, row} -> {:ok, row}
+      {:error, changeset} -> {:error, "era5_cds_jobs insert failed: #{inspect(changeset.errors)}"}
+    end
+  end
+
+  defp enqueue_poll(%Era5CdsJob{id: id}) do
+    %{"era5_cds_job_id" => id}
+    |> Era5PollWorker.new(schedule_in: @poll_delay_seconds)
+    |> Oban.insert()
+  end
+
+  defp find_in_flight(year, month, tile_lat, tile_lon) do
+    Repo.one(
+      from j in Era5CdsJob,
+        where:
+          j.year == ^year and j.month == ^month and j.tile_lat == ^tile_lat and
+            j.tile_lon == ^tile_lon,
+        limit: 1
+    )
+  end
+
+  defp month_tile_cached?(year, month, tile_lat, tile_lon) do
+    {first_dt, last_dt} = month_bounds(year, month)
+    tile_degrees = Era5BatchClient.tile_degrees()
+
+    Era5Profile
+    |> where(
+      [p],
+      p.valid_time >= ^first_dt and p.valid_time < ^last_dt and
+        p.lat >= ^(tile_lat * 1.0) and p.lat <= ^(tile_lat * 1.0 + tile_degrees) and
+        p.lon >= ^(tile_lon * 1.0) and p.lon <= ^(tile_lon * 1.0 + tile_degrees)
+    )
+    |> Repo.exists?()
+  end
+
+  defp month_bounds(year, month) do
+    {:ok, first} = Date.new(year, month, 1)
+    {:ok, first_dt} = DateTime.new(first, ~T[00:00:00], "Etc/UTC")
+    days = Calendar.ISO.days_in_month(year, month)
+    last_dt = DateTime.add(first_dt, days * 86_400, :second)
+    {first_dt, last_dt}
+  end
+
+  defp tile_area(tile_lat, tile_lon) do
+    tile_degrees = Era5BatchClient.tile_degrees()
+
+    Enum.map(
+      [tile_lat + tile_degrees, tile_lon, tile_lat, tile_lon + tile_degrees],
+      &(&1 * 1.0)
+    )
+  end
+
+  defp single_request(year, month, days, area) do
+    %{
+      "product_type" => ["reanalysis"],
+      "variable" => Era5Client.single_level_vars(),
+      "year" => [Integer.to_string(year)],
+      "month" => [pad(month)],
+      "day" => Enum.map(1..days, &pad/1),
+      "time" => Enum.map(0..23, &"#{pad(&1)}:00"),
+      "area" => area,
+      "data_format" => "grib"
+    }
+  end
+
+  defp pressure_request(year, month, days, area) do
+    %{
+      "product_type" => ["reanalysis"],
+      "variable" => Era5Client.pressure_level_vars(),
+      "pressure_level" => Era5Client.pressure_levels(),
+      "year" => [Integer.to_string(year)],
+      "month" => [pad(month)],
+      "day" => Enum.map(1..days, &pad/1),
+      "time" => Enum.map(0..23, &"#{pad(&1)}:00"),
+      "area" => area,
+      "data_format" => "grib"
+    }
+  end
+
+  defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
+
+  defp fetch_int!(args, key) do
+    case Map.fetch!(args, key) do
+      v when is_integer(v) -> v
+      v when is_binary(v) -> String.to_integer(v)
+    end
+  end
+end
diff --git a/priv/repo/migrations/20260413210544_create_era5_cds_jobs.exs b/priv/repo/migrations/20260413210544_create_era5_cds_jobs.exs
new file mode 100644
index 00000000..51d7042c
--- /dev/null
+++ b/priv/repo/migrations/20260413210544_create_era5_cds_jobs.exs
@@ -0,0 +1,28 @@
+defmodule Microwaveprop.Repo.Migrations.CreateEra5CdsJobs do
+  use Ecto.Migration
+
+  # Tracks in-flight CDS requests for the split Era5SubmitWorker / Era5PollWorker
+  # design. A row is created when Era5SubmitWorker successfully submits both
+  # CDS jobs, and is deleted when Era5PollWorker downloads the results and
+  # inserts profiles (or when either leg permanently fails). Persisting the
+  # CDS job IDs lets a pod restart resume polling instead of losing the work.
+  def change do
+    create table(:era5_cds_jobs, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :year, :integer, null: false
+      add :month, :integer, null: false
+      add :tile_lat, :integer, null: false
+      add :tile_lon, :integer, null: false
+      add :single_job_id, :string, null: false
+      add :pressure_job_id, :string, null: false
+      add :submitted_at, :utc_datetime, null: false
+      add :last_polled_at, :utc_datetime
+      add :poll_count, :integer, default: 0, null: false
+
+      timestamps(type: :utc_datetime)
+    end
+
+    # Uniqueness so duplicate submits for the same tile-month collapse.
+    create unique_index(:era5_cds_jobs, [:year, :month, :tile_lat, :tile_lon])
+  end
+end
diff --git a/test/microwaveprop/weather/era5_cds_job_test.exs b/test/microwaveprop/weather/era5_cds_job_test.exs
new file mode 100644
index 00000000..543ff58d
--- /dev/null
+++ b/test/microwaveprop/weather/era5_cds_job_test.exs
@@ -0,0 +1,82 @@
+defmodule Microwaveprop.Weather.Era5CdsJobTest do
+  use Microwaveprop.DataCase, async: true
+
+  alias Microwaveprop.Weather.Era5CdsJob
+
+  describe "changeset/2" do
+    @valid_attrs %{
+      year: 2014,
+      month: 3,
+      tile_lat: 32,
+      tile_lon: -98,
+      single_job_id: "abc-single",
+      pressure_job_id: "abc-pressure",
+      submitted_at: ~U[2026-04-13 21:00:00Z]
+    }
+
+    test "accepts a complete attrs map" do
+      changeset = Era5CdsJob.changeset(%Era5CdsJob{}, @valid_attrs)
+      assert changeset.valid?
+    end
+
+    test "requires year, month, tile coordinates, and both CDS job IDs" do
+      changeset = Era5CdsJob.changeset(%Era5CdsJob{}, %{})
+
+      refute changeset.valid?
+
+      required = ~w(year month tile_lat tile_lon single_job_id pressure_job_id submitted_at)a
+      errors = Keyword.keys(changeset.errors)
+
+      for field <- required do
+        assert field in errors, "expected #{inspect(field)} to be required"
+      end
+    end
+
+    test "last_polled_at and poll_count default to nil/0 and can be updated" do
+      {:ok, row} =
+        %Era5CdsJob{}
+        |> Era5CdsJob.changeset(@valid_attrs)
+        |> Repo.insert()
+
+      assert row.poll_count == 0
+      assert is_nil(row.last_polled_at)
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+
+      {:ok, bumped} =
+        row
+        |> Era5CdsJob.changeset(%{last_polled_at: now, poll_count: 1})
+        |> Repo.update()
+
+      assert bumped.poll_count == 1
+      assert bumped.last_polled_at == now
+    end
+  end
+
+  describe "uniqueness" do
+    test "collapses duplicate (year, month, tile_lat, tile_lon) inserts" do
+      attrs = %{
+        year: 2014,
+        month: 3,
+        tile_lat: 32,
+        tile_lon: -98,
+        single_job_id: "first-single",
+        pressure_job_id: "first-pressure",
+        submitted_at: ~U[2026-04-13 21:00:00Z]
+      }
+
+      {:ok, _} =
+        %Era5CdsJob{}
+        |> Era5CdsJob.changeset(attrs)
+        |> Repo.insert()
+
+      {:error, changeset} =
+        %Era5CdsJob{}
+        |> Era5CdsJob.changeset(%{attrs | single_job_id: "second-single", pressure_job_id: "second-pressure"})
+        |> Repo.insert()
+
+      refute changeset.valid?
+      assert {:tile_month, _} = List.keyfind(changeset.errors, :tile_month, 0) || {:tile_month, nil}
+    end
+  end
+end
diff --git a/test/microwaveprop/weather/era5_client_test.exs b/test/microwaveprop/weather/era5_client_test.exs
new file mode 100644
index 00000000..d20fff6e
--- /dev/null
+++ b/test/microwaveprop/weather/era5_client_test.exs
@@ -0,0 +1,176 @@
+defmodule Microwaveprop.Weather.Era5ClientTest do
+  # async: false — tests mutate the global ERA5_CDS_API_KEY env var to match
+  # production's lookup path. Running concurrently with Era5BatchClientTest
+  # (which *deletes* the env var to exercise the missing-key error path)
+  # leaks state in both directions.
+  use ExUnit.Case, async: false
+
+  alias Microwaveprop.Weather.Era5Client
+
+  setup do
+    # A real key — stubs don't care about value, only presence.
+    System.put_env("ERA5_CDS_API_KEY", "test-key")
+    on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
+    :ok
+  end
+
+  describe "submit_job/2" do
+    test "returns {:ok, job_id} when CDS accepts the request" do
+      Req.Test.stub(Era5Client, fn conn ->
+        assert conn.method == "POST"
+        assert String.ends_with?(conn.request_path, "/reanalysis-era5-single-levels/execution")
+        Req.Test.json(conn, %{"jobID" => "cds-job-123", "status" => "accepted"})
+      end)
+
+      assert {:ok, "cds-job-123"} =
+               Era5Client.submit_job("reanalysis-era5-single-levels", %{"foo" => "bar"})
+    end
+
+    test "returns {:error, _} when CDS returns 400" do
+      Req.Test.stub(Era5Client, fn conn ->
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.resp(400, ~s({"detail":"bad request"}))
+      end)
+
+      assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
+      assert reason =~ "HTTP 400"
+    end
+
+    test "returns {:error, _} when API key is missing" do
+      System.delete_env("ERA5_CDS_API_KEY")
+      assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
+      assert reason =~ "ERA5_CDS_API_KEY"
+    end
+  end
+
+  describe "check_status/1" do
+    test "returns :running while CDS job is accepted" do
+      Req.Test.stub(Era5Client, fn conn ->
+        assert conn.method == "GET"
+        assert String.ends_with?(conn.request_path, "/jobs/cds-job-123")
+        Req.Test.json(conn, %{"status" => "accepted"})
+      end)
+
+      assert :running = Era5Client.check_status("cds-job-123")
+    end
+
+    test "returns :running while CDS job is running" do
+      Req.Test.stub(Era5Client, fn conn ->
+        Req.Test.json(conn, %{"status" => "running"})
+      end)
+
+      assert :running = Era5Client.check_status("cds-job-123")
+    end
+
+    test "returns {:done, {:url, href}} when CDS job completes with an asset href" do
+      Req.Test.stub(Era5Client, fn conn ->
+        case conn.request_path do
+          "/api/retrieve/v1/jobs/cds-job-123" ->
+            Req.Test.json(conn, %{"status" => "successful"})
+
+          "/api/retrieve/v1/jobs/cds-job-123/results" ->
+            Req.Test.json(conn, %{
+              "asset" => %{"value" => %{"href" => "https://example.com/file.grib"}}
+            })
+        end
+      end)
+
+      assert {:done, {:url, "https://example.com/file.grib"}} =
+               Era5Client.check_status("cds-job-123")
+    end
+
+    test "returns {:failed, _} when CDS reports the job failed" do
+      Req.Test.stub(Era5Client, fn conn ->
+        assert conn.request_path == "/api/retrieve/v1/jobs/cds-job-123"
+        Req.Test.json(conn, %{"status" => "failed", "message" => "input out of range"})
+      end)
+
+      assert {:failed, reason} = Era5Client.check_status("cds-job-123")
+      assert reason =~ "input out of range"
+    end
+  end
+
+  describe "delete_job/1" do
+    test "sends DELETE to the CDS jobs endpoint and returns :ok on 204" do
+      test_pid = self()
+
+      Req.Test.stub(Era5Client, fn conn ->
+        send(test_pid, {:delete_request, conn.method, conn.request_path})
+        Plug.Conn.resp(conn, 204, "")
+      end)
+
+      assert :ok = Era5Client.delete_job("cds-job-123")
+      assert_received {:delete_request, "DELETE", "/api/retrieve/v1/jobs/cds-job-123"}
+    end
+
+    test "also treats 200/202/404 as success (CDS sometimes returns 200 or the job may already be gone)" do
+      for status <- [200, 202, 404] do
+        Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, status, "") end)
+        assert :ok = Era5Client.delete_job("cds-job-#{status}")
+      end
+    end
+
+    test "returns {:error, _} for unexpected statuses" do
+      Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, 500, "server error") end)
+      assert {:error, reason} = Era5Client.delete_job("cds-job-500")
+      assert reason =~ "HTTP 500"
+    end
+  end
+
+  describe "download_source_to_file/3" do
+    test "writes a {:body, binary} source directly to the target path" do
+      path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
+
+      try do
+        assert :ok = Era5Client.download_source_to_file({:body, "grib-payload"}, nil, path)
+        assert File.read!(path) == "grib-payload"
+      after
+        File.rm(path)
+      end
+    end
+
+    test "streams a {:url, href} source through Req into the target path" do
+      Req.Test.stub(Era5Client, fn conn ->
+        assert conn.method == "GET"
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/octet-stream")
+        |> Plug.Conn.resp(200, "streamed-grib-data")
+      end)
+
+      path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
+
+      try do
+        assert :ok =
+                 Era5Client.download_source_to_file(
+                   {:url, "https://example.com/file.grib"},
+                   "test-key",
+                   path
+                 )
+
+        assert File.read!(path) == "streamed-grib-data"
+      after
+        File.rm(path)
+      end
+    end
+
+    test "removes partial file and returns {:error, _} on HTTP error" do
+      Req.Test.stub(Era5Client, fn conn ->
+        Plug.Conn.resp(conn, 503, "")
+      end)
+
+      path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
+
+      assert {:error, reason} =
+               Era5Client.download_source_to_file(
+                 {:url, "https://example.com/file.grib"},
+                 "test-key",
+                 path
+               )
+
+      assert reason =~ "HTTP 503"
+      refute File.exists?(path)
+    end
+  end
+end
diff --git a/test/microwaveprop/workers/era5_poll_worker_test.exs b/test/microwaveprop/workers/era5_poll_worker_test.exs
new file mode 100644
index 00000000..2b73074b
--- /dev/null
+++ b/test/microwaveprop/workers/era5_poll_worker_test.exs
@@ -0,0 +1,96 @@
+defmodule Microwaveprop.Workers.Era5PollWorkerTest do
+  use Microwaveprop.DataCase, async: false
+
+  alias Microwaveprop.Weather.Era5CdsJob
+  alias Microwaveprop.Weather.Era5Client
+  alias Microwaveprop.Workers.Era5PollWorker
+
+  setup do
+    System.put_env("ERA5_CDS_API_KEY", "test-key")
+    on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
+    :ok
+  end
+
+  defp insert_cds_job(overrides \\ %{}) do
+    attrs =
+      Map.merge(
+        %{
+          year: 2014,
+          month: 3,
+          tile_lat: 32,
+          tile_lon: -98,
+          single_job_id: "cds-single-123",
+          pressure_job_id: "cds-pressure-456",
+          submitted_at: ~U[2026-04-13 21:00:00Z]
+        },
+        overrides
+      )
+
+    {:ok, row} = %Era5CdsJob{} |> Era5CdsJob.changeset(attrs) |> Repo.insert()
+    row
+  end
+
+  describe "perform/1 — row lifecycle" do
+    test "returns :ok and does nothing when the era5_cds_jobs row no longer exists" do
+      # Row was already processed/deleted by a previous poll run; snooze is
+      # scheduled behind us in time; just no-op.
+      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted" end)
+
+      missing_id = Ecto.UUID.generate()
+
+      assert :ok =
+               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => missing_id}})
+    end
+  end
+
+  describe "perform/1 — still running" do
+    test "returns {:snooze, seconds} and bumps poll_count when either CDS job is still running" do
+      row = insert_cds_job()
+
+      Req.Test.stub(Era5Client, fn conn ->
+        assert conn.method == "GET"
+        Req.Test.json(conn, %{"status" => "running"})
+      end)
+
+      assert {:snooze, snooze_s} =
+               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
+
+      assert snooze_s > 0
+
+      reloaded = Repo.get!(Era5CdsJob, row.id)
+      assert reloaded.poll_count == 1
+      assert reloaded.last_polled_at
+    end
+  end
+
+  describe "perform/1 — failure" do
+    test "returns {:error, _}, deletes the row, and deletes both CDS jobs when a leg fails" do
+      row = insert_cds_job()
+      test_pid = self()
+
+      Req.Test.stub(Era5Client, fn conn ->
+        cond do
+          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
+            Req.Test.json(conn, %{"status" => "failed", "message" => "bad input"})
+
+          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
+            Req.Test.json(conn, %{"status" => "running"})
+
+          conn.method == "DELETE" ->
+            send(test_pid, {:delete, conn.request_path})
+            Plug.Conn.resp(conn, 204, "")
+        end
+      end)
+
+      assert {:error, reason} =
+               Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
+
+      assert reason =~ "bad input"
+      refute Repo.get(Era5CdsJob, row.id)
+
+      # Both CDS jobs are deleted to free server-side quota.
+      assert_received {:delete, "/api/retrieve/v1/jobs/cds-single-123"}
+      assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
+    end
+  end
+end
diff --git a/test/microwaveprop/workers/era5_submit_worker_test.exs b/test/microwaveprop/workers/era5_submit_worker_test.exs
new file mode 100644
index 00000000..4a91c762
--- /dev/null
+++ b/test/microwaveprop/workers/era5_submit_worker_test.exs
@@ -0,0 +1,136 @@
+defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do
+  use Microwaveprop.DataCase, async: false
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  import Ecto.Query
+
+  alias Microwaveprop.Weather.Era5CdsJob
+  alias Microwaveprop.Weather.Era5Client
+  alias Microwaveprop.Weather.Era5Profile
+  alias Microwaveprop.Workers.Era5PollWorker
+  alias Microwaveprop.Workers.Era5SubmitWorker
+
+  @default_args %{"year" => 2014, "month" => 3, "tile_lat" => 32, "tile_lon" => -98}
+
+  setup do
+    System.put_env("ERA5_CDS_API_KEY", "test-key")
+    on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
+    :ok
+  end
+
+  describe "perform/1 — short-circuit paths" do
+    test "returns {:ok, :cached} when the month-tile already has era5_profiles" do
+      # Seed a profile inside the (2014-03, tile 32,-98) window.
+      %Era5Profile{}
+      |> Era5Profile.changeset(%{
+        lat: 32.5,
+        lon: -97.25,
+        valid_time: ~U[2014-03-15 12:00:00Z],
+        profile: []
+      })
+      |> Repo.insert!()
+
+      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted on cache hit" end)
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert {:ok, :cached} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
+        assert [] = all_enqueued(worker: Era5PollWorker)
+      end)
+    end
+
+    test "returns {:ok, :already_submitted} when an era5_cds_jobs row already exists" do
+      {:ok, _} =
+        %Era5CdsJob{}
+        |> Era5CdsJob.changeset(%{
+          year: 2014,
+          month: 3,
+          tile_lat: 32,
+          tile_lon: -98,
+          single_job_id: "pre-existing-single",
+          pressure_job_id: "pre-existing-pressure",
+          submitted_at: DateTime.truncate(DateTime.utc_now(), :second)
+        })
+        |> Repo.insert()
+
+      Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted on duplicate submit" end)
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert {:ok, :already_submitted} =
+                 Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
+      end)
+    end
+  end
+
+  describe "perform/1 — happy path" do
+    test "submits both CDS jobs, persists the row, and enqueues a poll worker" do
+      Req.Test.stub(Era5Client, fn conn ->
+        assert conn.method == "POST"
+
+        dataset_id =
+          cond do
+            String.contains?(conn.request_path, "single-levels") -> "single-job-123"
+            String.contains?(conn.request_path, "pressure-levels") -> "pressure-job-456"
+          end
+
+        Req.Test.json(conn, %{"jobID" => dataset_id})
+      end)
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert {:ok, %Era5CdsJob{} = row} =
+                 Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
+
+        assert row.single_job_id == "single-job-123"
+        assert row.pressure_job_id == "pressure-job-456"
+        assert row.year == 2014
+        assert row.month == 3
+        assert row.tile_lat == 32
+        assert row.tile_lon == -98
+
+        # Row is persisted in the DB.
+        assert Repo.aggregate(
+                 from(j in Era5CdsJob,
+                   where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
+                 ),
+                 :count,
+                 :id
+               ) == 1
+
+        # A poll worker was enqueued for this row, scheduled ~5 min from now.
+        [poll_job] = all_enqueued(worker: Era5PollWorker)
+        assert poll_job.args["era5_cds_job_id"] == row.id
+        assert DateTime.after?(poll_job.scheduled_at, DateTime.utc_now())
+      end)
+    end
+
+    test "returns {:error, _} and persists no row if either CDS submit fails" do
+      Req.Test.stub(Era5Client, fn conn ->
+        if String.contains?(conn.request_path, "pressure-levels") do
+          Plug.Conn.resp(conn, 400, ~s({"detail":"bad pressure request"}))
+        else
+          Req.Test.json(conn, %{"jobID" => "single-ok"})
+        end
+      end)
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert {:error, reason} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
+        assert is_binary(reason)
+
+        # No orphaned row.
+        assert Repo.aggregate(Era5CdsJob, :count, :id) == 0
+
+        # No poll worker enqueued.
+        assert [] = all_enqueued(worker: Era5PollWorker)
+      end)
+    end
+  end
+
+  describe "unique constraint" do
+    test "collapses duplicate (year, month, tile_lat, tile_lon) Oban enqueues" do
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, j1} = Oban.insert(Era5SubmitWorker.new(@default_args))
+        {:ok, j2} = Oban.insert(Era5SubmitWorker.new(@default_args))
+        assert j1.id == j2.id
+      end)
+    end
+  end
+end