From 94b83e472f1cb1044f913e70f63e87b3e85f464b Mon Sep 17 00:00:00 2001 From: Bill Guowei Yang Date: Wed, 9 Sep 2026 15:56:29 -0400 Subject: [PATCH 1/2] fix: allow eight minutes between polls before failing health checks --- millpond/main.py | 4 ++-- millpond/server.py | 3 ++- tests/unit/test_server.py | 18 ++++++++++++++++++ 3 files changed, 22 insertions(+), 3 deletions(-) diff --git a/millpond/main.py b/millpond/main.py index 1abd693..9547030 100644 --- a/millpond/main.py +++ b/millpond/main.py @@ -23,9 +23,9 @@ log = logging.getLogger(__name__) _LAG_SAMPLE_INTERVAL_S = 60.0 # how often to query watermark offsets for lag metrics -_HEARTBEAT_INTERVAL_S = 60.0 # periodic log when idle (well under 300s liveness timeout) +_HEARTBEAT_INTERVAL_S = 60.0 # periodic log when idle (well under 480s liveness timeout) # Longest a single consume() may block. record_poll() runs only after consume -# returns, and server.health marks the process dead at max_poll_age_s=300 — +# returns, and server.health marks the process dead at max_poll_age_s=480 — # so a consume timeout derived from a large FLUSH_INTERVAL_MS (e.g. 10min) # would starve the liveness probe on a quiet topic and SIGKILL the pod. # 60s also keeps the idle heartbeat cadence honest. diff --git a/millpond/server.py b/millpond/server.py index 42cc142..dee6149 100644 --- a/millpond/server.py +++ b/millpond/server.py @@ -11,7 +11,8 @@ class _HealthState: """Tracks recency of poll and flush for health checks.""" - def __init__(self, max_poll_age_s: float = 300): + # Allow slow writes to finish while staying below Kafka's 600s max poll interval. + def __init__(self, max_poll_age_s: float = 480): self.max_poll_age_s = max_poll_age_s self._last_poll: float = 0 self._last_flush: float = 0 diff --git a/tests/unit/test_server.py b/tests/unit/test_server.py index 8a17e55..5596e1b 100644 --- a/tests/unit/test_server.py +++ b/tests/unit/test_server.py @@ -13,6 +13,24 @@ def test_started(self): h.mark_started() assert h.is_alive() + def test_default_timeout_allows_slow_write_then_expires(self, monkeypatch): + now = 1000.0 + monkeypatch.setattr("millpond.server.time.monotonic", lambda: now) + h = _HealthState() + h.mark_started() + + now += 479 + assert h.is_alive() + assert h.is_ready() + + now += 1 + assert not h.is_alive() + assert not h.is_ready() + + h.record_poll() + assert h.is_alive() + assert h.is_ready() + def test_stale_poll(self): h = _HealthState(max_poll_age_s=0.01) h.mark_started() From 336a83434f9758f28257d41d9e828628bb6faf4d Mon Sep 17 00:00:00 2001 From: Bill Guowei Yang Date: Wed, 9 Sep 2026 15:57:56 -0400 Subject: [PATCH 2/2] test: remove unnecessary health threshold test --- tests/unit/test_server.py | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/tests/unit/test_server.py b/tests/unit/test_server.py index 5596e1b..8a17e55 100644 --- a/tests/unit/test_server.py +++ b/tests/unit/test_server.py @@ -13,24 +13,6 @@ def test_started(self): h.mark_started() assert h.is_alive() - def test_default_timeout_allows_slow_write_then_expires(self, monkeypatch): - now = 1000.0 - monkeypatch.setattr("millpond.server.time.monotonic", lambda: now) - h = _HealthState() - h.mark_started() - - now += 479 - assert h.is_alive() - assert h.is_ready() - - now += 1 - assert not h.is_alive() - assert not h.is_ready() - - h.record_poll() - assert h.is_alive() - assert h.is_ready() - def test_stale_poll(self): h = _HealthState(max_poll_age_s=0.01) h.mark_started()