fix(host_agent): an ingest writes the newest five samples, so a full backlog can land (#5149)

Measured on dev (f802c9f): every dropped ingest carried samples=21,
metric_rows≈4,200 and never finished the INSERT flush inside the agent's
10 s ("completed: parse=0.1s"). The rollback left the agent's ring buffer
full, so each later post was the same size and failed the same way: no
host's metrics have landed since.

An ingest now keeps the newest MAX_SAMPLES_PER_INGEST (5) samples by ts,
logs how many it left out, and returns samples_dropped. The older samples
are the ones the ring buffer would have overwritten anyway. Why the
database writes so few rows a second is not yet known; the slow-ingest
line now measures it on a post that finishes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/host_agent/routes.py

@@ -191,6 +191,11 @@ async def ingest():
 
     token_hash = _hash_token(raw)
     phases = _Phases()
+    samples, dropped = newest_samples(samples, MAX_SAMPLES_PER_INGEST)
+    if dropped:
+        current_app.logger.warning(
+            "host_agent ingest: kept the newest %d of %d samples for token=%s…",
+            len(samples), len(samples) + dropped, token_hash[:8])
     try:
         async with current_app.db_sessionmaker() as session:
             reg = (await session.execute(
@@ -326,13 +331,31 @@ async def ingest():
     phases.report(host.name, accepted, len(metric_rows),
                   sum(len(c) for _, c in docker_snapshots),
                   sum(len(lines) for _, lines in docker_log_batches))
-    return jsonify({"ok": True, "samples_accepted": accepted}), 200
+    return jsonify({"ok": True, "samples_accepted": accepted, "samples_dropped": dropped}), 200
 
 
 # The agent gives up on a post after 10 s (agent.py); an ingest past this
 # says where its time went, so a slow phase shows in the logs (#5149).
 SLOW_INGEST_SECONDS = 2.0
 
+# How many samples one ingest writes (#5149). An agent that could not post
+# sends its whole buffer (up to 21 samples, ~4,000 metric rows); on the live
+# instance that did not finish inside the agent's 10 s, the rollback left the
+# buffer full, and every later post failed the same way. The newest samples
+# are kept: the older ones would be lost to the agent's ring buffer anyway.
+MAX_SAMPLES_PER_INGEST = 5
+
+
+def newest_samples(samples: list, limit: int) -> tuple[list, int]:
+    """(the newest `limit` samples, oldest first; how many were left out).
+
+    Ordered by their ISO-8601 `ts` (the agent writes them all in one form);
+    one without a ts sorts oldest and is rejected later as malformed."""
+    if len(samples) <= limit:
+        return samples, 0
+    ordered = sorted(samples, key=lambda s: str(s.get("ts") or "") if isinstance(s, dict) else "")
+    return ordered[-limit:], len(samples) - limit
+
 
 class _Phases:
     """Wall-clock per phase of one ingest request."""