fix(pipeline): resilient cursor save + SQLite cursor lock + gateway retry + scheduled autoretry [修复组④]

backend/channels/api_channel.py

@@ -156,8 +156,17 @@ async def _send(
         except httpx.TimeoutException as exc:
             raise ChannelFetchError(f"API request to {url} timed out") from exc
         except httpx.HTTPStatusError as exc:
+            # AUDIT C13: classify by status so a gateway blip (502/503/504,
+            # Cloudflare's 520/522/524, ...) reaches the celery retry boundary
+            # instead of defaulting to permanent alongside a genuine 4xx.
+            from backend.pipeline.error_taxonomy import is_retryable_http_status
+
+            status = exc.response.status_code
+            error_type = (
+                "RetryableHTTPStatus" if is_retryable_http_status(status) else "PermanentHTTPStatus"
+            )
             raise ChannelFetchError(
-                f"HTTP {exc.response.status_code}: {exc.response.text[:200]}"
+                f"HTTP {status}: {exc.response.text[:200]}", error_type=error_type
             ) from exc
         except Exception as exc:
             raise ChannelFetchError(f"API request failed: {exc}") from exc

backend/channels/rss_channel.py

@@ -151,7 +151,22 @@ async def fetch(self, ctx: FetchContext) -> FetchResult:
         if response.status_code == 304:
             # Not Modified — no new entries; preserve the cursor as-is.
             return FetchResult(items=[], next_cursor=(cursor or None), has_more=False)
-        response.raise_for_status()
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            # AUDIT C13: classify by status so a gateway blip (502/503/504,
+            # Cloudflare's 520/522/524, ...) reaches the celery retry boundary
+            # instead of defaulting to permanent alongside a genuine 4xx —
+            # this call used to be bare, so ANY status error (not just
+            # gateway ones) propagated as a raw HTTPStatusError with no
+            # retry-classification hint at all.
+            from backend.pipeline.error_taxonomy import is_retryable_http_status
+
+            status = exc.response.status_code
+            error_type = (
+                "RetryableHTTPStatus" if is_retryable_http_status(status) else "PermanentHTTPStatus"
+            )
+            raise ChannelFetchError(f"HTTP {status} fetching feed", error_type=error_type) from exc
 
         parsed = feedparser.parse(response.text)
         if parsed.bozo and not parsed.entries:

backend/migrations/versions/t9y0z1a2b3c4_add_source_cursor_version.py

@@ -0,0 +1,33 @@
+"""add version column to source_cursors (optimistic-concurrency cursor save)
+
+Revision ID: t9y0z1a2b3c4
+Revises: s8x9y0z1a2b3
+
+AUDIT C10: ``DBCursorStore.save()`` used ``SELECT ... FOR UPDATE`` to guard
+against two concurrent runs of the same source losing an update, but that
+clause is a silent no-op on SQLite (the dialect accepts it but never takes a
+row lock) — so the lost-update protection only ever existed on a Postgres
+deployment. This adds an integer ``version`` column so ``save()`` can use
+optimistic concurrency (``UPDATE ... WHERE version = ?``) instead, which
+works identically on SQLite and Postgres.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "t9y0z1a2b3c4"
+down_revision = "s8x9y0z1a2b3"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("source_cursors") as batch:
+        batch.add_column(
+            sa.Column("version", sa.Integer(), nullable=False, server_default="0")
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("source_cursors") as batch:
+        batch.drop_column("version")

backend/models/source_cursor.py

@@ -1,4 +1,4 @@
-from sqlalchemy import JSON, String, UniqueConstraint
+from sqlalchemy import JSON, Integer, String, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from backend.models.base import TimestampMixin
@@ -20,3 +20,10 @@ class SourceCursor(TimestampMixin):
 
     source_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
     cursor: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
+    #: Optimistic-concurrency guard (AUDIT C10): ``DBCursorStore.save()`` does
+    #: ``UPDATE ... WHERE version = ?`` and bumps this by one, so a losing
+    #: concurrent writer's update affects 0 rows (detectable, retryable)
+    #: instead of silently overwriting another writer's cursor. Works
+    #: identically on SQLite and Postgres, unlike ``SELECT ... FOR UPDATE``
+    #: (a silent no-op on SQLite).
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

backend/pipeline/cursor_store.py

@@ -84,38 +84,59 @@ async def load(self, source_id: str) -> dict[str, Any] | None:
             return dict(row.cursor) if row and row.cursor else None
 
     async def save(self, source_id: str, cursor: dict[str, Any]) -> CommitResult:
-        """Upsert the cursor row, serialized per-source so two concurrent runs of
-        the same source cannot lose an update.
+        """Upsert the cursor row using optimistic concurrency (a ``version``
+        column), serialized per-source so two concurrent runs of the same
+        source cannot lose an update.
 
         Returns a ``CommitResult`` reflecting the REAL commit: ``advanced`` is
         True when the stored value actually changed (including the
         first-ever row for a source), False when the caller re-saved an
         identical value onto an existing row (a true no-op write) — see
         ``CommitResult``'s docstring.
 
-        Plain SELECT-then-INSERT/UPDATE (the prior implementation) has a race: two
-        concurrent ``save()`` calls for the same ``source_id`` can both SELECT
-        before either commits, so whichever COMMITs last silently overwrites the
-        other's cursor — a lost update that manifests as skipped data on the next
-        incremental fetch (the loser's cursor value is gone as if it never
-        advanced). Locking the existing row with ``SELECT ... FOR UPDATE`` inside
-        this transaction closes that window: the second concurrent caller blocks
-        on the lock until the first commits, then reads the first's committed
-        value before applying its own write — no schema change needed (the row
-        already exists once the first save lands).
+        Plain SELECT-then-INSERT/UPDATE (the original implementation) has a race:
+        two concurrent ``save()`` calls for the same ``source_id`` can both
+        SELECT before either commits, so whichever COMMITs last silently
+        overwrites the other's cursor — a lost update that manifests as skipped
+        data on the next incremental fetch (the loser's cursor value is gone as
+        if it never advanced).
+
+        AUDIT C10: an earlier version of this method tried to close that window
+        with ``SELECT ... FOR UPDATE``. That is a silent no-op on SQLite — the
+        dialect accepts the clause but never actually takes a row lock — so on
+        this project's default (SQLite) deployment it protected nothing; only a
+        Postgres deployment ever got real locking from it. Optimistic
+        concurrency via ``SourceCursor.version`` instead works identically on
+        both backends, because it depends only on ordinary SQL semantics
+        (``UPDATE ... WHERE version = ?`` either matches the row — nobody else
+        committed since we read it — and advances the version, or matches zero
+        rows because someone else already advanced it) rather than a
+        backend-specific locking primitive: read the row's current ``version``,
+        then update conditioned on that same version. A losing writer's UPDATE
+        affects 0 rows; it detects that and retries against the fresh row
+        instead of silently overwriting (or being silently overwritten by) the
+        winner.
 
         The remaining race is the very first save for a source (no row yet): two
-        concurrent callers can both miss the row under ``FOR UPDATE`` (nothing to
-        lock) and both attempt an INSERT. ``source_cursors`` already has
+        concurrent callers can both miss the row (nothing to read a version
+        from) and both attempt an INSERT. ``source_cursors`` already has
         ``UniqueConstraint(source_id)``, so the loser's INSERT raises
-        ``IntegrityError``; that is caught and retried (bounded) as a locked
+        ``IntegrityError``; that is caught and retried (bounded) as a versioned
         UPDATE against the row the winner just created, rather than losing the
-        retry's cursor value. The retry loop (not just a single re-SELECT) also
-        covers same-connection dirty-read artifacts some SQLite pooling setups
-        can exhibit, where the row briefly appears absent to the loser even
-        after its own INSERT already conflicted.
+        retry's cursor value.
+
+        Both races share one bounded retry loop: a losing UPDATE (0 rows) or a
+        losing INSERT (IntegrityError) retries against a fresh read, up to
+        ``attempts`` times, raising rather than silently dropping the write if
+        contention outlives the budget — in practice there are at most two
+        concurrent savers for a given source_id (the scheduler and a manual
+        re-run), so exhausting this budget means genuine, unexpected contention
+        worth surfacing, not routine noise. The retry loop (not just a single
+        re-SELECT) also covers same-connection dirty-read artifacts some SQLite
+        pooling setups can exhibit, where the row briefly appears absent to the
+        loser even after its own INSERT already conflicted.
         """
-        from sqlalchemy import select
+        from sqlalchemy import select, update
         from sqlalchemy.exc import IntegrityError
 
         from backend.database import AsyncSessionLocal
@@ -126,15 +147,33 @@ async def save(self, source_id: str, cursor: dict[str, Any]) -> CommitResult:
             async with AsyncSessionLocal() as session:
                 row = (
                     await session.execute(
-                        select(SourceCursor)
-                        .where(SourceCursor.source_id == source_id)
-                        .with_for_update()
+                        select(SourceCursor).where(SourceCursor.source_id == source_id)
                     )
                 ).scalar_one_or_none()
                 if row is not None:
                     old_cursor = dict(row.cursor) if row.cursor else None
                     new_cursor = dict(cursor)
-                    row.cursor = new_cursor
+                    result = await session.execute(
+                        update(SourceCursor)
+                        .where(
+                            SourceCursor.id == row.id,
+                            SourceCursor.version == row.version,
+                        )
+                        .values(cursor=new_cursor, version=row.version + 1)
+                    )
+                    if result.rowcount == 0:
+                        # Lost the optimistic-lock race: another save() advanced
+                        # the version between our SELECT and this UPDATE. Retry
+                        # against the fresh row instead of silently dropping (or
+                        # clobbering) this call's cursor value.
+                        await session.rollback()
+                        if attempt == attempts - 1:
+                            raise RuntimeError(
+                                f"DBCursorStore.save() lost the optimistic-lock "
+                                f"race for source_id={source_id!r} after "
+                                f"{attempts} attempts"
+                            )
+                        continue
                     await session.commit()
                     return CommitResult(
                         advanced=(old_cursor != new_cursor),
@@ -145,7 +184,9 @@ async def save(self, source_id: str, cursor: dict[str, Any]) -> CommitResult:
                 try:
                     new_cursor = dict(cursor)
                     async with session.begin_nested():
-                        session.add(SourceCursor(source_id=source_id, cursor=new_cursor))
+                        session.add(
+                            SourceCursor(source_id=source_id, cursor=new_cursor, version=0)
+                        )
                     await session.commit()
                     # First-ever row for this source: there is now a persisted
                     # value where there wasn't one — that counts as advanced.

backend/pipeline/error_taxonomy.py

@@ -79,11 +79,16 @@ def effective_error_type(exc: BaseException) -> str:
 
 
 def is_retryable_http_status(status_code: int) -> bool:
-    """429/5xx are handled by RateLimitedClient's own backoff before ever
+    """408/429/5xx are handled by RateLimitedClient's own backoff before ever
     reaching the pipeline layer; if one leaks through anyway, treat it as
-    retryable (transient server-side condition). Other 4xx are permanent —
-    retrying the same malformed/unauthorized request won't change the
-    outcome."""
-    if status_code == 429 or status_code >= 500:
+    retryable (transient — a slow upstream or a rate limit, not a durably
+    broken request). Other 4xx are permanent — retrying the same
+    malformed/unauthorized request won't change the outcome.
+
+    AUDIT C13: 408 (Request Timeout) was previously left out of this set —
+    the same request against the same upstream can easily succeed on a later
+    attempt, same as a 429 or a 5xx, so it belongs with them rather than
+    defaulting to permanent."""
+    if status_code in (408, 429) or status_code >= 500:
         return True
     return False

backend/pipeline/http_client.py

@@ -16,7 +16,11 @@
 import httpx
 
 #: HTTP statuses the client retries (rate limit + transient server errors).
-RETRY_STATUS = frozenset({429, 500, 502, 503})
+#: AUDIT C13: 504/520/522/524 (gateway timeout + Cloudflare's own gateway-error
+#: codes) belong alongside 502/503 — a slow/misbehaving upstream through a
+#: proxy surfaces as any of these, not just 502/503, and none of them are
+#: reasons to give up permanently.
+RETRY_STATUS = frozenset({429, 500, 502, 503, 504, 520, 522, 524})
 
 
 def parse_rate(rate: str) -> float:

backend/pipeline/pipeline.py

@@ -369,10 +369,38 @@ async def run_pipeline(
     if pending_cursor is not None and cursor_source_id is not None:
         from backend.pipeline.cursor_store import DBCursorStore
 
-        commit_result = await DBCursorStore().save(cursor_source_id, pending_cursor)
-        cursor_advanced = commit_result.advanced
-        logger.info("[task:%s] cursor committed post-write | source=%s advanced=%s",
-                    task_id, cursor_source_id, cursor_advanced)
+        # AUDIT C11: this used to be the only post-sink-write step without
+        # error handling — the sink already durably committed this batch's
+        # records above, so a cursor-save failure here must not fail the run
+        # (that would false-fail a run whose data landed, and Celery would
+        # re-collect the same window on retry, re-storing already-stored
+        # records). Log loud enough to find the stuck cursor, surface a
+        # warning event when there's a run to attach it to, and keep going
+        # with cursor_advanced=False — the measurement below then honestly
+        # reflects "didn't advance" rather than guessing.
+        try:
+            commit_result = await DBCursorStore().save(cursor_source_id, pending_cursor)
+        except Exception as exc:
+            logger.error(
+                "[task:%s] cursor save failed (non-fatal, run stays successful) | "
+                "source=%s cursor=%r | %s",
+                task_id, cursor_source_id, pending_cursor, exc,
+            )
+            if run_id:
+                await events.emit(
+                    run_id, "store",
+                    f"游标保存失败（不影响本次任务结果）: {exc}",
+                    level="warning",
+                    detail={
+                        "source_id": cursor_source_id,
+                        "cursor": pending_cursor,
+                        "error": str(exc),
+                    },
+                )
+        else:
+            cursor_advanced = commit_result.advanced
+            logger.info("[task:%s] cursor committed post-write | source=%s advanced=%s",
+                        task_id, cursor_source_id, cursor_advanced)
 
     # Step 4: AI processing
     effective_ai_config = agent_config or source.ai_config

backend/worker/tasks.py

@@ -109,13 +109,25 @@ async def _mark_retries_exhausted(
     base=_AlertOnRetriesExhaustedTask,
     name="run_collection",
     max_retries=3,
-    default_retry_delay=60,
     # run_pipeline() only re-raises exceptions its error taxonomy classified
     # as retryable (backend.pipeline.error_taxonomy.is_retryable) — anything
     # deterministic is already swallowed into a returned PipelineResult
     # before it gets here. So catching broadly at this boundary is correct:
     # the filtering already happened one layer down, not duplicated here.
     autoretry_for=(Exception,),
+    # AUDIT C14: a fixed 60s delay for every retry means every failed run on
+    # a source retries in lockstep (same delay, same jitter-free instant) —
+    # a source down for minutes hammers it 3x on a synchronous 60s/60s/60s
+    # cadence instead of backing off. retry_backoff=True + retry_jitter=True
+    # is celery's standard autoretry_for pairing: countdown = min(
+    # retry_backoff_max, 1 * 2**retries), then full-jittered — 1st retry
+    # ~0-1s, 2nd ~0-2s, 3rd ~0-4s, capped at 600s. default_retry_delay is
+    # unused once retry_backoff is set (celery computes countdown from the
+    # backoff formula instead), so it's dropped rather than left as
+    # never-consulted dead config.
+    retry_backoff=True,
+    retry_backoff_max=600,
+    retry_jitter=True,
 )
 def run_collection(self: Task, task_id: str, parameters: dict | None = None) -> dict:
     """Execute the full collection pipeline for a task."""
@@ -128,7 +140,25 @@ def run_collection(self: Task, task_id: str, parameters: dict | None = None) ->
     ))
 
 
-@celery_app.task(name="run_scheduled_collection")
+@celery_app.task(
+    name="run_scheduled_collection",
+    max_retries=3,
+    # AUDIT C14: run_scheduled_pipeline() funnels into the same
+    # run_collection_pipeline() -> run_pipeline() as run_collection above, so
+    # the same contract applies — run_pipeline() only re-raises exceptions
+    # already classified retryable (backend.pipeline.error_taxonomy.
+    # is_retryable), everything else is swallowed into a returned dict before
+    # it gets here. Without autoretry_for here, those retryable exceptions
+    # just failed the celery task outright: a scheduled run had zero retries
+    # where a manually-triggered one (run_collection) got 3 with backoff,
+    # contradicting pipeline.py's own "let this propagate ... so its
+    # autoretry_for policy applies" comment. Same backoff+jitter shape as
+    # run_collection for the same reason (see there).
+    autoretry_for=(Exception,),
+    retry_backoff=True,
+    retry_backoff_max=600,
+    retry_jitter=True,
+)
 def run_scheduled_collection(schedule_id: str, source_id: str, parameters: dict | None = None) -> dict:
     """Create a CollectionTask for a scheduled run, execute pipeline, auto-disable if one-time."""
     from backend.pipeline.runner import run_scheduled_pipeline