fix(syslog): a device through Docker's address stays one source when that address changes (#6717)

Docker's ingress address depends on the node Steward's task is on, and sources
were keyed by (address, sender). After a task move the same device arrived under
a new key and became a new "offered" row, so its lines were counted, not kept.

A named sender is now one row whatever address it last came from, enforced by a
partial unique index. Migration syslog_005 folds rows sharing a name into the
added one and moves their lines and patterns across.

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

plugins/syslog/migrations/versions/syslog_005_one_row_per_sender.py

@@ -0,0 +1,64 @@
+"""Syslog: one row per name at Docker's address, whichever address it last came
+from. Rows sharing a name fold into the added one, or else the newest.
+
+Revision ID: syslog_005_one_row_per_sender
+Revises: syslog_004_device_address
+Create Date: 2026-10-30
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "syslog_005_one_row_per_sender"
+down_revision: Union[str, None] = "syslog_004_device_address"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_STATE_FIRST = {"accepted": 0, "ignored": 1, "offered": 2}
+
+
+def _fold(bind, sender: str) -> None:
+    rows = bind.execute(sa.text(
+        "SELECT id, address, state, device_address, host_id, hostname_seen, first_seen_at, last_seen_at, "
+        "lines_total, last_severity, last_line FROM syslog_sources WHERE sender = :sender"
+    ), {"sender": sender}).mappings().all()
+    keep = min(rows, key=lambda r: (_STATE_FIRST.get(r["state"], 3), -r["last_seen_at"].timestamp()))
+    newest = max(rows, key=lambda r: r["last_seen_at"])
+    others = [r["id"] for r in rows if r["id"] != keep["id"]]
+    in_others = sa.bindparam("others", expanding=True)
+    for table in ("syslog_lines", "syslog_patterns"):
+        bind.execute(sa.text(f"UPDATE {table} SET source_id = :keep WHERE source_id IN :others").bindparams(in_others),
+                     {"keep": keep["id"], "others": others})
+    bind.execute(sa.text("DELETE FROM syslog_sources WHERE id IN :others").bindparams(in_others), {"others": others})
+    bind.execute(sa.text(
+        "UPDATE syslog_sources SET address = :address, device_address = :device_address, host_id = :host_id, "
+        "hostname_seen = :hostname_seen, first_seen_at = :first_seen_at, last_seen_at = :last_seen_at, "
+        "lines_total = :lines_total, last_severity = :last_severity, last_line = :last_line WHERE id = :id"
+    ), {
+        "id": keep["id"],
+        "address": newest["address"],
+        "device_address": next((r["device_address"] for r in rows if r["device_address"]), ""),
+        "host_id": keep["host_id"] or next((r["host_id"] for r in rows if r["host_id"]), None),
+        "hostname_seen": newest["hostname_seen"],
+        "first_seen_at": min(r["first_seen_at"] for r in rows),
+        "last_seen_at": newest["last_seen_at"],
+        "lines_total": sum(r["lines_total"] or 0 for r in rows),
+        "last_severity": newest["last_severity"],
+        "last_line": newest["last_line"],
+    })
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    senders = bind.execute(sa.text(
+        "SELECT sender FROM syslog_sources WHERE sender <> '' GROUP BY sender HAVING count(*) > 1"
+    )).scalars().all()
+    for sender in senders:
+        _fold(bind, sender)
+    op.create_index("uq_syslog_sources_sender", "syslog_sources", ["sender"], unique=True,
+                    postgresql_where=sa.text("sender <> ''"))
+
+
+def downgrade() -> None:
+    op.drop_index("uq_syslog_sources_sender", table_name="syslog_sources")

plugins/syslog/models.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 from datetime import datetime, timezone
 
 from sqlalchemy import (JSON, BigInteger, Boolean, DateTime, ForeignKey, Index, SmallInteger, String, Text,
-                        UniqueConstraint)
+                        UniqueConstraint, text)
 from sqlalchemy.orm import Mapped, mapped_column
 
 from steward.core.ids import time_ordered_id
@@ -17,6 +17,8 @@ class SyslogSource(Base):
     Through a port Swarm publishes on its ingress (the default), every device
     arrives from one address of Docker's own; there a row is a device by the
     name its lines carry, `sender` (#6096). Everywhere else `sender` is "".
+    That address changes when Steward's task changes node, so a name has one
+    row and `address` follows it.
 
     A sender Steward hasn't been told about is a row too, `state="offered"`:
     its lines are counted and its newest kept as a sample, so the page can
@@ -25,7 +27,10 @@ class SyslogSource(Base):
     on; `"ignored"` stops the offer and keeps counting.
     """
     __tablename__ = "syslog_sources"
-    __table_args__ = (UniqueConstraint("address", "sender", name="uq_syslog_sources_address_sender"),)
+    __table_args__ = (
+        UniqueConstraint("address", "sender", name="uq_syslog_sources_address_sender"),
+        Index("uq_syslog_sources_sender", "sender", unique=True, postgresql_where=text("sender <> ''")),
+    )
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id)
     address: Mapped[str] = mapped_column(String(64), nullable=False)

plugins/syslog/receiver.py

@@ -23,7 +23,7 @@ from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 
 from steward.core.scheduler import ScheduledTask
 
@@ -246,21 +246,36 @@ async def store_lines(session, lines: list[Received]) -> None:
     for line in lines:
         by_sender.setdefault(sender_of(line), []).append(line)
     addresses = list({address for address, _ in by_sender})
-    sources = {(s.address, s.sender): s for s in (await session.execute(
-        select(SyslogSource).where(SyslogSource.address.in_(addresses)))).scalars()}
-    new = [k for k in by_sender if k not in sources]
-    for key in [k for k in new if k[1]]:
-        # A row from before devices at Docker's address were told apart by
-        # name, last seen under this one, is this device: it keeps its state
-        # and name rather than a new row being offered beside it.
-        before = sources.get((key[0], ""))
-        if before is not None and before.hostname_seen == key[1]:
-            before.sender = key[1]
-            sources[key] = sources.pop((key[0], ""))
-            new.remove(key)
+    names = list({name for _, name in by_sender if name})
+    rows = (await session.execute(select(SyslogSource).where(
+        or_(SyslogSource.address.in_(addresses), SyslogSource.sender.in_(names))))).scalars().all()
+    sources = {(s.address, s.sender): s for s in rows}
+    by_name = {s.sender: s for s in rows if s.sender}
+    new = []
+    for key in by_sender:
+        address, name = key
+        if name in by_name:
+            # Docker's address depends on the node Steward's task is on, so a
+            # device is its name wherever it now arrives from.
+            by_name[name].address = address
+            sources[key] = by_name[name]
+        elif key not in sources:
+            # A row from before devices at Docker's address were told apart by
+            # name, last seen under this one, is this device: it keeps its state
+            # and name rather than a new row being offered beside it.
+            before = sources.get((address, ""))
+            if name and before is not None and before.hostname_seen == name:
+                before.sender = name
+                by_name[name] = sources[key] = sources.pop((address, ""))
+            else:
+                new.append(key)
     if new:
         known = (await session.execute(select(func.count()).select_from(SyslogSource))).scalar_one()
         for key in new:
+            if key[1] in by_name:         # the same name from a second address in this round
+                by_name[key[1]].address = key[0]
+                sources[key] = by_name[key[1]]
+                continue
             if known >= MAX_SENDERS:
                 LISTENER.unoffered += len(by_sender.pop(key))
                 continue
@@ -268,6 +283,8 @@ async def store_lines(session, lines: list[Received]) -> None:
             source = SyslogSource(address=key[0], sender=key[1], first_seen_at=first.at, last_seen_at=first.at)
             session.add(source)
             sources[key] = source
+            if key[1]:
+                by_name[key[1]] = source
             known += 1
         await session.flush()