SEP-1859: Record per-entity sync freshness and failure state (#1436)

Adds four per-entity sync-health columns to the inventory entities and
writes them from the syncer's own per-entity boundary.

**Inventory service.** `SyncHealthBase` carries `last_synced_at`,
`last_sync_error`, `sync_failing_since` and `consecutive_failures`; it
is mixed into `Node`, `Service`, `Schema` and `Table` and into their
read responses, and deliberately not into the `*Write` models — the same
shape `RetiredAtBase` already has. One Alembic revision on the inventory
track adds the four columns to all four tables; `consecutive_failures`
is `NOT NULL` with a `0` server default so existing rows backfill, and
the default is kept for the rolling-upgrade window.

**A narrow write path.** `POST /{entity}/{id}/sync-health` on each of
the four routers takes a `SyncHealthWrite` carrying the *outcome* —
`success` or `failure`, an `error`, and the `attempted_at` the syncer
began with — rather than the column values, so a future data `PUT` never
has to carry sync bookkeeping. All four carry `IsServicePrincipalDep`:
the columns are written only by the syncer and are not
operator-editable. All four resolve the entity through the retirable
dep, so a retirement concurrent with the sync does not turn bookkeeping
into a failed sync item.

**Server-side transitions.** `SyncHealthManagerMixin.record_sync_health`
applies one atomic `UPDATE` per outcome. The increment and the
first-failure-only rule are expressed in SQL (`consecutive_failures +
1`, and a `CASE` keeping the earlier of the stored `sync_failing_since`
and this attempt), so no increment is lost to a read-modify-write.
`last_synced_at` is stamped with the attempt time, not the moment the
report arrived. The syncer reads that time from the clock directly
rather than through `utc_now`, whose second truncation would hand two
attempts within one second the same ordering key — and the guards admit
an equal one, so the later arrival would win whichever attempt was
actually newer.

Two ordering guards keep a late report from overwriting a newer one.
Both statements carry `last_synced_at IS NULL OR last_synced_at <=
:attempted_at`, so a stale failure cannot restart a run a newer success
closed and a stale success cannot pull the freshness backwards. The
success statement additionally carries `sync_failing_since IS NULL OR
sync_failing_since <= :attempted_at`: a failure deliberately never moves
`last_synced_at`, so the first guard is blind to one, and without the
second an older success arriving late would clear a run a newer attempt
had just opened — reporting a clean row whose latest attempt failed.

Two failures of one run are ordered by construction rather than by a
guard: `sync_failing_since` keeps the *earlier* of the stored value and
the reported attempt (a portable `CASE`, which subsumes the `COALESCE`
it replaces since the null row falls to the `ELSE`), so the run stays
opened at its true start whichever report lands first. `last_sync_error`
is the one field arrival order still decides; see Known limitations.

`attempted_at` is refused when it sits more than
`SYNC_ATTEMPT_MAX_CLOCK_SKEW` (5 minutes) ahead of the inventory
service's clock. The guards admit anything not older than the stored
attempt, so a reporter running fast would otherwise stamp a freshness
nothing later could supersede — and would lock itself out for the whole
interval once its clock was corrected. Refusing leaves the entity
looking stale, the direction the rest of this mechanism already errs in.

**Syncer side.** A new `app/sep/sync/health.py` holds the whole
reporting mechanism: the attempt marker, the genuine-attempt semantics,
the error-description contract and the best-effort POST. `BaseSyncer`
reaches it through a `sync_health` property and nests `record(...)`
*inside* `manage_sync_item` in all four `sync_*` methods, so a failure
is recorded before that boundary marks the SyncItem failed. A success is
recorded only once the block called `mark_compared()`, which is what
excludes the filtered-out `fetch_* -> None` early return — that return
leaves `manage_sync_item` on the same clean-exit path a real sync takes,
so the clean exit alone cannot be trusted. `hold_entity`, the four
`retire_*` methods and `sync_inventory` are not wrapped and write
nothing.

**Which levels a syncer owns** is policy, declared per class as a
`mirrors_entity_levels` ClassVar beside the existing
`reads_retired_entities`: `PMMSyncer` for Node and Service, the MySQL
syncer for Schema and Table, `SystemFactsSyncer` explicitly none. A
syncer that only traverses a level to reach its children confirms
nothing about that entity's mirrored values, and refreshing them there
would clear a failing PMM mirror into a false all-clear — the MySQL and
system-facts runs interleave with PMM's on the same Node and Service
rows.

A `SyncFailError` reaching the reporter is the one exception the block
does not attribute to its own entity. It can only come from a *nested*
level — this level's own boundary raises it after the reporter has
already exited — and the syncers walk to children from inside the
parent's `perform_*_sync`, so a child's failure passes through the
parent. Whether it does at all depends on `break_on_error`, since
otherwise the child's own boundary swallows it; attributing it upward
would report a node as failing whose own fields were just confirmed, and
would make the columns describe an identical outcome differently in the
two modes. The child records its failure on its own row.

`INVENTORY_PATH_SEGMENTS` moves verbatim from `app/sep/sync/models.py`
to a new leaf `app/sep/sync/constants.py` so `health.py` can build
entity paths without importing `models.py`, which imports `health.py`.

**What `last_sync_error` may hold.** The column is durable and readable
through the ordinary inventory read routes — which are
`IsAuthenticatedDep`, while the write requires the service principal —
so `_describe_sync_error` is an allowlist rather than a scrubber over
arbitrary text. An `HTTPException` contributes only its status code,
because its `detail` is built from the remote response body. Otherwise
the full message is kept only for the exception classes named in
`_MESSAGE_SAFE_ERRORS`, each of which interpolates nothing but sync
bookkeeping; everything else contributes its type name alone. Membership
is by *exact type* (`type(error) in _MESSAGE_SAFE_ERRORS`), not
`isinstance`, so a subclass added later has to be opted in deliberately
instead of inheriting persistence from its base — a subclass is free to
interpolate remote context the base never did.
`ExecutorHostNotFoundError` is the case that makes the distinction
load-bearing: it is a `SyncError`, it reaches this path from the
`fetch_schema` / `fetch_table` task-target lookup, and its message
carries the entire Tasks-API executor-host map. The type name always
leads, because several exceptions here stringify to `""` and an empty
error is refused by the write model. The exception itself is re-raised
to the boundary that logs it with a traceback.

### Bundled fixes

- `tests/app/inventory/migrations/test_mandatory_pmm_origin.py` compared
whole rows across an upgrade/downgrade boundary, so any column added by
a later revision broke it. The captured expectation is now restricted to
the columns the pre-origin schema declares; the values and the
pre-origin column set are still both asserted.
- `app/sep/sync/syncers/system_facts/syncer.py` lost two pre-existing
`:vartype` directives from the class docstring this change edits —
annotations are the source of truth, and the docstring gate treats a
directive inside an edited docstring as in scope.
- Three existing syncer tests asserted `post.assert_awaited_once()` /
`assert_not_awaited()` on a client that now also carries sync-health
writes. They assert on the entity POSTs specifically, via a shared
`entity_posts` helper, so the original claim ("exactly one create",
"nothing created or revived") is unchanged.

Author: yyyyyyyan

app/inventory/constants.py

@@ -15,6 +15,7 @@
 
 """Define constants for the Inventory service."""
 
+from datetime import timedelta
 from enum import StrEnum
 from typing import Final
 
@@ -40,6 +41,20 @@
     "System observation not collected yet for this service"
 )
 
+#: Longest error text stored on an entity's ``last_sync_error``. The manager
+#: truncates to this, so a caller may send an exception message of any length.
+SYNC_ERROR_MAX_LENGTH: Final = 1000
+
+#: How far ahead of this service's clock a reported ``attempted_at`` may sit
+#: before the report is refused. The ordering guards admit anything not older
+#: than the stored attempt, so a reporter whose clock runs fast stamps a
+#: freshness no later report can supersede until wall-clock time catches up —
+#: and one whose clock is then corrected is locked out for the whole interval.
+#: Refusing leaves the entity looking stale, which is the direction the rest of
+#: this mechanism already errs in. Wide enough to absorb the drift between two
+#: containers of one deployment without absorbing a misconfigured clock.
+SYNC_ATTEMPT_MAX_CLOCK_SKEW: Final = timedelta(minutes=5)
+
 
 class RetirableEntityName(StrEnum):
     """Name the inventory entity types that carry a retirement tombstone.

app/inventory/crud.py

@@ -21,7 +21,7 @@
 from typing import Any, ClassVar, Final, TYPE_CHECKING
 
 from fastapi import HTTPException
-from sqlalchemy import and_, func, literal, Update, update
+from sqlalchemy import and_, case, func, literal, Update, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import aliased, joinedload
 from sqlalchemy.sql import ColumnElement, ColumnExpressionArgument
@@ -38,7 +38,12 @@
 )
 from app.core.pagination import Pagination
 from app.core.utils.date_time import utc_now
-from app.inventory.constants import ACTIVE_RETIREMENT_KEY, RetirableEntityName
+from app.core.utils.strings import shorten_text
+from app.inventory.constants import (
+    ACTIVE_RETIREMENT_KEY,
+    RetirableEntityName,
+    SYNC_ERROR_MAX_LENGTH,
+)
 from app.inventory.models import (
     ExternalIdentityAlias,
     HostSystemObservation,
@@ -52,6 +57,9 @@
     Service,
     ServiceSystemObservation,
     SourceEnum,
+    SyncHealthBase,
+    SyncHealthWrite,
+    SyncOutcomeEnum,
     Table,
 )
 
@@ -94,6 +102,135 @@ def _revive(
     )
 
 
+#: Keeps the sync-health writes off SQLAlchemy's post-UPDATE session sync. Its
+#: "evaluate" strategy re-runs the guards below in Python against whatever the
+#: identity map holds, comparing a stored timestamp that is naive on an engine
+#: dropping the offset (SQLite) against a timezone-aware attempt time — which
+#: raises ``TypeError`` rather than falling through to a fetch. Sessions are
+#: built with ``expire_on_commit=False``, so a loaded instance stays stale for
+#: the rest of the request; no route reads one of these rows back after the
+#: write, and one that starts to must re-read rather than trust the instance.
+_UNSYNCHRONIZED: Final = {"synchronize_session": False}
+
+
+def _not_superseded(
+    model: type[SyncHealthBase], attempted_at: datetime
+) -> ColumnElement[bool]:
+    """Match only rows whose recorded sync is not newer than this attempt.
+
+    Reports cross the service boundary over HTTP, so arrival order is not
+    attempt order. Comparing against ``last_synced_at`` discards a report from
+    an attempt that had already been superseded by a completed later one.
+
+    :param model: The table being written.
+    :param attempted_at: When the reporting syncer began its attempt.
+    :return: The guard predicate.
+    """
+    return or_(
+        col(model.last_synced_at).is_(None),
+        col(model.last_synced_at) <= attempted_at,
+    )
+
+
+def _no_newer_failure(
+    model: type[SyncHealthBase], attempted_at: datetime
+) -> ColumnElement[bool]:
+    """Match only rows whose open failure run did not start after this attempt.
+
+    A failure never moves ``last_synced_at``, so :func:`_not_superseded` cannot
+    see one: an older success arriving late would otherwise clear a run a newer
+    attempt had just opened, reporting a clean row whose latest attempt failed.
+    ``sync_failing_since`` names the *earliest* failure of the run, so a success
+    landing between two failures of one run is still admitted — closing that
+    would take a column recording the newest attempt seen.
+
+    :param model: The table being written.
+    :param attempted_at: When the reporting syncer began its attempt.
+    :return: The guard predicate.
+    """
+    return or_(
+        col(model.sync_failing_since).is_(None),
+        col(model.sync_failing_since) <= attempted_at,
+    )
+
+
+def _record_sync_success(
+    model: type[SyncHealthBase],
+    *whereclause: ColumnExpressionArgument[bool],
+    synced_at: datetime,
+) -> Update:
+    """Build the UPDATE stamping a clean sync on every row matching the clauses.
+
+    ``synced_at`` is the syncer's attempt time, not the moment this statement
+    runs, so ``last_synced_at`` answers "when was this confirmed against its
+    source" rather than "when did the report arrive".
+
+    :param model: The table to record the success in.
+    :param whereclause: Clauses narrowing the rows to write.
+    :param synced_at: When the reporting syncer began its attempt.
+    :return: The UPDATE statement.
+    """
+    return (
+        update(model)
+        .where(
+            _not_superseded(model, synced_at),
+            _no_newer_failure(model, synced_at),
+            *whereclause,
+        )
+        .values(
+            last_synced_at=synced_at,
+            last_sync_error=None,
+            sync_failing_since=None,
+            consecutive_failures=0,
+        )
+        .execution_options(**_UNSYNCHRONIZED)
+    )
+
+
+def _record_sync_failure(
+    model: type[SyncHealthBase],
+    *whereclause: ColumnExpressionArgument[bool],
+    error: str,
+    failed_at: datetime,
+) -> Update:
+    """Build the UPDATE recording a failed sync on every row matching the clauses.
+
+    ``sync_failing_since`` keeps the earlier of the stored value and this
+    attempt rather than being assigned, so it names the *first* failure after
+    the last success even when two failures of one run arrive out of order —
+    coalescing alone would leave it on whichever landed first. The counter is
+    incremented in SQL so concurrent runs cannot lose an increment to a
+    read-modify-write. ``last_sync_error`` is still assigned unconditionally,
+    so an out-of-order pair leaves the older message there; naming the newest
+    failure would take a column recording the newest attempt seen, which the
+    entity does not carry. ``last_synced_at`` is deliberately absent: a failure
+    never moves it — which is also why the guard compares against it rather
+    than being skipped here.
+
+    :param model: The table to record the failure in.
+    :param whereclause: Clauses narrowing the rows to write.
+    :param error: The bounded description to store.
+    :param failed_at: When the reporting syncer began its attempt.
+    :return: The UPDATE statement.
+    """
+    return (
+        update(model)
+        .where(_not_superseded(model, failed_at), *whereclause)
+        .values(
+            last_sync_error=error,
+            consecutive_failures=col(model.consecutive_failures) + 1,
+            sync_failing_since=case(
+                (
+                    col(model.sync_failing_since) < failed_at,
+                    col(model.sync_failing_since),
+                ),
+                else_=failed_at,
+            ),
+        )
+        .execution_options(**_UNSYNCHRONIZED)
+    )
+
+
 def _retained(
     model: type[RetirableSQLModel],
     retired_before: datetime,
@@ -154,6 +291,55 @@ def _retained_descendant_exists(
     )
 
 
+class SyncHealthManagerMixin(BaseSQLModelManager):
+    """Record the outcome of one syncer attempt on an entity."""
+
+    @classmethod
+    async def record_sync_health(
+        cls,
+        session: AsyncSession,
+        instance: RetirableSQLModel,
+        outcome: SyncHealthWrite,
+    ) -> None:
+        """Apply one sync outcome to an entity's four sync-health columns.
+
+        The statement is hand-built rather than routed through ``update``, for
+        the reason :meth:`RetirableManagerMixin.retire`'s is: the transitions
+        are expressed in SQL so an increment cannot be lost to a
+        read-modify-write, and the write must reach a row the manager's own
+        retired filter would hide.
+
+        :param session: The asynchronous database session to use.
+        :param instance: The entity the outcome was observed for. Typed as the
+            retirable base ``retire`` takes rather than as ``SyncHealthBase``,
+            which carries the columns but not the ``id`` this addresses the row
+            by; ``cls.Model`` is what confines the write to a level that has
+            them.
+        :param outcome: What the syncer reported.
+        :raises ValueError: If the outcome names no branch here, which an
+            outcome added to :class:`SyncOutcomeEnum` without a transition
+            would. Failing loudly beats routing it to the failure branch, where
+            the body model does not require an ``error``.
+        """
+        if outcome.outcome is SyncOutcomeEnum.SUCCESS:
+            statement = _record_sync_success(
+                cls.Model,
+                col(cls.Model.id) == instance.id,
+                synced_at=outcome.attempted_at,
+            )
+        elif outcome.error is not None:
+            statement = _record_sync_failure(
+                cls.Model,
+                col(cls.Model.id) == instance.id,
+                error=shorten_text(outcome.error, SYNC_ERROR_MAX_LENGTH),
+                failed_at=outcome.attempted_at,
+            )
+        else:
+            raise ValueError(f"No sync-health transition for {outcome.outcome!r}")
+        await cls._exec(session, statement)  # call-shape-dup-ok: the manager idiom
+        await session.commit()
+
+
 class RetirableManagerMixin(BaseSQLModelManager):
     """Confine an entity's reads to the tombstone policy.
 
@@ -1411,7 +1597,7 @@ async def unlink_identity(
         await cls._commit_identity_change(session, statements, rows)
 
 
-class NodeManager(AliasableManagerMixin, BaseSQLModelManager):
+class NodeManager(AliasableManagerMixin, SyncHealthManagerMixin, BaseSQLModelManager):
     """Manage Node operations, including retrieval, listing, and retirement.
 
     :ivar Model: The SQLModel class this manager is responsible for (``Node``).
@@ -1479,7 +1665,9 @@ async def _identity_source(
         return entity.source
 
 
-class ServiceManager(AliasableManagerMixin, BaseSQLModelChildManager):
+class ServiceManager(
+    AliasableManagerMixin, SyncHealthManagerMixin, BaseSQLModelChildManager
+):
     """Manage Service operations, including retrieval, listing, and retirement.
 
     :ivar Model: The SQLModel class this manager is responsible for (``Service``).
@@ -1635,7 +1823,9 @@ async def _require_revivable(
             )
 
 
-class SchemaManager(RetirableManagerMixin, BaseSQLModelChildManager):
+class SchemaManager(
+    RetirableManagerMixin, SyncHealthManagerMixin, BaseSQLModelChildManager
+):
     """Manage Schema operations, including retrieval, listing, and retirement.
 
     :ivar Model: The SQLModel class this manager is responsible for (`Schema`).
@@ -1664,7 +1854,9 @@ class SchemaManager(RetirableManagerMixin, BaseSQLModelChildManager):
     )
 
 
-class TableManager(RetirableManagerMixin, BaseSQLModelChildManager):
+class TableManager(
+    RetirableManagerMixin, SyncHealthManagerMixin, BaseSQLModelChildManager
+):
     """Manage Table operations, including retrieval, listing, and retirement.
 
     :ivar Model: The SQLModel class this manager is responsible for (`Table`).

app/inventory/migrations/versions/2026_08_31_2200-9f2c14d6b8a7_add_sync_health_columns_to_inventory_.py

@@ -0,0 +1,90 @@
+# Copyright (C) 2026 Percona LLC
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""add sync health columns to inventory entities
+
+Revision ID: 9f2c14d6b8a7
+Revises: 3c39abf7a429
+Create Date: 2026-08-31 22:00:00.000000
+
+Add ``last_synced_at``, ``last_sync_error``, ``sync_failing_since`` and
+``consecutive_failures`` to the four inventory entities, so a row records how
+recently and how successfully the syncer that mirrors it last confirmed it.
+
+The three timestamps and the error text are nullable, which is the correct
+reading for a row nothing has reported on yet. ``consecutive_failures`` is NOT
+NULL with a ``0`` server default so existing rows land on "not failing" rather
+than on an ambiguous NULL; the default is kept rather than dropped, exactly as
+``retirement_key``'s is in ``c7d1e94ab3f2``, so a release still running the
+previous code can insert without the column.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "9f2c14d6b8a7"
+down_revision = "3c39abf7a429"
+branch_labels = None
+depends_on = None
+
+#: Every entity a syncer mirrors, and therefore every table carrying the
+#: per-entity sync-health columns.
+_SYNCABLE_TABLES = ("node", "service", "schema", "table")
+
+#: The columns added to each table, in the order they are added.
+_SYNC_HEALTH_COLUMN_NAMES = (
+    "last_synced_at",
+    "last_sync_error",
+    "sync_failing_since",
+    "consecutive_failures",
+)
+
+
+def upgrade() -> None:
+    """Add the four sync-health columns to every syncable table.
+
+    The columns are built inline per table rather than hoisted into a shared
+    tuple: a ``Column`` binds to the first table it is added to.
+    """
+    for table_name in _SYNCABLE_TABLES:
+        op.add_column(
+            table_name,
+            sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True),
+        )
+        op.add_column(
+            table_name,
+            sa.Column("last_sync_error", sa.Text(), nullable=True),
+        )
+        op.add_column(
+            table_name,
+            sa.Column("sync_failing_since", sa.DateTime(timezone=True), nullable=True),
+        )
+        op.add_column(
+            table_name,
+            sa.Column(
+                "consecutive_failures",
+                sa.Integer(),
+                nullable=False,
+                server_default=sa.text("0"),
+            ),
+        )
+
+
+def downgrade() -> None:
+    """Drop the sync-health columns, reversing the order they were added in."""
+    for table_name in reversed(_SYNCABLE_TABLES):
+        for column_name in reversed(_SYNC_HEALTH_COLUMN_NAMES):
+            op.drop_column(table_name, column_name)