From 423493e2bf39f971e9924b2583f82b8854acd736 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Thu, 15 Oct 2026 10:08:38 -0400
Subject: [PATCH] fix(docker): a restarted agent's re-sent log lines are stored
 once (#5809)

An agent keeps its log cursors in memory, so after a restart it seeds each
container from its last 50 lines, which Steward already holds; docker_logs
has no unique key, so every update or reboot doubled them. The ingest now
reads what it already has for each container over the batch's own span,
in one query, and skips those lines. A push of new lines overlaps nothing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 plugins/docker/ingest.py         | 28 ++++++++++++++++--
 tests/integration/test_docker.py | 49 ++++++++++++++++++++++++++++++++
 2 files changed, 75 insertions(+), 2 deletions(-)

diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py
index 9efc563..027ce95 100644
--- a/plugins/docker/ingest.py
+++ b/plugins/docker/ingest.py
@@ -147,12 +147,36 @@ async def _persist_logs(session, host, log_batches) -> None:
     if not await get_setting(session, "docker.logs.enabled"):
         return
     exclude = set(await get_setting(session, "docker.logs.exclude") or ())
-    for row in _log_rows(log_batches, host.id):
-        if row["container_name"] in exclude:
+    rows = [r for r in _log_rows(log_batches, host.id) if r["container_name"] not in exclude]
+    stored = await _stored_lines(session, host.id, rows)
+    for row in rows:
+        if (row["container_name"], row["ts"], row["stream"], row["line"]) in stored:
             continue
         session.add(DockerLog(**row))
 
 
+async def _stored_lines(session, host_id: str, rows: list) -> set:
+    """The lines of `rows` this host already has, as (container, ts, stream,
+    line). A restarted agent has lost its cursors and re-sends each
+    container's last lines (#5809); a push of new lines overlaps nothing, so
+    the one query, bounded per container to the batch's own span, finds none."""
+    from sqlalchemy import and_, or_, select
+    from .models import DockerLog
+
+    spans: dict[str, list] = {}
+    for r in rows:
+        span = spans.setdefault(r["container_name"], [r["ts"], r["ts"]])
+        span[0], span[1] = min(span[0], r["ts"]), max(span[1], r["ts"])
+    if not spans:
+        return set()
+    found = await session.execute(
+        select(DockerLog.container_name, DockerLog.ts, DockerLog.stream, DockerLog.line)
+        .where(DockerLog.host_id == host_id, or_(*(
+            and_(DockerLog.container_name == name, DockerLog.ts.between(lo, hi))
+            for name, (lo, hi) in spans.items()))))
+    return {tuple(r) for r in found}
+
+
 async def _persist_swarm(session, host, swarm: dict) -> None:
     """Upsert this manager's swarm topology; drop rows no longer reported.
 
diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py
index 80abc77..909efa5 100644
--- a/tests/integration/test_docker.py
+++ b/tests/integration/test_docker.py
@@ -221,6 +221,55 @@ def test_persist_logs_stores_lines(app):
     ]
 
 
+@_NEEDS_DB
+def test_persist_logs_skips_lines_a_restarted_agent_sends_again(app):
+    """#5809: a restarted agent has lost its cursors and re-sends each
+    container's last lines. Those are stored once; a new line, or another line
+    at the same instant, is still kept."""
+    from types import SimpleNamespace
+    from sqlalchemy import text
+    from steward.models.hosts import Host
+
+    persist = _persist_fn(app)
+    now = datetime.now(timezone.utc)
+
+    def line(container, at, text_, stream="stdout"):
+        return {"container": container, "stream": stream,
+                "ts": f"2026-01-01T00:00:0{at}+00:00", "line": text_}
+
+    first = [(now, [line("web", 1, "started"), line("web", 2, "listening"),
+                    line("db", 1, "ready")])]
+    again = [(now, [line("web", 1, "started"), line("web", 2, "listening"),
+                    line("web", 2, "worker 2 up"), line("web", 3, "GET /"),
+                    line("db", 1, "ready"), line("db", 1, "ready", stream="stderr")])]
+
+    hid = str(uuid.uuid4())
+    host = SimpleNamespace(id=hid, name="loghost-again")  # as below: no expiry across commits
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(Host(id=hid, name="loghost-again", address="10.0.0.19"))
+                await s.flush()
+                await persist(s, host, [], None, None, first)
+            async with s.begin():
+                await persist(s, host, [], None, None, again)
+            rows = (await s.execute(text(
+                "SELECT container_name, stream, line FROM docker_logs "
+                "WHERE host_id = :h ORDER BY container_name, ts, stream, line"),
+                {"h": hid})).all()
+            return [tuple(r) for r in rows]
+
+    assert asyncio.run(_go()) == [
+        ("db", "stderr", "ready"),
+        ("db", "stdout", "ready"),
+        ("web", "stdout", "started"),
+        ("web", "stdout", "listening"),
+        ("web", "stdout", "worker 2 up"),
+        ("web", "stdout", "GET /"),
+    ]
+
+
 @_NEEDS_DB
 def test_persist_logs_respects_toggle_and_exclude(app):
     """Server-side controls (Settings): an excluded container's lines are dropped