diff --git a/podcasts/management/commands/refresh_feeds.py b/podcasts/management/commands/refresh_feeds.py index 1c6f601..9a37afb 100644 --- a/podcasts/management/commands/refresh_feeds.py +++ b/podcasts/management/commands/refresh_feeds.py @@ -1,16 +1,15 @@ -import sys - from django.core.management.base import BaseCommand +from django.utils import timezone from podcasts.models import PodcastFeed -from podcasts.views import _refresh_feeds +from podcasts.views import _in_backoff_q, _refresh_feeds class Command(BaseCommand): help = 'Refresh podcast feeds (fetch new episodes from RSS)' def add_arguments(self, parser): - parser.add_argument('--feed-id', type=int, help='Refresh only this feed ID') + parser.add_argument('--feed-id', type=int, help='Refresh only this feed ID (even when backing off)') parser.add_argument('--limit', type=int, default=0, help='Max number of feeds to refresh') def handle(self, *args, **options): @@ -18,6 +17,10 @@ class Command(BaseCommand): if options['feed_id']: qs = qs.filter(pk=options['feed_id']) + else: + # A feed failing 3+ times in a row is tried every 2, 4, 8, 16, + # at most 24 hours instead of every hour. + qs = qs.exclude(_in_backoff_q(timezone.now())) if options['limit']: qs = qs[: options['limit']] diff --git a/podcasts/migrations/0008_feed_http_cache.py b/podcasts/migrations/0008_feed_http_cache.py new file mode 100644 index 0000000..519d989 --- /dev/null +++ b/podcasts/migrations/0008_feed_http_cache.py @@ -0,0 +1,38 @@ +# Generated by Django 6.1 on 2026-10-02 12:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('podcasts', '0007_feed_settings'), + ] + + operations = [ + migrations.AddField( + model_name='podcastfeed', + name='http_etag', + field=models.CharField(blank=True, max_length=500), + ), + migrations.AddField( + model_name='podcastfeed', + name='http_last_modified', + field=models.CharField(blank=True, max_length=100), + ), + migrations.AddField( + model_name='podcastfeed', + name='last_error', + field=models.CharField(blank=True, max_length=500), + ), + migrations.AddField( + model_name='podcastfeed', + name='last_error_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AddField( + model_name='podcastfeed', + name='error_count', + field=models.PositiveIntegerField(default=0), + ), + ] diff --git a/podcasts/models.py b/podcasts/models.py index ff19425..32382a4 100644 --- a/podcasts/models.py +++ b/podcasts/models.py @@ -27,6 +27,15 @@ class PodcastFeed(models.Model): # Always the same as new_episodes == 'queue' (see save()); kept for the # web UI's toggle and older clients. auto_queue = models.BooleanField(default=False) + # HTTP validators of the last full (200) response, sent back as + # If-None-Match / If-Modified-Since so an unchanged feed costs a 304. + http_etag = models.CharField(max_length=500, blank=True) + http_last_modified = models.CharField(max_length=100, blank=True) + # The last failed refresh (network, HTTP status, not a feed); cleared on + # success. error_count counts failures in a row and drives the backoff. + last_error = models.CharField(max_length=500, blank=True) + last_error_at = models.DateTimeField(null=True, blank=True) + error_count = models.PositiveIntegerField(default=0) # Playback, for every device: null = the device's own speed. playback_speed = models.FloatField(null=True, blank=True) diff --git a/podcasts/tests.py b/podcasts/tests.py index 8ed25df..3a12bb6 100644 --- a/podcasts/tests.py +++ b/podcasts/tests.py @@ -7,11 +7,16 @@ from django.contrib.auth.models import User from django.test import TestCase from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings -from .views import _refresh_feed +from .views import FeedFetch, _refresh_feed _parse = feedparser.parse +def _ok(rss, **kw): + """What _fetch_feed returns for a 200 with this RSS.""" + return FeedFetch(200, _parse(rss), **kw) + + def _rss(items): """Items are (guid, date) or (guid, date, extra XML inside the ); the guid is the title too.""" @@ -161,7 +166,7 @@ class AutoQueueTests(TestCase): ]) def refresh(self): - with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): + with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)): _refresh_feed(self.feed) order = PodcastQueue.objects.filter(user=self.user).order_by('position') return [q.episode.guid for q in order] @@ -193,7 +198,7 @@ class FirstFetchTests(TestCase): self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(5)]) def subscribe(self): - with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): + with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)): return self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/f'}), content_type='application/json') @@ -211,7 +216,7 @@ class FirstFetchTests(TestCase): self.subscribe() feed = PodcastFeed.objects.get(user=self.user) self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(8)]) - with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): + with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)): _refresh_feed(feed) inbox = self.client.get('/podcasts/inbox/').json()['episodes'] self.assertEqual([e['title'] for e in inbox], ['e7', 'e6', 'e5', 'e4']) @@ -229,11 +234,11 @@ class RefreshAllTests(TestCase): self.bobs = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A') self.fetched = [] - def fake_fetch(self, url): + def fake_fetch(self, url, **kw): self.fetched.append(url) if url.endswith('/b'): raise ValueError('broken feed') - return _parse(_rss([(url[-1] + '1', 'Mon, 05 Jan 2026 10:00:00 GMT')])) + return _ok(_rss([(url[-1] + '1', 'Mon, 05 Jan 2026 10:00:00 GMT')])) def test_refresh_all_reports_counts_and_skips_recent(self): with patch('podcasts.views._fetch_feed', self.fake_fetch): @@ -323,7 +328,7 @@ class FeedExtrasParsingTests(TestCase): self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed') def refresh(self, rss): - with patch('podcasts.views._fetch_feed', lambda url: _parse(rss)): + with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(rss)): _refresh_feed(self.feed) def test_feedparser_keys(self): @@ -1070,7 +1075,7 @@ class NewEpisodeRuleTests(TestCase): self.feed.refresh_from_db() def refresh(self): - with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)): + with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)): return _refresh_feed(self.feed) def inbox(self): @@ -1252,6 +1257,332 @@ class KeepSortedTests(TestCase): feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/n', title='N', new_episodes='queue', last_refreshed_at=_at(1, 1)) with patch('podcasts.views._fetch_feed', - lambda url: _parse(_rss([('n1', 'Fri, 02 Jan 2026 12:00:00 GMT')]))): + lambda url, **kw: _ok(_rss([('n1', 'Fri, 02 Jan 2026 12:00:00 GMT')]))): _refresh_feed(feed) self.assertEqual(self.queue(), ['e0', 'e2', 'n1', 'e1']) + + +# --------------------------------------------------------------------------- +# Feed refresh: conditional GET, errors, permanent redirects, backoff +# --------------------------------------------------------------------------- + +class _Hop: + def __init__(self, url, status): + self.url, self.status_code = url, status + + +class _HttpResponse: + """Just enough of a requests.Response for _fetch_feed.""" + + def __init__(self, url, status=200, content=b'', headers=None, history=()): + self.url, self.status_code, self.content = url, status, content + self.headers = headers or {} + self.history = list(history) + + +class FetchFeedTests(TestCase): + """_fetch_feed itself, with requests.get faked.""" + + RSS = _rss([('a', 'Mon, 05 Jan 2026 10:00:00 GMT')]).encode() + + def fetch(self, response, url='https://example.org/feed', **kw): + calls = [] + + def fake_get(u, **kwargs): + calls.append(kwargs['headers']) + if isinstance(response, Exception): + raise response + return response + + from .views import _fetch_feed + with patch('podcasts.views.requests.get', fake_get): + result = _fetch_feed(url, **kw) + return result, calls[0] + + def error(self, response): + from .views import FeedFetchError + with self.assertRaises(FeedFetchError) as cm: + self.fetch(response) + return str(cm.exception) + + def test_200_returns_parsed_feed_and_validators(self): + r, headers = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS, headers={ + 'ETag': '"v1"', 'Last-Modified': 'Mon, 05 Jan 2026 10:00:00 GMT'})) + self.assertEqual(r.status, 200) + self.assertEqual([e.title for e in r.parsed.entries], ['a']) + self.assertEqual((r.etag, r.last_modified), ('"v1"', 'Mon, 05 Jan 2026 10:00:00 GMT')) + self.assertEqual(r.permanent_url, 'https://example.org/feed') + self.assertNotIn('If-None-Match', headers) + self.assertNotIn('If-Modified-Since', headers) + + def test_conditional_request_and_304(self): + r, headers = self.fetch(_HttpResponse('https://example.org/feed', status=304), + etag='"v1"', last_modified='Mon, 05 Jan 2026 10:00:00 GMT') + self.assertEqual(headers['If-None-Match'], '"v1"') + self.assertEqual(headers['If-Modified-Since'], 'Mon, 05 Jan 2026 10:00:00 GMT') + self.assertEqual(r.status, 304) + self.assertIsNone(r.parsed) + + def test_304_without_validators_is_an_error(self): + self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=304)), 'HTTP 304') + + def test_errors(self): + import requests + self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=404)), 'HTTP 404') + self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=503)), 'HTTP 503') + self.assertEqual(self.error(requests.Timeout()), 'timeout') + self.assertEqual(self.error(requests.ConnectionError()), 'connection failed') + self.assertEqual(self.error(requests.TooManyRedirects()), 'too many redirects') + self.assertEqual(self.error(_HttpResponse('https://example.org/feed', content=b'

Hi')), + 'not a feed') + self.assertEqual(self.error(_HttpResponse('https://example.org/feed', content=b'\x00\xff garbage')), + 'not a feed') + + def test_permanent_redirect_chain(self): + r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[ + _Hop('https://example.org/feed', 301), _Hop('https://b.org/feed', 308)])) + self.assertEqual(r.permanent_url, 'https://c.org/feed') + self.assertEqual(r.final_url, 'https://c.org/feed') + + def test_temporary_hop_ends_the_permanent_part(self): + r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[ + _Hop('https://example.org/feed', 301), _Hop('https://b.org/feed', 302)])) + self.assertEqual(r.permanent_url, 'https://b.org/feed') + for status in (302, 307): + r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[ + _Hop('https://example.org/feed', status), _Hop('https://b.org/feed', 301)])) + self.assertEqual(r.permanent_url, 'https://example.org/feed') + + +class FeedRefreshTests(TestCase): + """What a refresh stores: validators, 304s, errors, moved feeds.""" + + V = {'etag': '"v1"', 'last_modified': 'Mon, 05 Jan 2026 10:00:00 GMT'} + + def setUp(self): + self.user = User.objects.create_user(username='alice', password='pw12345678') + self.client.force_login(self.user) + self.feed = PodcastFeed.objects.create( + user=self.user, rss_url='https://example.org/feed', title='Feed', + last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + http_etag=self.V['etag'], http_last_modified=self.V['last_modified'], + ) + PodcastEpisode.objects.create(feed=self.feed, guid='old', title='Old', audio_url='https://example.org/old.mp3') + self.calls = [] + self.answer = None + + def fake_fetch(self, url, **kw): + self.calls.append((url, kw)) + if isinstance(self.answer, Exception): + raise self.answer + return self.answer + + def refresh(self, feeds=None): + from .views import _refresh_feeds + with patch('podcasts.views._fetch_feed', self.fake_fetch): + return _refresh_feeds(feeds or [self.feed]) + + def test_304_touches_no_episodes(self): + from .views import FeedFetch + self.feed.error_count, self.feed.last_error = 2, 'timeout' + self.feed.last_error_at = datetime(2026, 1, 2, tzinfo=timezone.utc) + self.feed.save() + self.answer = FeedFetch(304, None, **self.V) + with patch('podcasts.views.PodcastEpisode.objects.get_or_create') as create, \ + patch('podcasts.views.PodcastEpisode.objects.bulk_update') as update: + self.assertEqual(self.refresh(), (1, 0, 0)) + create.assert_not_called() + update.assert_not_called() + self.assertEqual(self.calls, [('https://example.org/feed', self.V)]) + self.feed.refresh_from_db() + self.assertGreater(self.feed.last_refreshed_at, datetime(2026, 1, 1, tzinfo=timezone.utc)) + self.assertEqual((self.feed.error_count, self.feed.last_error, self.feed.last_error_at), (0, '', None)) + self.assertEqual((self.feed.http_etag, self.feed.http_last_modified), (self.V['etag'], self.V['last_modified'])) + self.assertEqual(self.feed.episodes.count(), 1) + + def test_200_stores_new_validators(self): + self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"', last_modified='') + self.assertEqual(self.refresh(), (1, 0, 1)) + self.feed.refresh_from_db() + self.assertEqual((self.feed.http_etag, self.feed.http_last_modified), ('"v2"', '')) + + def test_first_fetch_is_never_conditional(self): + self.feed.last_refreshed_at = None + self.feed.save() + self.answer = _ok(_rss([])) + self.refresh() + self.assertEqual(self.calls, [('https://example.org/feed', {})]) + # Nothing stored yet either: unconditional. + self.calls.clear() + new = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/n', title='N') + self.refresh([new]) + self.assertEqual(self.calls, [('https://example.org/n', {})]) + + def test_shared_url_validators_only_when_all_agree(self): + from .views import FeedFetch + bob = User.objects.create_user(username='bob', password='pw12345678') + bobs = PodcastFeed.objects.create( + user=bob, rss_url='https://example.org/feed', title='Feed', + last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + http_etag=self.V['etag'], http_last_modified=self.V['last_modified'], + ) + self.answer = FeedFetch(304, None, **self.V) + self.assertEqual(self.refresh([self.feed, bobs]), (2, 0, 0)) + self.assertEqual(self.calls, [('https://example.org/feed', self.V)]) + + # Bob's copy has other validators: one full download for both. + bobs.http_etag = '"v0"' + bobs.save() + self.calls.clear() + self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"') + self.assertEqual(self.refresh([self.feed, bobs]), (2, 0, 2)) + self.assertEqual(self.calls, [('https://example.org/feed', {})]) + self.assertEqual(set(PodcastFeed.objects.values_list('http_etag', flat=True)), {'"v2"'}) + + # Carol just subscribed (never fetched): a full download as well. + carol = User.objects.create_user(username='carol', password='pw12345678') + carols = PodcastFeed.objects.create(user=carol, rss_url='https://example.org/feed', title='Feed', + http_etag='"v2"') + self.calls.clear() + self.refresh([self.feed, PodcastFeed.objects.get(pk=bobs.pk), carols]) + self.assertEqual(self.calls, [('https://example.org/feed', {})]) + self.assertEqual(PodcastEpisode.objects.filter(feed=carols).count(), 1) + + def test_errors_are_counted_and_cleared(self): + from .views import FeedFetchError + self.answer = FeedFetchError('HTTP 404') + self.assertEqual(self.refresh(), (0, 1, 0)) + self.feed.refresh_from_db() + self.assertEqual(self.refresh([self.feed]), (0, 1, 0)) + self.feed.refresh_from_db() + self.assertEqual((self.feed.last_error, self.feed.error_count), ('HTTP 404', 2)) + self.assertIsNotNone(self.feed.last_error_at) + # Unchanged otherwise: still the last good refresh. + self.assertEqual(self.feed.last_refreshed_at, datetime(2026, 1, 1, tzinfo=timezone.utc)) + + feeds = self.client.get('/podcasts/feeds/').json()['feeds'] + self.assertEqual(feeds[0]['last_error'], 'HTTP 404') + self.assertEqual(feeds[0]['error_count'], 2) + self.assertEqual(feeds[0]['last_error_at'], self.feed.last_error_at.isoformat()) + state = self.client.get('/podcasts/state/').json()['feeds'] + self.assertEqual(state[0]['error_count'], 2) + + self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')])) + self.assertEqual(self.refresh([self.feed]), (1, 0, 1)) + feeds = self.client.get('/podcasts/feeds/').json()['feeds'] + self.assertEqual((feeds[0]['last_error'], feeds[0]['last_error_at'], feeds[0]['error_count']), ('', None, 0)) + + def test_refresh_now_reports_and_stores_the_error(self): + from .views import FeedFetchError + self.answer = FeedFetchError('not a feed') + with patch('podcasts.views._fetch_feed', self.fake_fetch): + r = self.client.post('/podcasts/feeds/refresh/', json.dumps({'feed_id': self.feed.id}), + content_type='application/json') + self.assertEqual(r.status_code, 502) + self.assertEqual(r.json()['error'], 'not a feed') + self.assertEqual(self.calls, [('https://example.org/feed', self.V)]) + self.feed.refresh_from_db() + self.assertEqual((self.feed.last_error, self.feed.error_count), ('not a feed', 1)) + + def test_failed_subscription_is_not_kept(self): + from .views import FeedFetchError + self.answer = FeedFetchError('HTTP 404') + with patch('podcasts.views._fetch_feed', self.fake_fetch): + r = self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/gone'}), + content_type='application/json') + self.assertEqual(r.status_code, 400) + self.assertIn('HTTP 404', r.json()['error']) + self.assertEqual(self.calls, [('https://example.org/gone', {})]) + self.assertFalse(PodcastFeed.objects.filter(rss_url='https://example.org/gone').exists()) + + def test_permanent_redirect_moves_the_subscription(self): + self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), + final_url='https://new.org/feed', permanent_url='https://new.org/feed') + self.assertEqual(self.refresh(), (1, 0, 1)) + self.feed.refresh_from_db() + self.assertEqual(self.feed.rss_url, 'https://new.org/feed') + self.assertEqual(self.feed.episodes.count(), 2) + + def test_permanent_redirect_to_a_url_the_user_has_already(self): + other = PodcastFeed.objects.create(user=self.user, rss_url='https://new.org/feed', title='Other') + self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"', + final_url='https://new.org/feed', permanent_url='https://new.org/feed') + self.assertEqual(self.refresh(), (1, 0, 1)) + self.feed.refresh_from_db() + self.assertEqual(self.feed.rss_url, 'https://example.org/feed') + self.assertEqual((self.feed.error_count, self.feed.http_etag), (0, '"v2"')) + self.assertEqual(PodcastFeed.objects.get(pk=other.pk).rss_url, 'https://new.org/feed') + + def test_temporary_redirect_keeps_the_url(self): + self.answer = _ok(_rss([]), final_url='https://tmp.org/feed', permanent_url='https://example.org/feed') + self.refresh() + self.feed.refresh_from_db() + self.assertEqual(self.feed.rss_url, 'https://example.org/feed') + + +class FeedBackoffTests(TestCase): + """The hourly job tries a feed failing 3+ times in a row less often.""" + + def setUp(self): + from django.utils import timezone as dj_tz + self.user = User.objects.create_user(username='alice', password='pw12345678') + self.client.force_login(self.user) + now = dj_tz.now() + long_ago = datetime(2026, 1, 1, tzinfo=timezone.utc) + # (name, error_count, hours since the last error) -> due in the hourly job? + self.cases = { + 'ok': (0, None, True), + 'two': (2, 0.2, True), + 'three-early': (3, 1, False), # waits 2 h + 'three-due': (3, 2.5, True), + 'four-early': (4, 3, False), # waits 4 h + 'four-due': (4, 4.5, True), + 'six-early': (6, 15, False), # waits 16 h + 'six-due': (6, 17, True), + 'ten-early': (10, 23, False), # capped at 24 h + 'ten-due': (10, 25, True), + } + self.feeds = {} + for name, (count, hours, _) in self.cases.items(): + self.feeds[name] = PodcastFeed.objects.create( + user=self.user, rss_url=f'https://example.org/{name}', title=name, + last_refreshed_at=long_ago, error_count=count, last_error='timeout' if count else '', + last_error_at=now - timedelta(hours=hours) if hours is not None else None, + ) + self.fetched = [] + + def fake_fetch(self, url, **kw): + self.fetched.append(url.rsplit('/', 1)[1]) + return _ok(_rss([])) + + def run_command(self, *args): + from io import StringIO + from django.core.management import call_command + out = StringIO() + with patch('podcasts.views._fetch_feed', self.fake_fetch): + call_command('refresh_feeds', *args, stdout=out) + return out.getvalue() + + def test_hourly_job_skips_feeds_in_backoff(self): + out = self.run_command() + due = {name for name, (_, _, is_due) in self.cases.items() if is_due} + self.assertEqual(set(self.fetched), due) + self.assertIn(f'{len(due)} feeds refreshed, 0 failed', out) + + def test_explicit_feed_id_ignores_backoff(self): + self.run_command('--feed-id', str(self.feeds['ten-early'].id)) + self.assertEqual(self.fetched, ['ten-early']) + + def test_user_refresh_all_tries_them_but_respects_max_age(self): + from django.utils import timezone as dj_tz + recent = self.feeds['three-early'] + recent.last_refreshed_at = dj_tz.now() - timedelta(seconds=60) + recent.save() + with patch('podcasts.views._fetch_feed', self.fake_fetch): + r = self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json') + self.assertEqual(set(self.fetched), set(self.cases) - {'three-early'}) + self.assertEqual(r.json()['refreshed'], len(self.cases) - 1) + + def test_backoff_hours(self): + from .views import _backoff_hours + self.assertEqual([_backoff_hours(n) for n in range(0, 9)], [0, 0, 0, 2, 4, 8, 16, 24, 24]) diff --git a/podcasts/views.py b/podcasts/views.py index 952ca6e..7e0df46 100644 --- a/podcasts/views.py +++ b/podcasts/views.py @@ -2,12 +2,14 @@ import json import logging import urllib.parse from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass from datetime import timedelta import xml.etree.ElementTree as ET import feedparser import requests from django.conf import settings +from django.db import IntegrityError, transaction from django.db.models import ( Case, Count, Exists, F, FilteredRelation, IntegerField, Max, OuterRef, Q, Sum, Value, When, ) @@ -264,11 +266,141 @@ EXTRA_FIELDS = ['description', 'transcripts', 'chapters_url', 'chapters', 'chapt # Shared feed refresh helper # --------------------------------------------------------------------------- -def _fetch_feed(url): - """Download and parse a feed (with a timeout, unlike feedparser's own fetch).""" - r = requests.get(url, timeout=FEED_TIMEOUT, headers={'User-Agent': 'diora podcast refresh'}) - r.raise_for_status() - return feedparser.parse(r.content, response_headers={'content-location': r.url, **r.headers}) +class FeedFetchError(Exception): + """A feed that could not be fetched or is not a feed; str() is the short + message stored in PodcastFeed.last_error ("HTTP 404", "timeout", …).""" + + +@dataclass +class FeedFetch: + """What _fetch_feed got. status 200: `parsed` is the feedparser result and + etag/last_modified are the new validators ('' when the server sent none). + status 304: unchanged since the validators that were sent, parsed is None. + final_url is where the content came from after all redirects; + permanent_url is the requested URL after following only the 301/308 hops + at the start of the redirect chain (== the requested URL without one).""" + status: int + parsed: object = None + etag: str = '' + last_modified: str = '' + final_url: str = '' + permanent_url: str = '' + + +PERMANENT_REDIRECTS = (301, 308) + + +def _fetch_feed(url, etag='', last_modified=''): + """Download and parse a feed (with a timeout, unlike feedparser's own + fetch). With etag/last_modified it is a conditional GET. Returns a + FeedFetch; raises FeedFetchError when there is nothing usable.""" + headers = {'User-Agent': 'diora podcast refresh'} + if etag: + headers['If-None-Match'] = etag + if last_modified: + headers['If-Modified-Since'] = last_modified + try: + r = requests.get(url, timeout=FEED_TIMEOUT, headers=headers) + except requests.Timeout: + raise FeedFetchError('timeout') + except requests.TooManyRedirects: + raise FeedFetchError('too many redirects') + except requests.ConnectionError: + raise FeedFetchError('connection failed') + except requests.RequestException as e: + raise FeedFetchError(f'request failed ({type(e).__name__})') + + # Each hop's target is the next hop's URL (the last one's is r.url). + permanent_url = url + targets = [h.url for h in r.history[1:]] + [r.url] + for hop, target in zip(r.history, targets): + if hop.status_code not in PERMANENT_REDIRECTS: + break + permanent_url = target + + if r.status_code == 304: + if not (etag or last_modified): + raise FeedFetchError('HTTP 304') # nothing to be unchanged from + return FeedFetch(304, None, etag, last_modified, r.url, permanent_url) + if r.status_code >= 400: + raise FeedFetchError(f'HTTP {r.status_code}') + + parsed = feedparser.parse(r.content, response_headers={'content-location': r.url, **r.headers}) + if parsed.get('bozo') and not parsed.entries and not parsed.feed.get('title'): + raise FeedFetchError('not a feed') + new_etag = r.headers.get('ETag', '') + new_modified = r.headers.get('Last-Modified', '') + return FeedFetch( + r.status_code, parsed, + # A validator too long to store is dropped (a truncated one never matches). + new_etag if len(new_etag) <= 500 else '', + new_modified if len(new_modified) <= 100 else '', + r.url, permanent_url, + ) + + +def _validators(feeds): + """If-None-Match / If-Modified-Since for one download that serves all + `feeds` (several users' subscriptions of the same URL): only when every + one of them was fetched before and stored the same validators, because a + 304 then skips all of them. Otherwise {} (a full response).""" + pairs = {(f.http_etag, f.http_last_modified) for f in feeds} + if len(pairs) != 1 or any(f.last_refreshed_at is None for f in feeds): + return {} + etag, last_modified = pairs.pop() + if not (etag or last_modified): + return {} + return {'etag': etag, 'last_modified': last_modified} + + +def _error_text(e): + if isinstance(e, FeedFetchError): + return str(e)[:500] + return (str(e) or type(e).__name__)[:500] + + +def _record_error(feed_obj, e): + feed_obj.last_error = _error_text(e) + feed_obj.last_error_at = timezone.now() + feed_obj.error_count += 1 + try: + feed_obj.save(update_fields=['last_error', 'last_error_at', 'error_count']) + except Exception: + log.exception('could not store the refresh error of feed %s', feed_obj.pk) + + +ERROR_FIELDS = ['last_error', 'last_error_at', 'error_count'] + + +def _clear_error(feed_obj): + feed_obj.last_error = '' + feed_obj.last_error_at = None + feed_obj.error_count = 0 + + +BACKOFF_AFTER = 3 # failures in a row before the hourly job backs off +BACKOFF_MAX_HOURS = 24 + + +def _backoff_hours(error_count): + """Hours between hourly-job attempts for a feed failing error_count + times in a row: 2, 4, 8, 16, 24 … from the third failure on.""" + if error_count < BACKOFF_AFTER: + return 0 + return min(2 ** (error_count - 2), BACKOFF_MAX_HOURS) + + +def _in_backoff_q(now=None): + """Q for feeds that failed too often to be tried again yet.""" + now = now or timezone.now() + n = BACKOFF_AFTER + while _backoff_hours(n) < BACKOFF_MAX_HOURS: + n += 1 + # From n failures on the wait is capped; below, one term per count. + q = Q(error_count__gte=n, last_error_at__gt=now - timedelta(hours=BACKOFF_MAX_HOURS)) + for count in range(BACKOFF_AFTER, n): + q |= Q(error_count=count, last_error_at__gt=now - timedelta(hours=_backoff_hours(count))) + return q def _refresh_feeds(feeds, chunk=32): @@ -279,27 +411,31 @@ def _refresh_feeds(feeds, chunk=32): refreshed = failed = new = 0 for start in range(0, len(feeds), chunk): part = feeds[start:start + chunk] - urls = sorted({f.rss_url for f in part}) + by_url = {} + for f in part: + by_url.setdefault(f.rss_url, []).append(f) results = {} def fetch(url): try: - return url, _fetch_feed(url), None + return url, _fetch_feed(url, **_validators(by_url[url])), None except Exception as e: # network, HTTP or parser trouble: report and go on return url, None, e + urls = sorted(by_url) with ThreadPoolExecutor(max_workers=min(FEED_WORKERS, len(urls) or 1)) as pool: - for url, parsed, err in pool.map(fetch, urls): - results[url] = (parsed, err) + for url, fetched, err in pool.map(fetch, urls): + results[url] = (fetched, err) for feed in part: - parsed, err = results[feed.rss_url] + fetched, err = results[feed.rss_url] if err is not None: failed += 1 log.warning('refreshing %s failed: %s', feed.rss_url, err) + _record_error(feed, err) continue try: - new += _refresh_feed(feed, parsed) + new += _refresh_feed(feed, fetched) refreshed += 1 except Exception as e: failed += 1 @@ -307,10 +443,47 @@ def _refresh_feeds(feeds, chunk=32): return refreshed, failed, new -def _refresh_feed(feed_obj, parsed=None): - """Parse rss_url and upsert episodes. Returns count of new episodes.""" - if parsed is None: - parsed = _fetch_feed(feed_obj.rss_url) +def _refresh_feed(feed_obj, fetched=None): + """Fetch rss_url (unless `fetched`, a FeedFetch, is given) and upsert + episodes. Returns count of new episodes. A failure is stored on the feed + (last_error, error_count) and raised again.""" + if fetched is None: + try: + fetched = _fetch_feed(feed_obj.rss_url, **_validators([feed_obj])) + except Exception as e: + _record_error(feed_obj, e) + raise + try: + return _apply_feed(feed_obj, fetched) + except Exception as e: + _record_error(feed_obj, e) + raise + + +def _move_to_permanent_url(feed_obj, new_url): + """The feed answered with a permanent redirect: subscribe to the new URL, + unless the user already has that one too (then nothing changes).""" + if not new_url or new_url == feed_obj.rss_url or len(new_url) > 1000: + return + try: + with transaction.atomic(): + PodcastFeed.objects.filter(pk=feed_obj.pk).update(rss_url=new_url) + except IntegrityError: # the user's other subscription has it + return + log.info('feed %s moved permanently: %s -> %s', feed_obj.pk, feed_obj.rss_url, new_url) + feed_obj.rss_url = new_url + + +def _apply_feed(feed_obj, fetched): + now = timezone.now() + if fetched.status == 304: + # Unchanged since the last full download: nothing to parse or write. + feed_obj.last_refreshed_at = now + _clear_error(feed_obj) + feed_obj.save(update_fields=['last_refreshed_at', *ERROR_FIELDS]) + return 0 + + parsed = fetched.parsed channel = parsed.feed # The first fetch of a new subscription brings the whole back catalogue. first_fetch = feed_obj.last_refreshed_at is None @@ -328,7 +501,7 @@ def _refresh_feed(feed_obj, parsed=None): or image.get('url', '') )[:1000] - feed_obj.last_refreshed_at = timezone.now() + feed_obj.last_refreshed_at = now feed_obj.save() max_ep = getattr(settings, 'PODCAST_MAX_EPISODES_PER_FEED', 200) @@ -456,6 +629,13 @@ def _refresh_feed(feed_obj, parsed=None): ordered = newest_first if where != PodcastSettings.BACK else newest_first[::-1] _enqueue(feed_obj.user, [e.id for e in ordered], where) + # Only now, with every episode written: the validators (a 304 next time + # skips all of the above) and a clean error state. + feed_obj.http_etag = fetched.etag + feed_obj.http_last_modified = fetched.last_modified + _clear_error(feed_obj) + feed_obj.save(update_fields=['http_etag', 'http_last_modified', *ERROR_FIELDS]) + _move_to_permanent_url(feed_obj, fetched.permanent_url) return new_count @@ -517,13 +697,15 @@ def _feeds_data(user): user.podcast_feeds .annotate(latest_episode_at=Max('episodes__pub_date')) .values('id', 'title', 'artwork_url', 'rss_url', 'last_refreshed_at', 'author', 'added_at', - 'latest_episode_at', *FEED_SETTING_FIELDS) + 'latest_episode_at', 'last_error', 'last_error_at', 'error_count', *FEED_SETTING_FIELDS) ) for f in feeds: f['feed_id'] = f['id'] # as in feeds//settings/ if f['last_refreshed_at']: f['last_refreshed_at'] = f['last_refreshed_at'].isoformat() + if f['last_error_at']: + f['last_error_at'] = f['last_error_at'].isoformat() if f['added_at']: f['added_at'] = f['added_at'].isoformat() if f['latest_episode_at']: @@ -1120,7 +1302,8 @@ def refresh_feed_now(request): def refresh_all_feeds(request): """Refresh all of the user's feeds on the server in one go. Feeds the hourly job (or an earlier call) refreshed within `max_age` seconds are - skipped. Clients then fetch the result once (state/, inbox/ …).""" + skipped. Clients then fetch the result once (state/, inbox/ …). + The user asked, so feeds the hourly job backs off from are tried too.""" if not request.user.is_authenticated: return JsonResponse({'error': 'authentication required'}, status=401) try: