TD-P005: heartbeat process_metrics report lifetime aggregates as instantaneous (#45)

Author: durable-workflow-ops

CHANGELOG.md

@@ -6,6 +6,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- Worker heartbeat `process_metrics` now report instantaneous values
+  instead of process-lifetime aggregates. `cpu_percent` is the share of
+  wall time the worker spent on CPU during the interval since the
+  previous heartbeat (previously the lifetime average, which converged
+  to a fixed value within minutes and stopped tracking live load), and
+  `memory_bytes` is the current resident set size sampled from
+  `/proc/self/statm` on Linux (previously `ru_maxrss`, which is the
+  process-lifetime high-water mark and never decreased after a startup
+  spike). Platforms without `/proc` no longer report `memory_bytes`
+  rather than reporting a misleading peak. The heartbeat protocol shape
+  is unchanged; the server records whatever the worker sends, so the
+  Worker Status surface starts showing accurate live numbers as soon as
+  workers upgrade.
+
 ### Changed
 - `tests/test_client.py` now closes the `schedule.history` polyglot
   parity slice. `test_get_schedule_history_matches_polyglot_fixture`

src/durable_workflow/worker.py

@@ -334,6 +334,13 @@ def __init__(
         self._activity_inflight = 0
         self._heartbeat_interval = float(heartbeat_interval)
         self._process_started_at = time.time()
+        # CPU sampling baseline. The heartbeat reports an *instantaneous*
+        # cpu_percent — CPU time burned in the interval since the previous
+        # heartbeat, divided by that interval — rather than the lifetime
+        # average. ``_last_cpu_sample_at`` is None until the first sample
+        # has been taken; subsequent samples diff against it.
+        self._last_cpu_sample_at: float | None = None
+        self._last_cpu_total_seconds: float = 0.0
         configured_metrics = metrics if metrics is not None else getattr(client, "metrics", NOOP_METRICS)
         self.metrics: MetricsRecorder = configured_metrics or NOOP_METRICS
         self.interceptors = tuple(interceptors)
@@ -1152,32 +1159,54 @@ def _current_process_metrics(self) -> dict[str, Any]:
         import os
         import socket
 
+        now = time.time()
         metrics: dict[str, Any] = {
-            "process_uptime_seconds": int(time.time() - self._process_started_at),
+            "process_uptime_seconds": int(now - self._process_started_at),
             "process_id": os.getpid(),
         }
 
+        # ``memory_bytes`` is the *current* resident set size, not the
+        # lifetime peak. ``resource.getrusage().ru_maxrss`` is the high-
+        # water mark since the process started, which masks freed memory
+        # and never decreases — wrong shape for a heartbeat metric meant
+        # to show what the worker is using right now. On Linux we read
+        # the second field of ``/proc/self/statm`` (resident pages) and
+        # multiply by the page size. Platforms without ``/proc`` get no
+        # ``memory_bytes`` field rather than a misleading lifetime peak.
+        if sys.platform.startswith("linux"):
+            try:
+                with open("/proc/self/statm") as statm:
+                    fields = statm.read().split()
+                if len(fields) >= 2:
+                    metrics["memory_bytes"] = int(fields[1]) * os.sysconf("SC_PAGE_SIZE")
+            except (OSError, ValueError):
+                pass
+
+        # ``cpu_percent`` is the share of wall time the process spent on
+        # CPU during the interval since the previous heartbeat — not the
+        # lifetime average, which converges to a fixed value and stops
+        # tracking live load. The first sample bootstraps from process
+        # start so the very first heartbeat still has a number.
         try:
             import resource
 
             usage = resource.getrusage(resource.RUSAGE_SELF)
-            # ru_maxrss is kilobytes on Linux and bytes on macOS — normalize
-            # to bytes. The server stores whatever is sent so the units stay
-            # consistent across SDKs.
-            if sys.platform == "darwin":
-                metrics["memory_bytes"] = int(usage.ru_maxrss)
-            else:
-                metrics["memory_bytes"] = int(usage.ru_maxrss) * 1024
-
             cpu_seconds = float(usage.ru_utime) + float(usage.ru_stime)
-            wall_seconds = max(0.001, time.time() - self._process_started_at)
+            if self._last_cpu_sample_at is None:
+                interval = max(0.001, now - self._process_started_at)
+                delta_cpu = max(0.0, cpu_seconds)
+            else:
+                interval = max(0.001, now - self._last_cpu_sample_at)
+                delta_cpu = max(0.0, cpu_seconds - self._last_cpu_total_seconds)
             metrics["cpu_percent"] = max(
-                0.0, min(100.0, round((cpu_seconds / wall_seconds) * 100.0, 2))
+                0.0, min(100.0, round((delta_cpu / interval) * 100.0, 2))
             )
+            self._last_cpu_sample_at = now
+            self._last_cpu_total_seconds = cpu_seconds
         except (ImportError, OSError):
-            # `resource` is POSIX-only — Windows skips getrusage but still
-            # reports pid + uptime + host so the operator surface remains
-            # populated.
+            # ``resource`` is POSIX-only — Windows skips the CPU sample
+            # but still reports pid + uptime + host so the operator
+            # surface stays populated.
             pass
 
         try:

tests/test_worker.py

@@ -3,6 +3,7 @@
 import asyncio
 import contextlib
 import logging
+import sys
 from unittest.mock import AsyncMock
 
 import pytest
@@ -1436,6 +1437,80 @@ async def test_heartbeat_loop_survives_transient_errors(
             await run_task
         assert mock_client.heartbeat_worker.call_count >= 2
 
+    def test_process_metrics_cpu_percent_is_instantaneous_not_lifetime(
+        self, mock_client: AsyncMock, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """``cpu_percent`` reflects only the interval since the previous
+        heartbeat. A worker that was busy at startup and idle ever since
+        used to keep reporting the lifetime average forever, hiding the
+        fact that it is no longer doing CPU work."""
+
+        import resource
+
+        from durable_workflow import worker as worker_mod
+
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+        )
+        worker._process_started_at = 1000.0
+
+        class FakeUsage:
+            def __init__(self, utime: float, stime: float) -> None:
+                self.ru_utime = utime
+                self.ru_stime = stime
+                self.ru_maxrss = 0
+
+        fake_now = {"value": 1010.0}
+        fake_usage = {"value": FakeUsage(7.0, 1.0)}
+
+        monkeypatch.setattr(worker_mod.time, "time", lambda: fake_now["value"])
+        monkeypatch.setattr(resource, "getrusage", lambda _who: fake_usage["value"])
+
+        # First sample: 8s of CPU over 10s of wall time since process start = 80%.
+        first = worker._current_process_metrics()
+        assert first["cpu_percent"] == 80.0
+
+        # Ten more wall seconds with only 0.5s of additional CPU (the
+        # worker went idle). Lifetime average would still be 8.5/20 =
+        # 42.5%, but the instantaneous reading is 5%.
+        fake_now["value"] = 1020.0
+        fake_usage["value"] = FakeUsage(7.3, 1.2)
+        second = worker._current_process_metrics()
+        assert second["cpu_percent"] == 5.0
+
+        # Fully idle for ten more seconds. Used to be ~32% (lifetime
+        # average), should now be 0.
+        fake_now["value"] = 1030.0
+        third = worker._current_process_metrics()
+        assert third["cpu_percent"] == 0.0
+        assert third["process_uptime_seconds"] == 30
+
+    def test_process_metrics_memory_bytes_is_current_resident_set(
+        self, mock_client: AsyncMock
+    ) -> None:
+        """``memory_bytes`` is the current resident set size on Linux —
+        read from ``/proc/self/statm`` — not ``ru_maxrss``, which is the
+        process-lifetime high-water mark and never decreases after a
+        startup spike."""
+
+        if not sys.platform.startswith("linux"):
+            pytest.skip("memory_bytes is only sampled on Linux")
+
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+        )
+
+        metrics = worker._current_process_metrics()
+        assert "memory_bytes" in metrics
+        assert isinstance(metrics["memory_bytes"], int)
+        assert metrics["memory_bytes"] > 0
+
 
 class TestRunUntil:
     @pytest.mark.asyncio