From 158c38fe457325c0cbeecdb36ff5d1eae6c609f8 Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 19 Jul 2026 02:06:55 +0800 Subject: [PATCH] fix(pipeline): resilient cursor save + SQLite cursor lock + gateway retry classification + scheduled autoretry [C10,C11,C13,C14] --- backend/channels/api_channel.py | 11 ++- backend/channels/rss_channel.py | 17 +++- .../t9y0z1a2b3c4_add_source_cursor_version.py | 33 +++++++ backend/models/source_cursor.py | 9 +- backend/pipeline/cursor_store.py | 91 ++++++++++++++----- backend/pipeline/error_taxonomy.py | 15 ++- backend/pipeline/http_client.py | 6 +- backend/pipeline/pipeline.py | 36 +++++++- backend/worker/tasks.py | 34 ++++++- tests/unit/channels/test_api_channel.py | 57 ++++++++++++ tests/unit/channels/test_rss_fetch.py | 55 ++++++++++- tests/unit/pipeline/test_db_cursor_store.py | 81 +++++++++++++---- tests/unit/pipeline/test_error_taxonomy.py | 16 ++++ tests/unit/pipeline/test_http_client.py | 18 ++++ tests/unit/pipeline/test_pipeline_cursor.py | 76 ++++++++++++++++ tests/unit/worker/test_tasks.py | 32 ++++++- 16 files changed, 523 insertions(+), 64 deletions(-) create mode 100644 backend/migrations/versions/t9y0z1a2b3c4_add_source_cursor_version.py diff --git a/backend/channels/api_channel.py b/backend/channels/api_channel.py index a8e2515..7a6312a 100644 --- a/backend/channels/api_channel.py +++ b/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 diff --git a/backend/channels/rss_channel.py b/backend/channels/rss_channel.py index 20aa4a0..f69cb51 100644 --- a/backend/channels/rss_channel.py +++ b/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: diff --git a/backend/migrations/versions/t9y0z1a2b3c4_add_source_cursor_version.py b/backend/migrations/versions/t9y0z1a2b3c4_add_source_cursor_version.py new file mode 100644 index 0000000..3d01c1a --- /dev/null +++ b/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") diff --git a/backend/models/source_cursor.py b/backend/models/source_cursor.py index 7a137fd..2707c18 100644 --- a/backend/models/source_cursor.py +++ b/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) diff --git a/backend/pipeline/cursor_store.py b/backend/pipeline/cursor_store.py index 27848a4..7d4a087 100644 --- a/backend/pipeline/cursor_store.py +++ b/backend/pipeline/cursor_store.py @@ -84,8 +84,9 @@ 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 @@ -93,29 +94,49 @@ async def save(self, source_id: str, cursor: dict[str, Any]) -> CommitResult: 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. diff --git a/backend/pipeline/error_taxonomy.py b/backend/pipeline/error_taxonomy.py index 9414942..8caf683 100644 --- a/backend/pipeline/error_taxonomy.py +++ b/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 diff --git a/backend/pipeline/http_client.py b/backend/pipeline/http_client.py index d48c62d..4042271 100644 --- a/backend/pipeline/http_client.py +++ b/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: diff --git a/backend/pipeline/pipeline.py b/backend/pipeline/pipeline.py index 6170364..adb26c3 100644 --- a/backend/pipeline/pipeline.py +++ b/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 diff --git a/backend/worker/tasks.py b/backend/worker/tasks.py index 86fd6d3..9dd920d 100644 --- a/backend/worker/tasks.py +++ b/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 diff --git a/tests/unit/channels/test_api_channel.py b/tests/unit/channels/test_api_channel.py index 50030ea..9898a57 100644 --- a/tests/unit/channels/test_api_channel.py +++ b/tests/unit/channels/test_api_channel.py @@ -647,6 +647,63 @@ async def test_fetch_http_error_raises_channel_fetch_error(channel): await channel.fetch(ctx) +# ── AUDIT C13: gateway statuses classify retryable, other 4xx stay permanent ── + +def _mock_status_error(status: int): + import httpx + + return httpx.HTTPStatusError( + message=f"HTTP {status}", + request=MagicMock(), + response=MagicMock(status_code=status, text=f"error {status}"), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [502, 503, 504, 520, 522, 524]) +async def test_fetch_gateway_status_classified_retryable(channel, status): + """504/520/522/524 (and 502/503) must set error_type="RetryableHTTPStatus" + on the raised ChannelFetchError so pipeline.py's retry taxonomy sees them + as transient, not a permanent failure — previously these all leaked + through as a bare wrapper with no retry-classification hint.""" + mock_response = MagicMock() + mock_response.status_code = status + mock_response.raise_for_status = MagicMock(side_effect=_mock_status_error(status)) + http = AsyncMock() + http.request = AsyncMock(return_value=mock_response) + ctx = FetchContext( + config={"base_url": "https://api.example.com", "endpoint": "/err"}, + params={}, + http=http, + ) + + with pytest.raises(ChannelFetchError) as exc_info: + await channel.fetch(ctx) + + assert exc_info.value.error_type == "RetryableHTTPStatus" + + +@pytest.mark.asyncio +async def test_fetch_client_404_classified_permanent(channel): + """A genuine 4xx (not 408/429) stays permanent — retrying an unchanged + malformed/missing-resource request can't succeed.""" + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.raise_for_status = MagicMock(side_effect=_mock_status_error(404)) + http = AsyncMock() + http.request = AsyncMock(return_value=mock_response) + ctx = FetchContext( + config={"base_url": "https://api.example.com", "endpoint": "/missing"}, + params={}, + http=http, + ) + + with pytest.raises(ChannelFetchError) as exc_info: + await channel.fetch(ctx) + + assert exc_info.value.error_type == "PermanentHTTPStatus" + + # ── health_check (GOAL-4 PR-E: real per-source probe) ─────────────────────────── @pytest.mark.asyncio diff --git a/tests/unit/channels/test_rss_fetch.py b/tests/unit/channels/test_rss_fetch.py index 9cb5f6c..e0c4eb0 100644 --- a/tests/unit/channels/test_rss_fetch.py +++ b/tests/unit/channels/test_rss_fetch.py @@ -2,11 +2,11 @@ runner driving it end to end (the thin-channel / thick-runner vertical slice).""" from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest -from backend.channels.base import FetchContext +from backend.channels.base import ChannelFetchError, FetchContext from backend.channels.rss_channel import RSSChannel @@ -87,6 +87,57 @@ async def test_fetch_304_no_new_items_keeps_cursor_and_sends_conditional(): assert http.calls[0][1]["headers"]["If-None-Match"] == 'W/"v1"' +# ── AUDIT C13: gateway statuses classify retryable, other 4xx stay permanent ── + +class _StatusErrorResp: + """A response whose raise_for_status() raises a REAL httpx.HTTPStatusError + (unlike this file's own _Resp fake, whose raise_for_status raises a bare + RuntimeError) — needed to exercise fetch()'s httpx.HTTPStatusError + classification branch, which _Resp never triggers.""" + + def __init__(self, status_code): + import httpx + + self.status_code = status_code + self.headers = {} + self._exc = httpx.HTTPStatusError( + message=f"HTTP {status_code}", + request=MagicMock(), + response=MagicMock(status_code=status_code, text=f"error {status_code}"), + ) + + def raise_for_status(self): + raise self._exc + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [502, 503, 504, 520, 522, 524]) +async def test_fetch_gateway_status_classified_retryable(status): + """fetch()'s response.raise_for_status() used to be bare (no try/except + at all) — any status error, gateway or not, propagated as a raw + httpx.HTTPStatusError with no retry-classification hint. Now a gateway + status must raise ChannelFetchError(error_type="RetryableHTTPStatus").""" + http = _Http(_StatusErrorResp(status)) + ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) + + with pytest.raises(ChannelFetchError) as exc_info: + await RSSChannel().fetch(ctx) + + assert exc_info.value.error_type == "RetryableHTTPStatus" + + +@pytest.mark.asyncio +async def test_fetch_client_404_classified_permanent(): + """A genuine 4xx (not 408/429) stays permanent.""" + http = _Http(_StatusErrorResp(404)) + ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) + + with pytest.raises(ChannelFetchError) as exc_info: + await RSSChannel().fetch(ctx) + + assert exc_info.value.error_type == "PermanentHTTPStatus" + + @pytest.mark.asyncio async def test_run_channel_drives_rss_and_persists_cursor(): from backend.pipeline.channel_runner import run_channel diff --git a/tests/unit/pipeline/test_db_cursor_store.py b/tests/unit/pipeline/test_db_cursor_store.py index 1943852..7b7c27f 100644 --- a/tests/unit/pipeline/test_db_cursor_store.py +++ b/tests/unit/pipeline/test_db_cursor_store.py @@ -9,22 +9,31 @@ from backend.pipeline.cursor_store import CommitResult, DBCursorStore -# The concurrency tests below run on the default `db_engine` fixture (SQLite in -# CI/local), which does NOT actually enforce `SELECT ... FOR UPDATE` row locks — -# so on SQLite they prove the code PATH (insert-race fallback, no unhandled -# error, no lost value) but NOT that the lock genuinely serializes contending -# writers. The `test_concurrent_saves_postgres_*` variant re-runs the same -# scenario against a real Postgres (where FOR UPDATE is enforced), and is the -# one that actually closes AUDIT follow-up (c). It is skipped unless a Postgres -# URL is provided (env TEST_DATABASE_URL_PG, or DATABASE_URL if it's postgres), -# so it's inert locally/on SQLite and active in a Postgres-backed CI run. +# AUDIT C10: save() used to guard the lost-update race with +# `SELECT ... FOR UPDATE`, which is a silent no-op on SQLite (the dialect +# accepts the clause but never actually takes a row lock) — so on the default +# `db_engine` fixture (SQLite) the concurrency tests below could only prove +# the code PATH (insert-race fallback, no unhandled error, no lost value), +# never that a lock genuinely serialized contending writers; only the +# Postgres-gated variant further down proved that. +# +# save() now uses optimistic concurrency instead (a `version` column + +# `UPDATE ... WHERE version = ?`), which depends only on ordinary SQL +# semantics — an UPDATE's WHERE-match-then-write is atomic on every backend — +# rather than a backend-specific locking primitive. So the SQLite tests below +# now genuinely prove no lost update, not just the code path. The +# Postgres-gated variant is kept as a second-backend confidence check (same +# mechanism, a different driver/connection), not because it's uniquely +# load-bearing anymore; it's skipped unless a Postgres URL is provided (env +# TEST_DATABASE_URL_PG, or DATABASE_URL if it's postgres). _PG_URL = os.environ.get("TEST_DATABASE_URL_PG") or ( os.environ.get("DATABASE_URL", "") if os.environ.get("DATABASE_URL", "").startswith(("postgresql", "postgres")) else "" ) _requires_postgres = pytest.mark.skipif( - not _PG_URL, reason="no Postgres URL (TEST_DATABASE_URL_PG / postgres DATABASE_URL) — FOR UPDATE locking can only be verified on Postgres" + not _PG_URL, + reason="no Postgres URL (TEST_DATABASE_URL_PG / DATABASE_URL) — extra cross-backend check", ) @@ -105,11 +114,12 @@ async def test_concurrent_saves_no_lost_update_existing_row(db_engine): ) # Neither concurrent save should raise, and each returns a real - # CommitResult (not None) reflecting what it actually committed. (On - # SQLite, which doesn't enforce FOR UPDATE, exactly which of the two - # calls observes which "old" value isn't guaranteed — only that both - # return a real CommitResult with no exception, same as before this - # change's return-type upgrade.) + # CommitResult (not None) reflecting what it actually committed. + # Whichever call's SELECT reads the fresher version wins its UPDATE + # outright; the other's UPDATE ... WHERE version = ? affects 0 rows + # (its read was already stale) and the optimistic-lock retry loop + # re-reads and re-applies it on top — so exactly which call observes + # which "old" value isn't guaranteed, only that both land cleanly. assert all(isinstance(r, CommitResult) for r in results) final = await store.load("src-1") @@ -146,17 +156,48 @@ async def test_concurrent_saves_no_lost_update_first_insert_race(db_engine): assert final in ({"etag": "from-A"}, {"etag": "from-B"}) +# ── AUDIT C10: the optimistic-lock version column in isolation ───────────── + +@pytest.mark.asyncio +async def test_save_version_increments_on_each_successful_save(db_engine): + """The version column is the whole mechanism behind the optimistic lock: + it must actually advance by one on every successful save, or + `UPDATE ... WHERE version = ?` would never detect a stale read.""" + from sqlalchemy import select + + from backend.models.source_cursor import SourceCursor + + sessionmaker = _sessionmaker(db_engine) + with patch("backend.database.AsyncSessionLocal", sessionmaker): + store = DBCursorStore() + await store.save("src-ver", {"etag": "v0"}) + await store.save("src-ver", {"etag": "v1"}) + await store.save("src-ver", {"etag": "v2"}) + + async with sessionmaker() as session: + row = ( + await session.execute( + select(SourceCursor).where(SourceCursor.source_id == "src-ver") + ) + ).scalar_one() + + assert row.version == 2 + assert row.cursor == {"etag": "v2"} + + # ── AUDIT follow-up (c): FOR UPDATE locking verified on real Postgres ─────── @_requires_postgres @pytest.mark.asyncio async def test_concurrent_saves_postgres_for_update_serializes(): """Same no-lost-update scenario as the SQLite tests above, but against a - real Postgres where `SELECT ... FOR UPDATE` is actually enforced — this is - the run that genuinely proves the row lock serializes contending writers - (SQLite silently ignores FOR UPDATE, so the tests above can't). Skipped - unless a Postgres URL is configured; intended for the Postgres-backed CI - job (which already stands up Postgres + runs `alembic upgrade head`).""" + real Postgres connection — a second-backend confidence check that the + same optimistic-concurrency mechanism (version column + UPDATE ... WHERE + version = ?) behaves the same way through a different driver, not a + uniquely load-bearing proof anymore (the SQLite tests above already + genuinely exercise the same UPDATE...WHERE semantics). Skipped unless a + Postgres URL is configured; intended for the Postgres-backed CI job + (which already stands up Postgres + runs `alembic upgrade head`).""" from sqlalchemy.ext.asyncio import create_async_engine from backend.database import Base diff --git a/tests/unit/pipeline/test_error_taxonomy.py b/tests/unit/pipeline/test_error_taxonomy.py index 6ca23b4..5458139 100644 --- a/tests/unit/pipeline/test_error_taxonomy.py +++ b/tests/unit/pipeline/test_error_taxonomy.py @@ -44,3 +44,19 @@ def test_leaked_429_and_5xx_are_retryable(status): @pytest.mark.parametrize("status", [400, 401, 403, 404, 422]) def test_client_4xx_are_permanent(status): assert is_retryable_http_status(status) is False + + +# ── AUDIT C13: gateway statuses + 408 must classify retryable ─────────────── + +@pytest.mark.parametrize("status", [504, 520, 522, 524]) +def test_gateway_statuses_are_retryable(status): + """504 (gateway timeout) and Cloudflare's own 520/522/524 gateway-error + codes are transient upstream/proxy conditions, same as 502/503 — not + reasons to give up permanently.""" + assert is_retryable_http_status(status) is True + + +def test_request_timeout_408_is_retryable(): + """408 is a transient per-request timeout, not a durably broken request — + it belongs with 429/5xx, not with the permanent 4xx family.""" + assert is_retryable_http_status(408) is True diff --git a/tests/unit/pipeline/test_http_client.py b/tests/unit/pipeline/test_http_client.py index c73792a..bc382a7 100644 --- a/tests/unit/pipeline/test_http_client.py +++ b/tests/unit/pipeline/test_http_client.py @@ -93,5 +93,23 @@ async def test_gives_up_after_max_retries(monkeypatch): assert client.calls == 4 # initial + 3 retries +# ── AUDIT C13: gateway statuses (504 + Cloudflare 520/522/524) retry too ──── + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [504, 520, 522, 524]) +async def test_retries_on_gateway_statuses_then_succeeds(monkeypatch, status): + """504 (gateway timeout) and Cloudflare's 520/522/524 are transient + upstream/proxy conditions same as 502/503 — RETRY_STATUS previously + stopped at 503, so these leaked straight to the caller with zero retry.""" + monkeypatch.setattr("backend.pipeline.http_client.asyncio.sleep", lambda d: _noop()) + + client = FakeClient([httpx.Response(status), httpx.Response(200)]) + rl = RateLimitedClient(client, TokenBucket(1000), log=None) + resp = await rl.get("http://x") + + assert resp.status_code == 200 + assert client.calls == 2 + + async def _noop() -> None: return None diff --git a/tests/unit/pipeline/test_pipeline_cursor.py b/tests/unit/pipeline/test_pipeline_cursor.py index b3b0d95..43c6b4f 100644 --- a/tests/unit/pipeline/test_pipeline_cursor.py +++ b/tests/unit/pipeline/test_pipeline_cursor.py @@ -1,6 +1,7 @@ """Incremental cursor commit: the pipeline advances the persisted cursor ONLY after the write sink accepts the batch — never during fetch, never on sink failure.""" +import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -97,3 +98,78 @@ async def test_cursor_not_committed_when_sink_fails(db_session): # Sink raised → cursor must NOT advance past unwritten data. assert result.success is False save_mock.assert_not_awaited() + + +# ── AUDIT C11: cursor save must not turn a durably-written batch into a failed run ── + +@pytest.mark.asyncio +async def test_cursor_save_failure_keeps_run_successful(db_session, caplog): + """DBCursorStore().save() used to be the only post-sink-write step with + no error handling — if it raised, the sink had already durably committed + this batch's records, so failing the whole run here would be a false + failure (and Celery would re-collect the same window on retry, re-storing + already-stored records). A save() failure must log loud (task_id + the + cursor value that failed to persist) and keep the run successful.""" + from backend.pipeline.pipeline import run_pipeline + + source, task = await _seed(db_session) + save_mock = AsyncMock(side_effect=RuntimeError("disk full")) + cr = ChannelResult.ok( + [{"title": "x"}], __cursor_pending__={"etag": "v2"}, __cursor_source_id__=source.id + ) + + with ( + patch("backend.pipeline.collector.collect", return_value=cr), + patch("backend.pipeline.cursor_store.DBCursorStore") as DB, + caplog.at_level(logging.ERROR), + ): + DB.return_value.save = save_mock + result = await run_pipeline( + task.id, source, enable_ai=False, enable_notifications=False, sink=_ok_sink() + ) + + assert result.success is True + save_mock.assert_awaited_once_with(source.id, {"etag": "v2"}) + + error_records = [r for r in caplog.records if r.levelno >= logging.ERROR] + assert len(error_records) == 1 + message = error_records[0].getMessage() + assert task.id in message + assert source.id in message + assert "v2" in message + + +@pytest.mark.asyncio +async def test_cursor_save_failure_emits_warning_event_when_run_id_present(db_session): + """Same failure, but with a run_id: a warning event must reach the run's + own trace (not just a worker log line), without flipping the run to + failed.""" + from backend.pipeline.pipeline import run_pipeline + + source, task = await _seed(db_session) + save_mock = AsyncMock(side_effect=RuntimeError("disk full")) + cr = ChannelResult.ok( + [{"title": "x"}], __cursor_pending__={"etag": "v2"}, __cursor_source_id__=source.id + ) + emitted = [] + + async def fake_emit(run_id, step, message, level="info", detail=None, elapsed_ms=None): + emitted.append({"run_id": run_id, "step": step, "level": level, "detail": detail}) + + with ( + patch("backend.pipeline.collector.collect", return_value=cr), + patch("backend.pipeline.cursor_store.DBCursorStore") as DB, + patch("backend.pipeline.events.emit", new=fake_emit), + ): + DB.return_value.save = save_mock + result = await run_pipeline( + task.id, source, enable_ai=False, enable_notifications=False, + sink=_ok_sink(), run_id="run-cursor-fail-1", + ) + + assert result.success is True + + warnings = [e for e in emitted if e["level"] == "warning" and e["step"] == "store"] + assert len(warnings) == 1 + assert warnings[0]["detail"]["source_id"] == source.id + assert warnings[0]["detail"]["cursor"] == {"etag": "v2"} diff --git a/tests/unit/worker/test_tasks.py b/tests/unit/worker/test_tasks.py index 7352be6..aa9873c 100644 --- a/tests/unit/worker/test_tasks.py +++ b/tests/unit/worker/test_tasks.py @@ -4,7 +4,11 @@ import pytest -from backend.worker.tasks import _AlertOnRetriesExhaustedTask, run_collection +from backend.worker.tasks import ( + _AlertOnRetriesExhaustedTask, + run_collection, + run_scheduled_collection, +) def test_run_collection_autoretries_on_any_exception(): @@ -15,7 +19,31 @@ def test_run_collection_autoretries_on_any_exception(): boundary to autoretry on any Exception that reaches it.""" assert run_collection.autoretry_for == (Exception,) assert run_collection.max_retries == 3 - assert run_collection.default_retry_delay == 60 + + +# ── AUDIT C14: exponential backoff + jitter on retry (was a fixed 60s) ────── + +def test_run_collection_retries_use_exponential_backoff_and_jitter(): + """A fixed 60s retry delay means every failed run on a source retries in + lockstep; retry_backoff+retry_jitter makes celery compute an increasing, + randomized countdown instead (see celery.app.autoretry.add_autoretry_ + behaviour: countdown = min(retry_backoff_max, 1 * 2**retries), then + full-jittered).""" + assert run_collection.retry_backoff is True + assert run_collection.retry_backoff_max == 600 + assert run_collection.retry_jitter is True + + +def test_run_scheduled_collection_autoretries_with_backoff(): + """run_scheduled_pipeline() funnels into the same run_pipeline() as + run_collection, which only re-raises errors already classified retryable + — so this task needs the identical autoretry_for + backoff contract, not + just a bare task with zero retries (the pre-fix state).""" + assert run_scheduled_collection.autoretry_for == (Exception,) + assert run_scheduled_collection.max_retries == 3 + assert run_scheduled_collection.retry_backoff is True + assert run_scheduled_collection.retry_backoff_max == 600 + assert run_scheduled_collection.retry_jitter is True # ── P1 (test/ops audit): retry-exhausted alert ─────────────────────────────