Claude/worker logs memory redis a7vsv8

shepherd_broker/redis.conf

@@ -8,6 +8,10 @@ maxmemory 4gb
 # a TTL, so it can never evict live task streams (which carry no TTL) or,
 # in practice, in-flight queries -- those were written most recently and thus
 # have the most TTL remaining, making them the last candidates, not the first.
+# Worker heartbeat keys are likewise written WITHOUT a TTL for this reason: a
+# short-lived heartbeat would be the first thing evicted under pressure, which
+# would make live workers look crashed. They stay persistent and the monitor
+# reaps stale ones by last_seen age instead (see shepherd_utils/heartbeat.py).
 maxmemory-policy volatile-ttl
 loglevel notice
 # If we want to log to a file

shepherd_utils/config.py

@@ -240,6 +240,32 @@ class Settings(BaseSettings):
     # flood into just the ``broker_down``/``broker_recovered`` pair; a worker
     # that is *genuinely* still down once the window elapses alerts as normal.
     monitor_broker_recovery_grace_sec: int = 30
+    # When the Redis broker hits its maxmemory cap and starts evicting keys, the
+    # short-TTL worker heartbeat keys are prime eviction targets (the deployment
+    # uses volatile-ttl, which sheds nearest-expiry keys first). Evicting a live
+    # worker's heartbeat makes it momentarily read as zero-alive even though it
+    # never disconnected -- key eviction doesn't close client connections. To
+    # stop that from firing a flood of bogus worker-down/crash alerts, worker-down
+    # alerts are suppressed while eviction is active and for this long after the
+    # last observed eviction, giving heartbeats (interval 5s, TTL 15s) time to be
+    # re-registered. A worker that is genuinely still down once the window
+    # elapses alerts as normal.
+    monitor_redis_eviction_grace_sec: int = 30
+    # Redis broker memory-pressure alerting. Modeled on the broker/postgres
+    # health alerts (an automatic engine state machine, not a YAML rule): a
+    # single alert when usage crosses a threshold, at most one re-notify per
+    # cooldown while it stays elevated, and a recovery alert when it falls back
+    # below the warning threshold. The critical alert supersedes the warning
+    # one, so crossing the critical line never also sends the warning alert.
+    # All no-ops when the broker runs uncapped (maxmemory 0).
+    monitor_redis_memory_warning_pct: float = 85.0
+    monitor_redis_memory_critical_pct: float = 95.0
+    # An elevated level must hold this long before it alerts (and a drop must
+    # hold this long before it clears), so a brief spike/dip doesn't trip it.
+    monitor_redis_memory_grace_sec: float = 120.0
+    # Re-notify backoff while memory stays elevated -- a full day by default, so
+    # a sustained condition reminds once a day rather than every few minutes.
+    monitor_redis_memory_cooldown_sec: float = 86400.0
     # Postgres availability alerting. Same debounce idea as the broker: Postgres
     # must stay unreachable this long before the single ``postgres_down`` alert
     # fires, so a transient query timeout doesn't trip it. No recovery-grace

shepherd_utils/heartbeat.py

@@ -7,9 +7,23 @@
 Kubernetes -- this works identically under either, and handles autoscaling.
 
 Key format: ``worker:heartbeat:{stream}:{consumer}``
-TTL: refreshed every ``HEARTBEAT_INTERVAL_SEC``, expires after
-``HEARTBEAT_TTL_SEC``. If a worker crashes or is killed, the key disappears
-within the TTL window and the monitor will surface it as a worker-loss event.
+
+The key is written *without* a Redis TTL, deliberately. The broker runs a
+``volatile-ttl`` maxmemory policy (see ``shepherd_broker/redis.conf``), which
+evicts the keys nearest to expiry first when memory is tight. A short-TTL
+heartbeat would be the very first thing shed under pressure, so a live worker
+would vanish from the keyspace and the monitor would misread it as crashed --
+even though key eviction never actually disconnects the worker. Making the key
+persistent exempts it from eviction entirely (the policy only touches keys that
+carry a TTL, which is also why the no-TTL task streams are safe).
+
+Liveness is therefore judged from the payload's ``last_seen`` timestamp, not
+from the key's existence: a heartbeat is "fresh" (the worker is alive) if it was
+refreshed within ``HEARTBEAT_TTL_SEC``. Every reader agrees on this via
+``is_heartbeat_fresh``. Because a persistent key can't expire itself away when a
+worker crashes, the monitor reaps heartbeats left unrefreshed for longer than
+``HEARTBEAT_REAP_SEC`` so dead consumers don't accumulate. A worker deletes its
+own key on clean shutdown (see ``stop``/``_mark_shutdown_sync``).
 
 On SIGTERM/SIGINT we additionally write a *shutdown marker*
 ``worker:shutdown:{stream}:{consumer}`` synchronously before exiting. The
@@ -34,7 +48,16 @@
 HEARTBEAT_PREFIX = "worker:heartbeat"
 HEARTBEAT_SCAN_PATTERN = f"{HEARTBEAT_PREFIX}:*"
 HEARTBEAT_INTERVAL_SEC = 5
+# A worker whose heartbeat hasn't been refreshed within this window is considered
+# not-alive (``is_heartbeat_fresh`` returns False). Sized comfortably above the
+# ping interval so a single missed tick doesn't flap liveness.
 HEARTBEAT_TTL_SEC = 15
+# Heartbeat keys are persistent (see the module docstring), so a crashed worker's
+# key can't expire on its own. The monitor deletes any heartbeat left unrefreshed
+# for longer than this so dead consumers don't pile up in the keyspace. Well above
+# HEARTBEAT_TTL_SEC so a briefly-paused worker (or a broker blip) isn't reaped and
+# then forced to re-register; a genuinely dead worker is cleaned within ~2 min.
+HEARTBEAT_REAP_SEC = 120
 
 SHUTDOWN_PREFIX = "worker:shutdown"
 SHUTDOWN_SCAN_PATTERN = f"{SHUTDOWN_PREFIX}:*"
@@ -51,6 +74,29 @@ def shutdown_key(stream: str, consumer: str) -> str:
     return f"{SHUTDOWN_PREFIX}:{stream}:{consumer}"
 
 
+def is_heartbeat_fresh(raw: "str | bytes | None", now: "float | None" = None) -> bool:
+    """Whether a heartbeat payload was refreshed within ``HEARTBEAT_TTL_SEC``.
+
+    Heartbeat keys are persistent (no Redis TTL) so they survive maxmemory
+    eviction; liveness is therefore decided from the payload's ``last_seen``
+    age rather than from the key merely existing. This is the single definition
+    every "is this worker alive?" reader shares -- the monitor poller, the task
+    reclaimer, and the janitor -- so they can't drift apart. A missing/malformed
+    payload or absent ``last_seen`` reads as not-fresh (fail safe toward "dead").
+    """
+    if not raw:
+        return False
+    try:
+        last_seen = float(json.loads(raw).get("last_seen", 0) or 0)
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return False
+    if last_seen <= 0:
+        return False
+    if now is None:
+        now = time.time()
+    return (now - last_seen) <= HEARTBEAT_TTL_SEC
+
+
 def _read_rss_bytes() -> "int | None":
     """Resident set size of this process in bytes, read from ``/proc/self``.
 
@@ -66,6 +112,42 @@ def _read_rss_bytes() -> "int | None":
         return None
 
 
+def _read_cgroup_memory_limit() -> "int | None":
+    """The container's memory limit in bytes, or ``None`` if uncapped/unknown.
+
+    Read from the same ``/sys/fs/cgroup`` tree the CPU accounting uses, handling
+    both cgroup v2 (``memory.max``) and v1 (``memory/memory.limit_in_bytes``).
+    Both use a sentinel for "no limit": v2 writes the literal ``"max"``, v1 an
+    enormous page-counter maximum. We return ``None`` for either so the monitor
+    treats the worker as uncapped (and shows raw RSS without a percentage),
+    exactly like the Redis panel does when ``maxmemory`` is 0. Only meaningful
+    inside a container -- on a bare host the cgroup is the whole machine's.
+    """
+    if not _in_container():
+        return None
+    # cgroup v2: a single value or the literal "max".
+    try:
+        with open("/sys/fs/cgroup/memory.max", encoding="ascii") as f:
+            raw = f.read().strip()
+        if raw and raw != "max":
+            limit = int(raw)
+            # Guard against a v2 kernel also exposing a giant sentinel.
+            if 0 < limit < (1 << 62):
+                return limit
+        return None
+    except (OSError, ValueError):
+        pass
+    # cgroup v1: a byte count; "unlimited" is a near-INT64_MAX page-counter max.
+    try:
+        with open("/sys/fs/cgroup/memory/memory.limit_in_bytes", encoding="ascii") as f:
+            limit = int(f.read().strip())
+        if 0 < limit < (1 << 62):
+            return limit
+    except (OSError, ValueError):
+        pass
+    return None
+
+
 def _read_proc_self_cpu_seconds() -> "float | None":
     """Cumulative CPU time (user + system) of *this process* in seconds.
 
@@ -175,6 +257,10 @@ def __init__(
         # top-style "% of one core" reading is interpretable against the pod's
         # allocation.
         self._cpu_count = available_cpu_count()
+        # Container memory limit (cgroup cap), sampled once since it's static for
+        # the process's lifetime. Reported alongside rss_bytes so the monitor can
+        # show RSS as a percentage of the cap; None when uncapped.
+        self._mem_limit_bytes = _read_cgroup_memory_limit()
         # Previous CPU sample, so each ping can report utilization over the
         # interval since the last one rather than since process start.
         self._last_cpu_sec: float | None = None
@@ -237,15 +323,19 @@ async def _ping(self) -> None:
                 "task_limit": self.task_limit,
                 "in_flight": self._in_flight(),
                 "rss_bytes": _read_rss_bytes(),
+                "mem_limit_bytes": self._mem_limit_bytes,
                 "cpu_pct": self._cpu_pct(),
                 "cpu_count": self._cpu_count,
             }
         )
         try:
+            # No TTL: a persistent key is exempt from the broker's volatile-ttl
+            # eviction, so memory pressure can't shed a live worker's heartbeat.
+            # Liveness rides on the payload's last_seen instead; the monitor
+            # reaps keys left unrefreshed past HEARTBEAT_REAP_SEC.
             await broker_client.set(
                 heartbeat_key(self.stream, self.consumer),
                 payload,
-                ex=HEARTBEAT_TTL_SEC,
             )
         except Exception as e:
             self._logger.debug(f"Heartbeat ping failed: {e}")

shepherd_utils/reclaim.py

@@ -5,7 +5,10 @@
 stream's PEL and uses ``XCLAIM`` to take over messages whose owner is no
 longer alive so the work gets retried instead of dropped.
 
-Aliveness is determined by the heartbeat keys (TTL ~15s presence pings).
+Aliveness is determined by the heartbeat keys, which are persistent (no Redis
+TTL, so eviction can't shed them) and carry a ``last_seen`` timestamp refreshed
+every ~5s; a consumer counts as alive only while that timestamp is fresh (see
+``is_heartbeat_fresh``), so a stale leftover key doesn't shield a dead consumer.
 Two independent gates protect a busy-but-alive consumer from being robbed:
 
 1. **Heartbeat filter**: a message owned by a consumer with a live heartbeat
@@ -27,7 +30,7 @@
 
 from .broker import broker_client
 from .config import settings
-from .heartbeat import HEARTBEAT_PREFIX
+from .heartbeat import HEARTBEAT_PREFIX, is_heartbeat_fresh
 
 # Per-stream idle floor in seconds. A reclaim only triggers once a pending
 # message has been idle longer than this, which gates against yanking work
@@ -69,13 +72,27 @@ def min_idle_sec_for(stream: str, override: Optional[int] = None) -> int:
 
 
 async def _alive_consumers_on_stream(stream: str) -> set:
-    """Return the set of consumer names with a live heartbeat for ``stream``."""
+    """Return the set of consumer names with a *fresh* heartbeat for ``stream``.
+
+    Heartbeat keys are persistent, so key existence alone no longer proves
+    liveness -- a crashed consumer's key lingers until the monitor reaps it.
+    We read each payload and keep only consumers whose ``last_seen`` is fresh,
+    so a dead consumer's stale key can't wrongly shield its orphaned messages
+    from reclaim. This freshness check is intrinsic here (it reads last_seen
+    directly), so reclaim never depends on the monitor being up to reap.
+    """
     pattern = f"{HEARTBEAT_PREFIX}:{stream}:*"
-    alive: set = set()
+    keys: List[str] = []
     async for key in broker_client.scan_iter(match=pattern, count=200):
+        keys.append(key)
+    if not keys:
+        return set()
+    values = await broker_client.mget(keys)
+    alive: set = set()
+    for key, raw in zip(keys, values):
         # key format: worker:heartbeat:{stream}:{consumer}
         parts = key.split(":", 3)
-        if len(parts) == 4:
+        if len(parts) == 4 and is_heartbeat_fresh(raw):
             alive.add(parts[3])
     return alive