diff --git a/podcasts/migrations/0017_feed_body_hash.py b/podcasts/migrations/0017_feed_body_hash.py new file mode 100644 index 0000000..71c2641 --- /dev/null +++ b/podcasts/migrations/0017_feed_body_hash.py @@ -0,0 +1,18 @@ +# Generated by Django 6.1 on 2026-10-04 08:13 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('podcasts', '0016_playerstate_sleep'), + ] + + operations = [ + migrations.AddField( + model_name='podcastfeed', + name='http_body_hash', + field=models.CharField(blank=True, max_length=64), + ), + ] diff --git a/podcasts/models.py b/podcasts/models.py index 45fcbc2..87085bf 100644 --- a/podcasts/models.py +++ b/podcasts/models.py @@ -36,6 +36,10 @@ class PodcastFeed(models.Model): # 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) + # Fingerprint of the last feed body that was read: many servers ignore the + # validators above and send the whole feed every time; the same body again + # then counts as unchanged and isn't parsed (views._body_hash). + http_body_hash = models.CharField(max_length=64, 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) diff --git a/podcasts/tests.py b/podcasts/tests.py index 3998d5b..37d9207 100644 --- a/podcasts/tests.py +++ b/podcasts/tests.py @@ -307,7 +307,7 @@ class RefreshAllTests(TestCase): def test_refresh_all_reports_counts_and_skips_recent(self): with patch('podcasts.views._fetch_feed', self.fake_fetch): r = self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json') - self.assertEqual(r.json(), {'ok': True, 'refreshed': 1, 'failed': 1, 'new_episodes': 1}) + self.assertEqual(r.json(), {'ok': True, 'refreshed': 1, 'failed': 1, 'new_episodes': 1, 'done': True, 'left': 0}) self.assertEqual(sorted(self.fetched), ['https://example.org/a', 'https://example.org/b']) # A was just refreshed: a second call within max_age only retries B. self.fetched.clear() @@ -315,6 +315,27 @@ class RefreshAllTests(TestCase): self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json') self.assertEqual(self.fetched, ['https://example.org/b']) + def test_refresh_stops_at_its_deadline(self): + import time as _time + from .views import _refresh_feeds + + def slow(url, **kw): + _time.sleep(.5) + return self.fake_fetch(url, **kw) + with patch('podcasts.views._fetch_feed', slow): + refreshed, failed, new = _refresh_feeds([self.a, self.b], deadline=_time.monotonic() + .05) + self.assertEqual((refreshed, failed), (0, 0)) + self.a.refresh_from_db() + self.assertIsNone(self.a.last_refreshed_at) # not done: left as it was + + def test_the_body_fingerprint_is_kept_and_sent_again(self): + from .views import _validators + with patch('podcasts.views._fetch_feed', lambda url, **kw: FeedFetch(200, _parse(_rss([('x', 'Mon, 05 Jan 2026 10:00:00 GMT')])), body_hash='abc')): + self.client.post('/podcasts/feeds/refresh-all/', '{"max_age":0}', content_type='application/json') + self.a.refresh_from_db() + self.assertEqual(self.a.http_body_hash, 'abc') + self.assertEqual(_validators([self.a]), {'body_hash': 'abc'}) + def test_a_shared_feed_url_is_fetched_once(self): from .views import _refresh_feeds with patch('podcasts.views._fetch_feed', self.fake_fetch): @@ -1491,6 +1512,25 @@ class FetchFeedTests(TestCase): self.fetch(response) return str(cm.exception) + def test_the_same_body_again_counts_as_unchanged_without_parsing(self): + first, _ = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS)) + self.assertEqual(first.status, 200) + self.assertTrue(first.body_hash) + # The server ignores validators and sends the feed again: a 304, unparsed. + with patch('podcasts.views.feedparser.parse') as parse: + again, _ = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS), body_hash=first.body_hash) + parse.assert_not_called() + self.assertEqual((again.status, again.parsed, again.body_hash), (304, None, first.body_hash)) + # Only the build date or a comment differs: still the same. + stamped = self.RSS.replace(b'', b'Sun, 04 Oct 2026 08:00:00 GMT', 1) + same, _ = self.fetch(_HttpResponse('https://example.org/feed', content=stamped), body_hash=first.body_hash) + self.assertEqual(same.status, 304) + # A new episode: parsed. + newer = _rss([('b', 'Tue, 06 Jan 2026 10:00:00 GMT'), ('a', 'Mon, 05 Jan 2026 10:00:00 GMT')]).encode() + changed, _ = self.fetch(_HttpResponse('https://example.org/feed', content=newer), body_hash=first.body_hash) + self.assertEqual(changed.status, 200) + self.assertNotEqual(changed.body_hash, first.body_hash) + 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'})) diff --git a/podcasts/views.py b/podcasts/views.py index 3110930..158aa06 100644 --- a/podcasts/views.py +++ b/podcasts/views.py @@ -1,10 +1,12 @@ +import hashlib import json import logging import math +import re import time import urllib.parse import uuid -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout, as_completed from dataclasses import dataclass from datetime import timedelta, timezone as dt_timezone import xml.etree.ElementTree as ET @@ -37,7 +39,8 @@ log = logging.getLogger(__name__) FEED_TIMEOUT = 20 # seconds per feed download NEW_ON_SUBSCRIBE = 1 # episodes of a new subscription that count as new -FEED_WORKERS = 8 # feeds downloaded at the same time +FEED_WORKERS = 16 # feeds downloaded at the same time +REFRESH_WAIT = 40 # seconds a refresh-all (⟳) waits; feeds not done by then wait for the hourly job MAX_DESCRIPTION = 50_000 # characters of shownotes HTML kept per episode MAX_TRANSCRIPTS = 10 @@ -302,14 +305,27 @@ class FeedFetch: last_modified: str = '' final_url: str = '' permanent_url: str = '' + body_hash: str = '' + + +# What changes in a feed on every request without anything in it changing: +# the build date, comments with a generation time. +_VOLATILE = re.compile(rb'.*?|', re.S) + + +def _body_hash(content): + """Fingerprint of a feed body, without the parts that change on every request.""" + return hashlib.sha256(_VOLATILE.sub(b'', content)).hexdigest() PERMANENT_REDIRECTS = (301, 308) -def _fetch_feed(url, etag='', last_modified=''): +def _fetch_feed(url, etag='', last_modified='', body_hash=''): """Download and parse a feed (with a timeout, unlike feedparser's own - fetch). With etag/last_modified it is a conditional GET. Returns a + fetch). With etag/last_modified it is a conditional GET; with body_hash a + body like the last one counts as unchanged too (a 304), unparsed — + parsing is what takes the time (seconds for a big feed). Returns a FeedFetch; raises FeedFetchError when there is nothing usable.""" headers = {'User-Agent': 'diora podcast refresh'} if etag: @@ -342,6 +358,10 @@ def _fetch_feed(url, etag='', last_modified=''): if r.status_code >= 400: raise FeedFetchError(f'HTTP {r.status_code}') + new_hash = _body_hash(r.content) + if body_hash and new_hash == body_hash: + return FeedFetch(304, None, etag, last_modified, r.url, permanent_url, body_hash) + 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') @@ -352,7 +372,7 @@ def _fetch_feed(url, etag='', last_modified=''): # 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, + r.url, permanent_url, new_hash, ) @@ -361,13 +381,14 @@ def _validators(feeds): `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} + pairs = {(f.http_etag, f.http_last_modified, f.http_body_hash) 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} + etag, last_modified, body_hash = pairs.pop() + out = {'etag': etag, 'last_modified': last_modified} if etag or last_modified else {} + if body_hash: + out['body_hash'] = body_hash + return out def _error_text(e): @@ -425,45 +446,50 @@ AUDIO_EXTENSIONS = ('.mp3', '.m4a', '.m4b', '.ogg', '.oga', '.opus', '.aac', '.f VIDEO_EXTENSIONS = ('.mp4', '.m4v', '.mov', '.webm') -def _refresh_feeds(feeds, chunk=32, full=False): +def _refresh_feeds(feeds, full=False, deadline=None): """Refresh many feeds at once: the downloads run in parallel and each - feed URL is fetched once, even when several users subscribed to it; the - database writes then happen one by one. Returns (refreshed, failed, new). + feed URL is fetched once, even when several users subscribed to it; each + download is written as soon as it's there (the writes one by one, while + the others still load). Returns (refreshed, failed, new). `full`: download every feed again (no 304) and take in episodes the - reader skipped before (see _apply_feed).""" + reader skipped before (see _apply_feed). `deadline` (time.monotonic()): + stop there; feeds not done by then stay as they are.""" feeds = list(feeds) refreshed = failed = new = 0 - for start in range(0, len(feeds), chunk): - part = feeds[start:start + chunk] - by_url = {} - for f in part: - by_url.setdefault(f.rss_url, []).append(f) - results = {} + by_url = {} + for f in feeds: + by_url.setdefault(f.rss_url, []).append(f) + if not by_url: + return 0, 0, 0 - def fetch(url): - try: - return url, _fetch_feed(url, **({} if full else _validators(by_url[url]))), None - except Exception as e: # network, HTTP or parser trouble: report and go on - return url, None, e + def fetch(url): + try: + return url, _fetch_feed(url, **({} if full else _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, fetched, err in pool.map(fetch, urls): - results[url] = (fetched, err) - - for feed in part: - 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, fetched, full=full) - refreshed += 1 - except Exception as e: - failed += 1 - log.warning('updating %s failed: %s', feed.rss_url, e) + pool = ThreadPoolExecutor(max_workers=min(FEED_WORKERS, len(by_url))) + futures = [pool.submit(fetch, url) for url in sorted(by_url)] + try: + wait = None if deadline is None else max(0.0, deadline - time.monotonic()) + for future in as_completed(futures, timeout=wait): + url, fetched, err = future.result() + for feed in by_url[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, fetched, full=full) + refreshed += 1 + except Exception as e: + failed += 1 + log.warning('updating %s failed: %s', feed.rss_url, e) + except FuturesTimeout: + log.info('refresh stopped at its deadline: %d of %d feeds done', refreshed + failed, len(feeds)) + finally: + pool.shutdown(wait=False, cancel_futures=True) return refreshed, failed, new @@ -681,8 +707,9 @@ def _apply_feed(feed_obj, fetched, full=False): # skips all of the above) and a clean error state. feed_obj.http_etag = fetched.etag feed_obj.http_last_modified = fetched.last_modified + feed_obj.http_body_hash = fetched.body_hash _clear_error(feed_obj) - feed_obj.save(update_fields=['http_etag', 'http_last_modified', *ERROR_FIELDS]) + feed_obj.save(update_fields=['http_etag', 'http_last_modified', 'http_body_hash', *ERROR_FIELDS]) _move_to_permanent_url(feed_obj, fetched.permanent_url) return new_count @@ -1557,7 +1584,9 @@ 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/ …). - The user asked, so feeds the hourly job backs off from are tried too.""" + The user asked, so feeds the hourly job backs off from are tried too. + It answers within REFRESH_WAIT seconds: `left` feeds weren't done by + then (done: false) and wait for the next call or the hourly job.""" if not request.user.is_authenticated: return JsonResponse({'error': 'authentication required'}, status=401) try: @@ -1569,9 +1598,11 @@ def refresh_all_feeds(request): except (TypeError, ValueError): return JsonResponse({'error': 'max_age must be a number'}, status=400) cutoff = timezone.now() - timedelta(seconds=max_age) - feeds = request.user.podcast_feeds.filter(Q(last_refreshed_at__isnull=True) | Q(last_refreshed_at__lt=cutoff)) - refreshed, failed, new = _refresh_feeds(feeds) - return JsonResponse({'ok': True, 'refreshed': refreshed, 'failed': failed, 'new_episodes': new}) + feeds = list(request.user.podcast_feeds.filter(Q(last_refreshed_at__isnull=True) | Q(last_refreshed_at__lt=cutoff))) + refreshed, failed, new = _refresh_feeds(feeds, deadline=time.monotonic() + REFRESH_WAIT) + left = len(feeds) - refreshed - failed + return JsonResponse({'ok': True, 'refreshed': refreshed, 'failed': failed, 'new_episodes': new, + 'done': left == 0, 'left': left}) @csrf_exempt diff --git a/static/js/podcasts-web.js b/static/js/podcasts-web.js index 83c8c8e..466abea 100644 --- a/static/js/podcasts-web.js +++ b/static/js/podcasts-web.js @@ -164,11 +164,15 @@ try { if (fetchFeeds) { if (!(await flush())) return false; - await fetch('/podcasts/feeds/refresh-all/', { + const rr = await fetch('/podcasts/feeds/refresh-all/', { method: 'POST', credentials: 'same-origin', headers: {'Content-Type': 'application/json', 'X-CSRFToken': csrf()}, body: '{"max_age":60}', }); + // The server answers after ~40 s at most; when it didn't get to every feed, + // the state once more a minute later. + const done = await rr.json().then(d => d.done !== false, () => true); + if (!done) setTimeout(() => refresh(false), 60000); } // Progress is saved every few seconds while an episode plays; try again // when it slipped in while the state was on its way. @@ -1817,6 +1821,9 @@ if (!source) listen(); } }); + // Open for long (diora desktop, a pinned tab): the state every 15 minutes too — the + // server fetches the feeds hourly, new episodes show up without a reload. + setInterval(() => { if (document.visibilityState === 'visible') refresh(false); }, 15 * 60000); window.addEventListener('online', () => { flush(); if (!source) listen(); diff --git a/static/js/sw.js b/static/js/sw.js index c7d9909..a4305f9 100644 --- a/static/js/sw.js +++ b/static/js/sw.js @@ -2,7 +2,7 @@ * diora service worker — caches the app shell for offline use. */ -const CACHE = 'diora-v82'; +const CACHE = 'diora-v83'; const SHELL = [ '/static/css/app.css', '/static/js/app.js',