fix(sync): treat recently-started running jobs as active to stop concurrent pile-up (RES-948)

[jhkchan]

Problem (RES-948)

A burst of sync triggers piled up dozens of concurrent running sync_jobs on one channel (observed 58), thrashing the GPU. start_sync's guard rejected a new sync only when _is_task_active(channel_id) — an in-process check of self._active_tasks. A running row created moments ago by a concurrent trigger (or a different worker process) isn't in this process's registry, so the guard treated it as a crashed leftover, "recovered" it, and started another sync. Each trigger mis-recovered the others' fresh rows → pile-up.

Fix

Add _running_row_is_active(existing, channel_id): a running row blocks a new sync if either this process is running its task or it was started within _RECENT_RUNNING_GRACE_SECONDS (120s) — a concurrent/other-process run this registry can't see yet. Only a row that is both not-task-active here and older than the grace window is a genuine stale leftover, preserving the existing process-restart recovery path. Naive started_at (as Mongo may return) is coerced to UTC.

Tests

tests/services/test_sync_runner_concurrent_guard.py: recent-row-blocks-without-local-task, old-row-recoverable, task-active-always-blocks, missing-started_at, naive-utc-coercion. 5 passed locally.

Note: this closes the common race at the guard. A fully atomic one-job-per-channel claim (Mongo findAndModify/CAS) is a larger follow-up; the grace-window guard removes the observed pile-up without changing the recovery contract.

Part of epic RES-943 (RLP full-corpus scale + no-cloud gaps).

🤖 Generated with Claude Code

https://claude.ai/code/session_01PMMM6KQXmzAEA42UxpyMUm

[jhkchan]

Independent adversarial review — APPROVE as a burst mitigation (with a known-limitation follow-up).

Verified correct for what it claims: existing is a Pydantic SyncJob, the Mongo client is not tz_aware so started_at is naive UTC and the naive→UTC coercion is the real load-bearing path (correct, no throw), the legitimate crash-recovery path is preserved (old orphaned rows still recoverable), and no new permanent deadlock (a within-grace crash self-heals in ≤120s).

Honest scope call — [P1, by-design limitation not a bug]: this is a recency-based mitigation, not an atomic one-job-per-channel guarantee. For a sync that outlives the 120s window in a multi-process/replica deployment, at T+130s another process sees started > 120s, treats the still-running peer job as stale, "recovers" it, and starts a second sync — so the race is narrowed, not closed, and large-channel syncs routinely exceed 120s. Acceptable to ship for the observed 58-job burst. Follow-ups: (1) make the window a setting sized to p95 sync duration; (2) an atomic guard (partial unique index on sync_jobs {channel_id} where status=="running", or CAS find_one_and_update) for a true invariant; (3) make the recovery write conditional ({"id":job_id,"status":"running"}) so it can't clobber a peer's live job. Opening a follow-up ticket.