From 5259caa846b2c41cb2b3c34baefbc5cc98467670 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 5 Nov 2026 19:13:46 -0500 Subject: [PATCH] fix: dead external links are retried sparingly, not three times a day The daily recovery reset every dead link to a fresh three-attempt budget, so a link that can never succeed (an archive too big to import inside the fetch's hard limit, a file made private) was downloaded up to three times a day, every day. Recovery now gives a dead link one more attempt once it has been dead for 2**recoveries days (1, 2, 4, 8, 16), and stops after five. A recovered link that fails again goes straight back to dead. Retention prunes only links recovery has given up on, 30 days after their last attempt. A manual refetch of a post restores the full budgets. Migration 0118 adds external_link.recoveries. Co-Authored-By: Claude Opus 5.5 --- .../versions/0118_external_link_recoveries.py | 32 +++++++++ backend/app/models/external_link.py | 6 ++ backend/app/services/external_links.py | 7 +- backend/app/tasks/external.py | 49 +++++++++---- tests/test_external_worker.py | 71 +++++++++++++++++++ 5 files changed, 150 insertions(+), 15 deletions(-) create mode 100644 alembic/versions/0118_external_link_recoveries.py diff --git a/alembic/versions/0118_external_link_recoveries.py b/alembic/versions/0118_external_link_recoveries.py new file mode 100644 index 0000000..6f21a48 --- /dev/null +++ b/alembic/versions/0118_external_link_recoveries.py @@ -0,0 +1,32 @@ +"""external_link.recoveries: how many times recovery has retried a dead link. + +The daily recovery gives a dead link one more attempt, spaced further apart +each time, and leaves it dead once the count reaches its limit. A link that +can never succeed is then downloaded a handful of times in total, not every +day. + +Revision ID: 0118 +Revises: 0117 +Create Date: 2026-11-05 + +""" +import sqlalchemy as sa +from alembic import op + +revision = "0118" +down_revision = "0117" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "external_link", + sa.Column( + "recoveries", sa.Integer(), nullable=False, server_default=sa.text("0") + ), + ) + + +def downgrade() -> None: + op.drop_column("external_link", "recoveries") diff --git a/backend/app/models/external_link.py b/backend/app/models/external_link.py index b83aa8a..4647959 100644 --- a/backend/app/models/external_link.py +++ b/backend/app/models/external_link.py @@ -80,6 +80,12 @@ class ExternalLink(Base): Integer, nullable=False, server_default=text("0") ) last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + # Times the daily recovery has given this link another attempt after it + # went dead. Recovery stops at a limit, so a link that can never succeed + # isn't downloaded forever. + recoveries: Mapped[int] = mapped_column( + Integer, nullable=False, server_default=text("0") + ) attachment_id: Mapped[int | None] = mapped_column( ForeignKey("post_attachment.id", ondelete="SET NULL"), nullable=True ) diff --git a/backend/app/services/external_links.py b/backend/app/services/external_links.py index 2530e56..ed691c9 100644 --- a/backend/app/services/external_links.py +++ b/backend/app/services/external_links.py @@ -21,9 +21,9 @@ log = logging.getLogger(__name__) def refetch_links_for_post(session: Session, post_id: int) -> dict: - """Reset every settled ExternalLink on a post to pending (fresh attempt - budget) and dispatch its fetch. In-flight ('downloading') links are left - alone. Commits. Returns {links_total, links_reset}.""" + """Reset every settled ExternalLink on a post to pending (fresh attempt and + recovery budgets) and dispatch its fetch. In-flight ('downloading') links + are left alone. Commits. Returns {links_total, links_reset}.""" links = session.execute( select(ExternalLink).where(ExternalLink.post_id == post_id) ).scalars().all() @@ -33,6 +33,7 @@ def refetch_links_for_post(session: Session, post_id: int) -> dict: continue link.status = "pending" link.attempts = 0 + link.recoveries = 0 link.last_error = None link.completed_at = None reset_ids.append(link.id) diff --git a/backend/app/tasks/external.py b/backend/app/tasks/external.py index a3ac844..a99dacc 100644 --- a/backend/app/tasks/external.py +++ b/backend/app/tasks/external.py @@ -39,14 +39,20 @@ log = logging.getLogger(__name__) IMAGES_ROOT = Path("/images") # After this many failed attempts a link is dead-lettered (skipped by routine -# sweeps; an operator recovery still re-attempts). Mirrors the ingester ledger. +# sweeps; the daily recovery retries it, sparingly). Mirrors the ingester ledger. DEAD_LETTER_THRESHOLD = 3 # Per-fetch read + total budgets now live in external_fetch (a short read # timeout fails a stalled host fast; a generous total caps a big-but-flowing # download). The celery soft/hard time_limit is the outer backstop above those. # Links enqueued per sweep — bounds the burst when a big backfill records many. _SWEEP_BATCH = 50 -# Dead rows older than this are pruned (retention). +# Recovery gives a dead link one more attempt once it has been dead for +# 2**recoveries days (1, 2, 4, 8, 16), and stops after _MAX_RECOVERIES. A host +# outage gets several later chances; a link that can never succeed (too big to +# import in time, a file made private) costs a handful of downloads, not three +# a day. +_MAX_RECOVERIES = 5 +# A link recovery has given up on is pruned this long after its last attempt. _RETENTION_DAYS = 30 # Per-host serialize lock. TTL is a safety net for a worker that dies holding it # (normal completion/error releases it in `finally`); sized just past the fetch @@ -404,29 +410,48 @@ def sweep_external_links() -> dict: @celery.task(name="backend.app.tasks.external.recover_external_links") def recover_external_links() -> dict: - """Recovery sweep (rule #89): reset dead links back to retryable so a stuck - host outage or a since-fixed bug gets another pass, then enqueue. Bounded.""" + """Recovery sweep: give each dead link one more attempt once it + has been dead for 2**recoveries days, then enqueue. A recovered link that + fails again goes straight back to dead, and after _MAX_RECOVERIES it stays + dead until retention prunes it.""" SessionLocal = _sync_session_factory() with SessionLocal() as session: - session.execute( - update(ExternalLink).where(ExternalLink.status == "dead") - .values(status="failed", attempts=0, last_error=None) - ) + now = datetime.now(UTC) + candidates = session.execute( + select(ExternalLink).where( + ExternalLink.status == "dead", + ExternalLink.recoveries < _MAX_RECOVERIES, + ) + ).scalars().all() + recovered = 0 + for link in candidates: + wait = timedelta(days=2 ** link.recoveries) + if link.completed_at is not None and link.completed_at > now - wait: + continue + link.status = "failed" + link.attempts = DEAD_LETTER_THRESHOLD - 1 + link.recoveries += 1 + recovered += 1 session.commit() - return sweep_external_links() + return {**sweep_external_links(), "revived": recovered} @celery.task(name="backend.app.tasks.external.prune_external_links") def prune_external_links() -> dict: - """Retention (rule #89): delete long-dead links so the ledger doesn't grow - unbounded. Downloaded links are kept (they're the record of what we have).""" + """Retention: delete links recovery has given up on, once their + last attempt is _RETENTION_DAYS old, so the ledger doesn't grow unbounded. + Downloaded links are kept (they're the record of what we have).""" cutoff = datetime.now(UTC) - timedelta(days=_RETENTION_DAYS) SessionLocal = _sync_session_factory() with SessionLocal() as session: res = session.execute( delete(ExternalLink).where( ExternalLink.status == "dead", - ExternalLink.created_at < cutoff, + ExternalLink.recoveries >= _MAX_RECOVERIES, + or_( + ExternalLink.completed_at.is_(None), + ExternalLink.completed_at < cutoff, + ), ) ) session.commit() diff --git a/tests/test_external_worker.py b/tests/test_external_worker.py index 01f7b49..dc661e8 100644 --- a/tests/test_external_worker.py +++ b/tests/test_external_worker.py @@ -402,3 +402,74 @@ def test_sweep_recovers_a_link_whose_worker_died(db_sync, monkeypatch): row = db_sync.get(ExternalLink, lid) assert (row.status, row.attempts) == ("failed", 1) assert db_sync.get(ExternalLink, fresh_id).status == "downloading" + + +def test_recovery_gives_a_dead_link_one_more_attempt_at_growing_intervals(db_sync, monkeypatch): + """A dead link is retried once after 2**recoveries days, not given a fresh + three-attempt budget, so a link that always fails isn't downloaded daily.""" + from datetime import UTC, datetime, timedelta + + now = datetime.now(UTC) + _, due = _seed(db_sync, host="mega", status="dead") + _, waiting = _seed(db_sync, host="gdrive", status="dead") + _, spent = _seed(db_sync, host="dropbox", status="dead") + due.attempts = waiting.attempts = spent.attempts = ext.DEAD_LETTER_THRESHOLD + due.recoveries, due.completed_at = 2, now - timedelta(days=4, hours=1) + waiting.recoveries, waiting.completed_at = 2, now - timedelta(days=3) + spent.recoveries, spent.completed_at = ext._MAX_RECOVERIES, now - timedelta(days=90) + db_sync.commit() + due_id, waiting_id, spent_id = due.id, waiting.id, spent.id + + enqueued = [] + monkeypatch.setattr(ext.fetch_external_link, "delay", lambda lid: enqueued.append(lid)) + out = ext.recover_external_links() + + assert out["revived"] == 1 + assert enqueued == [due_id] + db_sync.expire_all() + row = db_sync.get(ExternalLink, due_id) + assert (row.status, row.attempts, row.recoveries) == ( + "failed", ext.DEAD_LETTER_THRESHOLD - 1, 3, + ) + assert db_sync.get(ExternalLink, waiting_id).status == "dead" + assert db_sync.get(ExternalLink, spent_id).status == "dead" + + +def test_a_recovered_link_that_fails_again_goes_straight_back_to_dead( + db_sync, tmp_path, monkeypatch, +): + _, link = _seed(db_sync, host="mega", status="dead") + link.attempts = ext.DEAD_LETTER_THRESHOLD + db_sync.commit() + link_id = link.id + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + monkeypatch.setattr(ext, "fetch_external", lambda *a, **k: FetchResult(error="too big")) + monkeypatch.setattr(ext.fetch_external_link, "delay", ext.fetch_external_link) + + ext.recover_external_links() + + db_sync.expire_all() + row = db_sync.get(ExternalLink, link_id) + assert (row.status, row.recoveries) == ("dead", 1) + + +def test_prune_removes_only_links_recovery_has_given_up_on(db_sync): + from datetime import UTC, datetime, timedelta + + old = datetime.now(UTC) - timedelta(days=ext._RETENTION_DAYS + 1) + _, spent = _seed(db_sync, host="mega", status="dead") + _, recovering = _seed(db_sync, host="gdrive", status="dead") + _, recent = _seed(db_sync, host="dropbox", status="dead") + spent.recoveries, spent.completed_at = ext._MAX_RECOVERIES, old + recovering.recoveries, recovering.completed_at = 1, old + recent.recoveries = ext._MAX_RECOVERIES + recent.completed_at = datetime.now(UTC) - timedelta(days=1) + db_sync.commit() + spent_id, recovering_id, recent_id = spent.id, recovering.id, recent.id + + assert ext.prune_external_links() == {"pruned": 1} + db_sync.expire_all() + assert db_sync.get(ExternalLink, spent_id) is None + assert db_sync.get(ExternalLink, recovering_id) is not None + assert db_sync.get(ExternalLink, recent_id) is not None