Re-queue worker pods on transient 502/503/504 from the Kubernetes API

[1fanwang]

Stacked on #68480. This PR's own change is the last commit (52ea143705); the
earlier commits belong to #68480, which it builds on. Review just that commit
(git diff 72dd7c3a0c..HEAD), or after #68480 merges and GitHub re-bases the diff.

Why

KubernetesExecutor._handle_pod_publish_error re-queues a worker-pod CREATE that fails with
an exceeded-quota / stale-version conflict, 500, or 429, but lets 502 / 503 / 504
fall through and fail the task immediately. Those are exactly the gateway / unavailable /
timeout codes that spike when the API server or an admission webhook is under load — the regime
this executor path is meant to survive. Concretely, the API server returns 503 with
Retry-After: 1 on every graceful shutdown (rollout, node drain, control-plane autoscale —
see apiserver waitgroup.go), so an in-flight pod CREATE during any apiserver roll currently
kills the task outright. client-go retries all 5xx and honours Retry-After; the executor does
not.

What

Add 502 / 503 / 504 to the transient set in the shared _handle_pod_publish_error, so a
task that hits one is re-queued (bounded by task_publish_max_retries) instead of failed. Because
both the sequential and the concurrent creation paths route through that one method, this hardens
both at once, using the executor's existing re-queue-next-loop model — no in-line blocking retry
that would stall the scheduler loop.

When the response carries Retry-After (a shutting-down apiserver always sends it on 503),
pause the whole loop until then via create_pods_after, exactly as the 429 path already does;
other transient 5xx without the header simply retry on the next loop like 500. The Retry-After
parse is guarded so a malformed value can't crash the scheduler loop.

Stacked on #68480 — the shared handler this extends was introduced there.

Tests

Unit, covering both clients through the shared handler: 502 / 504 re-queue and retry on the
next loop; 503 + Retry-After re-queues and backs the loop off; 503 with retries exhausted
fails; and the async client's 503 sets create_pods_after. Each fails without this change.

E2E

Live kind (real API server, real KubernetesExecutor.sync()), with a proxy injecting
503 + Retry-After: 1 on the first CREATE per task — identical config (task_publish_max_retries=3),
only the provider code differs:

Run (retries=3)	first loop	outcome	pods created
before (this branch's parent)	6 tasks → 503 → failed	tasks failed	0 / 6
after (async)	503 → all 6 re-queued, backoff	retry creates all	6 / 6
after (sequential)	503 → re-queued each loop, no failures	later loops create all	6 / 6

Raw — after (async path)

queued 6 tasks dag=retrydag run=after2
loop 0: queue 6->6 failed_events=0          # 503 on first attempt -> all re-queued
loop 1: queue 6->0 failed_events=0          # Retry-After backoff elapsed -> retry creates
FINAL run_id=after2 failed_events=0 queue_remaining=0

# injector
INJECT 503 (Retry-After=1) <- task_id=t000..t005   (6x, first attempt)
PASS   201                 <- task_id=t000..t005   (6x, retry)

# cluster
retrydag-t000..t005   Running   (6 pods)

Risk

Behaviour change on the default sequential path: a 502 / 503 / 504 now re-queues instead of
failing the task — bounded by task_publish_max_retries (default 0, i.e. inert unless retries
are already enabled). The existing 429 / 500 / quota / conflict handling is unchanged.