SEP-1894: Reconcile a sync run abandoned with no items left active (#1454)

## Summary

- A worker killed between its last `SyncItem` write and
`BaseSyncer.__aexit__` leaves a `SyncInstance` stuck at `RUNNING` with
zero `PENDING`/`RUNNING` items, and nothing reconciles it: the stale-run
reclaim only ran once `create()` had already detected an item conflict,
and its candidate query was keyed on non-terminal items — so an
all-terminal run was invisible to both halves. The inventory sync-status
endpoint kept reporting that run as running until ten later runs aged it
out of the window.
- `reclaim_stale_runs()` gains a second candidate class: a `RUNNING` run
whose items have all gone terminal. It is still grouped over `SyncItem`
rows, so a run that wrote no item at all — which carries no activity to
measure — stays out of scope, and `NOT IN` against the existing
in-progress subquery keeps the two candidate sets disjoint. `create()`
now calls the reclaim whether or not an item conflict exists, and reads
the conflicting items once, after the reclaim — reclaim → check on both
paths.
- The fence `UPDATE` re-asserts item staleness alongside the instance
status. The candidate query commits separately from the fence, so a
worker that resumed and either touched an existing item or opened a new
one is progressing after all and must not be failed retroactively.

`snapshot_complete` is deliberately left `NULL` on a reclaim, so a
partially applied run is never counted as a complete generation.

### Review round (`81582d60b`)

- **The pre-reclaim conflict read was dead on the reclaim path.** It was
assigned before the reclaim and then unconditionally overwritten
whenever `stale_after` was set, and `reclaim_stale_runs` consumes none
of it, so on the primary path that `SELECT` fed no later statement. The
read now happens once, below the reclaim block, which leaves both paths
behaving identically and drops a round trip per run start.
`test_create_refuses_a_run_that_resumed_during_the_reclaim` still fails
if the read moves back above the reclaim.
- **The fence's staleness re-assertion reused the candidate subquery
bare**, so it compiled to a `GROUP BY` over the whole `syncitem` table
with no syncer or candidate predicate. It now repeats the
`in_(stale_instance_ids)` narrowing, which lands as a `WHERE` before the
`GROUP BY` and makes the subquery index-seekable. The semantics are
unchanged because every group is one instance's items, so removing other
instances' rows cannot move a surviving group's `max()` — and the two
spare-conditions
(`test_reclaim_spares_a_run_whose_worker_touched_an_item`,
`test_reclaim_spares_an_idle_run_that_started_a_new_item`) still fail
when the predicate is dropped altogether.

### Risks

- **A live long-running run can now be fenced.** A run that writes its
top-level item terminal early and then keeps working has no heartbeat at
all, so once its last item write is older than `stale_after` this path
will reclaim it. Bounded by the documented contract on `stale_run_after`
("must exceed the longest expected runtime of a sync", shipped as
3600s), and it fails safe: `is_still_owned()` gates every retirement and
`finalize_run()` is guarded on `status == RUNNING`, so a wrongly-fenced
run stops retiring and leaves `snapshot_complete` `NULL` rather than
advancing the missing-grace counter. Heartbeating
`SyncInstance.updated_at` during a run would remove the exposure and is
out of scope here.
- **No new blocking mode.** The reclaim only ever moves a row from
`PENDING`/`RUNNING` to `FAILED` and releases items, and `create()` still
gates refusal on items, so it cannot start refusing runs it accepts
today. This is also why the follow-up advisory-lock ticket must not
replace the item-based check with a `SyncInstance.status` constraint —
and why the `create()` sequence is worth a look in review, since that
ticket will wrap it in a lock.
- **Clock sources differ**, pre-existing and inherited unchanged:
`updated_at` is written by `onupdate=func.now()` (server clock) while
the `having` compares against Python `utc_now()`. Not introduced here
and deliberately not fixed here.

The existing candidate query has to stay item-keyed rather than
instance-status-keyed: a reclaim interrupted between its fence and its
item release leaves an already-`FAILED` instance whose items are still
held, and only an item-keyed set still finds it to finish the release.

## Tested

- [ ] Plant the drift row against the dev DB: a `SyncInstance` for the
PMM syncer with `status='running'` plus one `SyncItem` with
`status='success'` and `updated_at` backdated beyond `STALE_RUN_AFTER`.
- [ ] `GET /api/inventory/sync-status` reports that run as `running` in
`last_runs`.
- [ ] Trigger an inventory sync, then re-read `GET
/api/inventory/sync-status`: the planted run now reads `failed` with
`snapshot_complete` null, and the new run is present and healthy.
- [ ] Add a second, recent `SyncItem` to the planted run and confirm a
sync leaves it `running` (progress is respected).

Automated: `tests/app/sep/test_crud.py` 67 passed; blast radius
(`test_crud` + `sync/` + `apps/inventory/`) 680 passed, 2 skipped;
`app/sep/crud.py` at 96% with no missing branches. Each new guard was
mutation-checked — dropping the fence's staleness predicate or widening
the new candidate class back to `PENDING` fails exactly the tests
written for it.

CI's `python / test` job has failed once on this branch, at `645a45d98`:
`tests/app/sep/test_config.py::TestDiagnosticsDeliveryInputs::test_defaults_to_not_configured`,
raising `APPS.15.MODULE_NAME Value error, No module named
app.sep.apps._scaffold_ci_scriptforward` — a scaffolding app name
reaching the settings the test builds, so workspace-dependent and
unrelated to these files. The later full runs on this branch are green.
Locally, `tests/app/sep/test_import_boundary.py` fails a varying subset
under xdist (3 tests one run, 2 the next) while passing 62/62 in
isolation, on a clean `main` too — the same environment-dependent class,
also unrelated.

## Checklist

- [x] New/modified functions have type hints and rST docstrings
- [x] New tests added for new features or bug fixes
- [ ] All tests pass locally (`make test`)
- [x] Pre-commit hooks pass (`make run-pre-commit`)
- [ ] Database migrations generated if models changed (`make
makemigrations`)
- [ ] User-facing changes documented (README, inline help, UI text)
- [ ] Configuration changes documented with examples
- [x] Changelog fragment added under `changelog.d/` if the change is
user-facing (`make changelog-add`), or confirmed N/A (internal-only
change, or a same-release-cycle fix for an unreleased sibling ticket)

Author: marcuscruz-percona

app/sep/crud.py

@@ -228,10 +228,11 @@ async def create(
         it raises a `SyncInstanceAlreadyInProgressError`. Otherwise, it creates and
         saves the new `SyncInstance`.
 
-        When ``stale_after`` is supplied, an in-progress conflict is first re-examined
-        for abandoned runs: items left behind by a killed worker would otherwise
-        block the syncer permanently, because the hanging-item sweep runs only when
-        the run exits through its context manager.
+        When ``stale_after`` is supplied, abandoned runs are reclaimed before that
+        check runs and without waiting for an item conflict to justify it, because a
+        worker killed after its last item write leaves a stale run that no item
+        conflict would ever surface. Whatever the reclaim spared is what the conflict
+        check then sees.
 
         :param session: The SQLAlchemy asynchronous session to use for database
             operations.
@@ -243,15 +244,14 @@ async def create(
         :raises SyncInstanceAlreadyInProgressError: If a SyncInstance with the same
             ``syncer`` is already in progress and could not be reclaimed as stale.
         """
+        if stale_after is not None:
+            await cls.reclaim_stale_runs(session, instance_create.syncer, stale_after)
+        # Read after the reclaim, never before it: the reclaim commits separately, so
+        # a run it spared for resuming mid-reclaim can hold an item an earlier read
+        # never saw, and refusing on a stale empty list would start a second run.
         syncs_in_progress = await cls._items_in_progress(
             session, instance_create.syncer
         )
-        if syncs_in_progress and stale_after is not None:
-            await cls.reclaim_stale_runs(session, instance_create.syncer, stale_after)
-            syncs_in_progress = await cls._items_in_progress(
-                session,
-                instance_create.syncer,
-            )
         if syncs_in_progress:
             raise SyncInstanceAlreadyInProgressError(syncs_in_progress)
         return await super().create(session, instance_create, **extra_fields)
@@ -291,12 +291,13 @@ async def reclaim_stale_runs(
         """Fail the runs of a syncer whose items stopped progressing long ago.
 
         A run is stale when the newest activity across **all** of its items predates
-        ``stale_after``, so a run still making progress is never reclaimed. The item
-        flip is a single conditional statement, so a second reclaimer arriving
-        concurrently matches no rows rather than reclaiming twice. It covers every
-        stale run already fenced as ``FAILED``, not only the ones this call fenced,
-        so a reclaim interrupted between its two statements resumes on the next
-        attempt instead of leaving the syncer blocked.
+        ``stale_after``, so a run still making progress is never reclaimed. Two
+        classes qualify: a run still holding ``PENDING``/``RUNNING`` items, and a
+        ``RUNNING`` run whose items have all gone terminal. A run that wrote no item
+        carries no activity to measure and is left alone.
+
+        Both the fence and the item flip are single conditional statements, so a
+        concurrent reclaimer matches no rows rather than reclaiming twice.
 
         ``snapshot_complete`` is deliberately left untouched: a partially applied run
         must never be counted as a complete generation.
@@ -323,36 +324,56 @@ async def reclaim_stale_runs(
         last_activity = func.max(
             func.coalesce(col(SyncItem.updated_at), col(SyncItem.created_at)),
         )
-        query = (
+        cutoff = utc_now() - stale_after
+        stale_activity = (
             select(col(SyncItem.sync_instance_id))
-            .where(col(SyncItem.sync_instance_id).in_(in_progress))
             .group_by(col(SyncItem.sync_instance_id))
-            .having(last_activity < utc_now() - stale_after)
+            .having(last_activity < cutoff)
         )
-        result = await cls._exec(session, query)
-        stale_instance_ids = list(result.all())
+        blocked = stale_activity.where(col(SyncItem.sync_instance_id).in_(in_progress))
+        # Kept as a second query rather than folded into the one above: that one must
+        # stay item-keyed so a reclaim interrupted after fencing still finds its run
+        # and releases the items. Excluding those runs makes the two sets disjoint,
+        # and grouping over items is what leaves a run that wrote none out of both.
+        idle = stale_activity.join(SyncInstance).where(
+            col(SyncInstance.syncer) == syncer,
+            col(SyncInstance.status) == SyncStatusEnum.RUNNING,
+            col(SyncItem.sync_instance_id).not_in(in_progress),
+        )
+        blocked_result = await cls._exec(session, blocked)
+        idle_result = await cls._exec(session, idle)
+        stale_instance_ids = [*blocked_result.all(), *idle_result.all()]
         if not stale_instance_ids:
             return []
         # The instance is fenced first, and only then are its items released.
         # Each statement commits on its own, so flipping the items first would
         # leave a window in which a reclaimed-but-live worker still reads
-        # ``RUNNING`` and walks into its retire phase. The status predicate keeps
-        # a run that finished between the query above and this update: it has
-        # already written its own verdict, and is no longer anyone's to reclaim.
+        # ``RUNNING`` and walks into its retire phase. Both predicates re-assert
+        # what the query above selected on, because it committed separately: a run
+        # that finished meanwhile has written its own verdict, and one whose worker
+        # resumed and touched an item is progressing after all. The re-assertion
+        # repeats the candidate narrowing so its grouping is index-seekable rather
+        # than spanning every item row; restricting the rows cannot change a
+        # surviving group's ``max()``, since every group is one instance's items.
         reclaimed_ids = await cls.update_where(
             session,
             {"status": SyncStatusEnum.FAILED},
             col(SyncInstance.id).in_(stale_instance_ids),
+            col(SyncInstance.id).in_(
+                stale_activity.where(
+                    col(SyncItem.sync_instance_id).in_(stale_instance_ids),
+                ),
+            ),
             col(SyncInstance.status).in_(
                 [SyncStatusEnum.PENDING, SyncStatusEnum.RUNNING],
             ),
             returning=["id"],
         )
         # Items are released for every stale instance already fenced, not only the
-        # ones this call fenced. A crash between the two statements leaves a FAILED
+        # ones this call fenced: a crash between the two statements leaves a FAILED
         # instance whose items were never released, and the update above then matches
-        # nothing on every later attempt -- so without this the syncer would stay
-        # blocked by exactly the abandoned run the reclaim exists to clear.
+        # nothing on every later attempt, blocking the syncer with exactly the
+        # abandoned run the reclaim exists to clear.
         fenced = await cls._exec(
             session,
             select(col(SyncInstance.id)).where(

changelog.d/SEP-1894.fixed.md

@@ -0,0 +1 @@
+Reconcile a sync run abandoned by a killed worker even when it left no unfinished items behind.