fix(node-agent): batch rsync, backlog trim, timeout 120s, backlog warn

Root cause of fleet staleness since ~OOM 2026-06-01: events/<node>/
grew to 291k files (1.2G on piha); rsync of the whole dir exceeded the
30s subprocess timeout every cycle; --remove-source-files never ran;
backlog compounded silently.

Four fixes:

1. Batch shipping (SHIP_BATCH_SIZE=1000): rsync sends only the oldest
   1000 files per cycle via --files-from=- on stdin instead of the whole
   directory.  Backlog drains across cycles; each push fits within timeout.

2. Timeout 30s → 120s: wider margin for large batches and slow links
   (piha → vps over Tailscale).

3. Backlog trim safety-net (_trim_events_backlog): if events/<node>/ exceeds
   BACKLOG_MAX (5000) files, oldest files are deleted to bring count back to
   5000.  Called each cycle before shipping.  Breaks the death-spiral
   independently of rsync success.  VPS excluded (uses _cleanup_control_plane_fs).

4. Backlog visibility: WARNING log with file count when unsent events
   exceed BACKLOG_WARN (2000).  "events backlog: N unsent files" — no more
   silent accumulation.

SHIP_BATCH_SIZE is env-configurable for tuning per-node.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

services/node-agent/src/node_agent.py

@@ -94,6 +94,12 @@ LAST_CLEANUP_FILE = STATE_DIR / "last-docker-cleanup"
 # How long to wait between full health-check cycles
 HEALTH_CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60"))
 
+# Event shipping: max files per rsync batch so each push fits within timeout
+SHIP_BATCH_SIZE = int(os.getenv("SHIP_BATCH_SIZE", "1000"))
+# Warn when unsent backlog exceeds this; hard-trim when exceeding BACKLOG_MAX
+BACKLOG_WARN = 2000
+BACKLOG_MAX  = 5000
+
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -524,42 +530,117 @@ class NodeAgent:
             logger.debug("No observer checkpoint present; skipping event cleanup")
 
     # ------------------------------------------------------------------
-    # Optional: ship events to VPS via rsync
+    # Event backlog safety-net
     # ------------------------------------------------------------------
 
-    def _ship_events_to_vps(self):
+    def _trim_events_backlog(self):
-        """
+        """Remove oldest local events when backlog exceeds BACKLOG_MAX.
-        Rsync local events to VPS so the observer can process them.
+
-        Requires:
+        Guards against the rsync-death-spiral: if shipping fails for an
-          - VPS_EVENTS_HOST env var set to the VPS hostname/IP
+        extended period, local events accumulate without bound.  When the
-          - SSH key accessible inside the container (mount via docker-compose)
+        directory exceeds BACKLOG_MAX files, the oldest (sorted ascending
-          - The node is NOT VPS itself
+        by filename, which encodes a Unix timestamp) are deleted first —
+        week-old health-check JSON is worthless; keeping the queue bounded
+        is more important.  Runs only on shipping nodes (VPS_EVENTS_HOST
+        set); VPS cleans its own events via _cleanup_control_plane_fs.
         """
         if not VPS_EVENTS_HOST or self.node_name == VPS_NODE_NAME:
             return
 
-        local_dir  = str(self._node_events_dir()) + "/"
+        events_dir = self._node_events_dir()
+        try:
+            files = sorted(events_dir.glob("*.json"))
+            count = len(files)
+        except Exception as exc:
+            logger.error(f"backlog trim: failed to scan {events_dir}: {exc}")
+            return
+
+        if count <= BACKLOG_MAX:
+            return
+
+        to_delete = files[:count - BACKLOG_MAX]  # oldest first (path sorts by ts)
+        deleted = 0
+        for f in to_delete:
+            try:
+                f.unlink()
+                deleted += 1
+            except Exception as exc:
+                logger.error(f"backlog trim: failed to delete {f.name}: {exc}")
+
+        logger.warning(
+            f"backlog trim: deleted {deleted} oldest events "
+            f"(had {count}, trimmed to {BACKLOG_MAX})"
+        )
+
+    # ------------------------------------------------------------------
+    # Optional: ship events to VPS via rsync
+    # ------------------------------------------------------------------
+
+    def _ship_events_to_vps(self):
+        """Rsync a batch of local events to VPS.
+
+        Sends at most SHIP_BATCH_SIZE (default 1000) files per cycle so
+        each push completes within the subprocess timeout even when a large
+        backlog has accumulated.  Files are selected oldest-first (sorted
+        by filename, which encodes a Unix timestamp); the observer processes
+        them in order via its per-node checkpoint.
+
+        Requires VPS_EVENTS_HOST set and an SSH key accessible inside the
+        container.  Does nothing on VPS itself.
+        """
+        if not VPS_EVENTS_HOST or self.node_name == VPS_NODE_NAME:
+            return
+
+        events_dir = self._node_events_dir()
+        try:
+            all_files = sorted(events_dir.glob("*.json"))
+        except Exception as exc:
+            logger.error(f"Failed to list events dir {events_dir}: {exc}")
+            return
+
+        total = len(all_files)
+        if total == 0:
+            return
+        if total > BACKLOG_WARN:
+            logger.warning(f"events backlog: {total} unsent files in {events_dir}")
+
+        batch = all_files[:SHIP_BATCH_SIZE]
+        # Pass filenames via stdin with --files-from=- to avoid ARG_MAX limits
+        # and to ship only the selected batch (not the whole directory).
+        files_input = "\n".join(f.name for f in batch)
+
         remote_dir = (f"{VPS_EVENTS_USER}@{VPS_EVENTS_HOST}:"
                       f"{VPS_EVENTS_PATH}/{self.node_name}/")
         cmd = [
             "rsync", "-az", "--remove-source-files",
+            "--files-from=-",
             # -F /dev/null: skip ~/.ssh/config entirely.  The .ssh dir is
-            # mounted from the host oskar user into the container which runs
+            # mounted from the host user into the container; OpenSSH rejects
-            # as root; OpenSSH rejects config files owned by a different UID.
+            # config files owned by a different UID.
             # UserKnownHostsFile=/dev/null pairs with StrictHostKeyChecking=no
-            # so we never try to write a known_hosts inside a read-only mount.
+            # so we never write a known_hosts inside a read-only mount.
             "-e", ("ssh -F /dev/null"
                    " -o StrictHostKeyChecking=no"
                    " -o UserKnownHostsFile=/dev/null"
                    " -o ConnectTimeout=10"
                    " -o BatchMode=yes"),
-            local_dir,
+            str(events_dir) + "/",
             remote_dir,
         ]
         try:
-            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+            result = subprocess.run(
+                cmd,
+                input=files_input,
+                capture_output=True,
+                text=True,
+                timeout=120,
+            )
             if result.returncode == 0:
-                logger.debug(f"Events shipped to {remote_dir}")
+                remaining = total - len(batch)
+                logger.debug(
+                    f"Shipped {len(batch)}/{total} events to {remote_dir}"
+                    + (f" ({remaining} remaining)" if remaining else "")
+                )
             else:
                 logger.warning(f"Event shipping failed: {result.stderr.strip()}")
         except Exception as exc:
@@ -639,6 +720,7 @@ class NodeAgent:
             {"disk_pct": disk_pct, "mem_pct": mem_pct, "cpu_pct": cpu_pct},
         )
 
+        self._trim_events_backlog()
         self._ship_events_to_vps()
 
     def loop(self, interval: int = HEALTH_CHECK_INTERVAL):