fix(weather): analyze_all is now fire-and-forget

Running ANALYZE serially against the 13M-row HRRR monthly partitions
took longer than the kubectl exec idle timeout (~300s per partition
on the Turing Pi 2), so the one-shot RPC kept getting killed before
the big partitions ran. Moves the loop into a supervised Task on
the caller's node so the RPC returns immediately; the BEAM keeps
running ANALYZE regardless of kubectl connection state.

Also sorts tables ascending by n_live_tup so small tables finish
first and their fresh stats are visible to the planner right away.
Progress is logged per table via `Logger.info`.

lib/microwaveprop/weather.ex

@@ -411,8 +411,12 @@ defmodule Microwaveprop.Weather do
   end
 
   @doc """
-  Run `ANALYZE` on every public-schema table that has not been auto-
-  analyzed recently. Callable from a release shell:
+  Kick off `ANALYZE` on every public-schema table that hasn't been
+  auto-analyzed recently. Returns immediately with the list of tables
+  it will visit; the actual work runs in a supervised Task on the
+  current node so a broken `kubectl exec` doesn't abort it mid-way.
+
+  Callable from a release shell:
 
       bin/microwaveprop rpc 'Microwaveprop.Weather.analyze_all()'
 
@@ -422,9 +426,11 @@ defmodule Microwaveprop.Weather do
   histograms to work with.
 
   `skip_recent` skips anything auto-analyzed in the last N seconds
-  (default 6 h) so re-running is cheap.
+  (default 6 h) so re-running is cheap. Progress is logged via
+  `Logger.info` — tail `kubectl -n prop logs deploy/prop-backfill`
+  to watch.
   """
-  @spec analyze_all(keyword()) :: %{analyzed: non_neg_integer(), skipped: non_neg_integer()}
+  @spec analyze_all(keyword()) :: %{queued: non_neg_integer(), tables: [String.t()]}
   def analyze_all(opts \\ []) do
     skip_recent = Keyword.get(opts, :skip_recent_seconds, 6 * 3600)
 
@@ -434,25 +440,33 @@ defmodule Microwaveprop.Weather do
     WHERE schemaname = 'public'
       AND (last_autoanalyze IS NULL
            OR last_autoanalyze < now() - ($1 || ' seconds')::interval)
+    ORDER BY n_live_tup ASC
     """
 
     %{rows: rows} = Repo.query!(q, [Integer.to_string(skip_recent)])
     tables = Enum.map(rows, fn [n] -> n end)
 
-    Enum.each(tables, fn t ->
-      Repo.query!("ANALYZE #{quote_ident(t)}", [], timeout: :infinity)
+    # Fire-and-forget: run each ANALYZE on the pod's own node so the
+    # caller returns before any single query finishes. Small tables
+    # first keeps progress visible early.
+    Task.Supervisor.start_child({:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}, fn ->
+      Enum.each(tables, fn t ->
+        Logger.info("Weather.analyze_all: ANALYZE #{t}")
+        started = System.monotonic_time(:millisecond)
+
+        try do
+          Repo.query!("ANALYZE #{quote_ident(t)}", [], timeout: :infinity)
+          elapsed = System.monotonic_time(:millisecond) - started
+          Logger.info("Weather.analyze_all: ANALYZE #{t} done in #{elapsed}ms")
+        rescue
+          e -> Logger.warning("Weather.analyze_all: ANALYZE #{t} failed: #{inspect(e)}")
+        end
+      end)
+
+      Logger.info("Weather.analyze_all: complete (#{length(tables)} tables)")
     end)
 
-    skipped =
-      Repo.query!("""
-      SELECT count(*) FROM pg_stat_user_tables
-      WHERE schemaname = 'public'
-        AND last_autoanalyze >= now() - ('#{skip_recent} seconds')::interval
-      """).rows
-      |> hd()
-      |> hd()
-
-    %{analyzed: length(tables), skipped: skipped}
+    %{queued: length(tables), tables: tables}
   end
 
   # Defensive: pg_stat_user_tables returns legitimate identifiers from