Writer liveness probe fails during prolonged icebox backpressure

Symptom

Writer pod is killed by kubelet (SIGTERM ~9 min after start) when the
icebox upstream is fully down. The pod is doing the right thing —
bounded backoff on 503 — but /healthz flips to 503 after ~5 min and
liveness trips.

Restarting doesn't help: same backend, same problem. End state is a
CrashLoop while millpond does the correct thing.

Why

server._HealthState.is_alive() returns true iff
(now - self._last_poll) < self.max_poll_age_s (default 300s).

record_poll() is only called in the main loop after kafka.consume()
returns (main.py:447). When the main thread is stuck inside
_flush → _write_with_retry → IceboxSink.write → IceboxClient.register_file,
_last_poll never refreshes.

Worst-case stuck-but-healthy window with current chart defaults:

• ICEBOX_MAX_ATTEMPTS=6, ICEBOX_MAX_BACKOFF_S=30 → ~3 min per
  register_file call (server-sent retry_after_s=120 is capped at
  max_backoff_s).
• _WRITE_MAX_RETRIES=3 outer → up to ~9 min total before
  _write_with_retry re-raises.

max_poll_age_s=300 is exceeded at ~5 min. Liveness
(periodSeconds=30, failureThreshold=3) kills the pod ~90s later.

Observed in mw-prod-us, millpond-events-icebox-0:

Proposed fix

Heartbeat the health state from inside the icebox retry sleeps. The
writer process IS alive and IS applying the correct backoff — the
liveness signal should reflect "the writer thread is still ticking",
not "data is flowing through".

Shape:

• Add heartbeat: Callable[[], None] | None = None to
  IceboxClient.__init__.
• Call it before each time.sleep in _sleep_for_attempt (before, not
  after — a single tick covers the entire upcoming sleep).
• Wire server.health.record_poll through from
  sink.make_sink / main.py.
• Default None keeps tests and library use unchanged.

record_poll semantics broaden slightly from "polled Kafka" to
"writer thread is ticking" — document in _HealthState.record_poll.

Alternatives considered

• Bump max_poll_age_s. Brittle: ties the constant to
  _WRITE_MAX_RETRIES × ICEBOX_MAX_ATTEMPTS × ICEBOX_MAX_BACKOFF_S.
  Any retry-knob change silently breaks the relationship.
• Fail-fast on IceboxBackpressureExhausted in
  _write_with_retry. Reduces the stuck window but crashes the pod
  on every transient icebox hiccup — Kafka rebalance churn is worse
  than the current behavior.

Test plan

• Unit: IceboxClient(heartbeat=Mock()) — assert called
  max_attempts - 1 times on a 503 storm.
• Unit: IceboxClient(heartbeat=None) — current behavior unchanged.
• Manual: in mw-dev, scale millpond-events-icebox-coord to 0,
  observe millpond-events-icebox-0 stays Running and /healthz keeps
  returning 200 across multiple backpressure cycles.

Out of scope

• Tuning ICEBOX_MAX_ATTEMPTS / ICEBOX_MAX_BACKOFF_S.
• Pod-level circuit breaker (skip writes when icebox is known-down).

Start with IceboxClient._sleep_for_attempt and the _HealthState.record_poll/is_alive flow; main.py:447 and sink.make_sink show the health callback wiring points. Run the heartbeat and no-heartbeat unit tests described in the issue, then verify in mw-dev that the writer stays Running and /healthz remains 200 during repeated backpressure cycles.