prop/lib/microwaveprop/application.ex
Graham McIntire 15ce50365f
perf(concurrency): partition Task.Supervisor to remove bottleneck on heavy async_stream paths
Wrap a Task.Supervisor in a PartitionSupervisor (one partition per
scheduler) and route the four sustained async_stream callers through
`{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}`.
Concurrent work — the hourly PropagationGridWorker pulling f00-f18
HRRR range downloads, the GEFS worker scoring its grid, and the
Recalibrator's 20-way positive/negative fan-out — no longer funnels
through a single supervisor PID.

Light/one-shot Task.async + Task.start sites (hrrr_client surface/
pressure fan-out, application boot warmup, weather fire-and-forget)
are left alone; partitioning only helps under sustained concurrency.
2026-04-21 14:20:29 -05:00

132 lines
4.8 KiB
Elixir

defmodule Microwaveprop.Application do
# See https://hexdocs.pm/elixir/Application.html
# for more information on OTP Applications
@moduledoc false
use Application
@build_timestamp DateTime.utc_now()
@impl true
def start(_type, _args) do
topologies = Application.get_env(:libcluster, :topologies, [])
# Auto-attach OpenTelemetry instrumentation handlers for Phoenix,
# Oban, and Bandit. Ecto instrumentation intentionally omitted —
# one span per query produces Oban-job traces with thousands of
# spans that overwhelm Tempo ingestion.
OpentelemetryPhoenix.setup(adapter: :bandit)
OpentelemetryBandit.setup()
OpentelemetryOban.setup()
children = [
MicrowavepropWeb.Telemetry,
Microwaveprop.PromEx,
Microwaveprop.Repo,
{Cluster.Supervisor, [topologies, [name: Microwaveprop.ClusterSupervisor]]},
{Phoenix.PubSub, name: Microwaveprop.PubSub},
# Partitioned Task.Supervisor — callers route heavy `async_stream`
# work through `{:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor,
# self()}}` so concurrent workers (recalibrator, HRRR/GEFS range
# downloads, GEFS scoring) do not serialize on a single supervisor PID.
{PartitionSupervisor,
child_spec: Task.Supervisor, name: Microwaveprop.TaskSupervisor, partitions: System.schedulers_online()},
Microwaveprop.Cache,
Microwaveprop.Propagation.ScoreCache,
Microwaveprop.Propagation.ScoreCacheReconciler,
Microwaveprop.Weather.GridCache,
{Microwaveprop.Weather.IemRateLimiter,
interval_ms: Application.get_env(:microwaveprop, :iem_rate_limiter_interval_ms, 700)},
Microwaveprop.Weather.NexradCache,
Microwaveprop.Qrz.Client,
{Oban, Application.fetch_env!(:microwaveprop, Oban)},
Microwaveprop.RepoListener,
# Start to serve requests, typically the last entry
MicrowavepropWeb.Endpoint,
Microwaveprop.Propagation.FreshnessMonitor
]
# See https://hexdocs.pm/elixir/Supervisor.html
# for other strategies and supported options
opts = [strategy: :one_for_one, name: Microwaveprop.Supervisor]
result = Supervisor.start_link(children, opts)
# Attach the Oban exception telemetry tap so per-worker recovery
# callbacks and structured logs fire on job failure. Must run after
# the Oban supervisor starts (above in `children`).
:ok = Microwaveprop.ObanErrorReporter.attach()
# Run migrations after Repo is started
_ = Microwaveprop.Release.migrate()
# Warm GridCache from the latest persisted ProfilesFile so /weather
# has data immediately after a pod restart. Runs after the GridCache
# GenServer is up (it's in `children` above).
#
# Must run in a short-lived process, not inline in the application
# master: when executed here the loaded rows live on the app master's
# heap and never GC (it's idle after boot), pinning ~800 MiB for the
# lifetime of the pod.
{:ok, _pid} = Task.start(fn -> Microwaveprop.Weather.warm_grid_cache_from_latest_profile() end)
# Load ML model in dev/test only (Nx/Axon not compiled for prod)
if Application.get_env(:microwaveprop, :load_ml_model, false) do
Microwaveprop.Propagation.load_ml_model()
end
result
end
@impl true
def config_change(changed, _new, removed) do
MicrowavepropWeb.Endpoint.config_change(changed, removed)
:ok
end
@doc """
Returns the deployment (image build) timestamp.
Sources, in priority order:
1. The `/app/BUILD_TIMESTAMP` file baked into the Docker image at build
time (the CI pipeline passes `--build-arg BUILD_TIMESTAMP=...` in the
final stage, so this value is fresh on every image and is not affected
by earlier-stage layer caching). The path is overridable via
`config :microwaveprop, :build_timestamp_file` for tests.
2. The `DEPLOY_TIMESTAMP` environment variable, for environments that set
it imperatively (e.g. `kubectl set env`).
3. The compile-time fallback captured when this module was compiled.
All accepted values are ISO 8601, e.g. `2026-04-17T18:30:00Z`.
"""
@spec build_timestamp() :: DateTime.t()
def build_timestamp do
with :error <- from_file(),
:error <- from_env() do
@build_timestamp
end
end
defp from_file do
path = Application.get_env(:microwaveprop, :build_timestamp_file, "/app/BUILD_TIMESTAMP")
case File.read(path) do
{:ok, contents} -> parse_iso8601(contents)
{:error, _} -> :error
end
end
defp from_env do
case System.get_env("DEPLOY_TIMESTAMP") do
nil -> :error
string -> parse_iso8601(string)
end
end
defp parse_iso8601(string) do
case string |> String.trim() |> DateTime.from_iso8601() do
{:ok, dt, _offset} -> dt
{:error, _} -> :error
end
end
end