Feeds abrufen: unveränderte Feeds nicht mehr parsen, ⟳ antwortet nach höchstens 40 s; sw v83
Der stündliche Abruf der 502 Feeds dauerte 114 s, fast nur Parsen: viele Server ignorieren ETag/Last-Modified und schicken jedes Mal den ganzen Feed (Lage der Nation 8,4 MB = 5,9 s). ⟳ in der App lief damit in ihr 60-s-Zeitlimit und schlug fehl. - Fingerabdruck des Feed-Inhalts (ohne lastBuildDate/Kommentare, http_body_hash, Migration 0017): derselbe Inhalt zählt als unverändert und wird nicht geparst - _refresh_feeds: ein Pool für alle Feeds (16 parallel), jeder Feed wird geschrieben, sobald er da ist, optionales Zeitlimit - refresh-all wartet höchstens 40 s (done/left in der Antwort); der Browser-Host holt den Stand dann eine Minute später noch einmal, und alle 15 Minuten Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
986ccad5c9
commit
14720b3669
6 changed files with 151 additions and 51 deletions
18
podcasts/migrations/0017_feed_body_hash.py
Normal file
18
podcasts/migrations/0017_feed_body_hash.py
Normal file
|
|
@ -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),
|
||||
),
|
||||
]
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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'<channel>', b'<channel><lastBuildDate>Sun, 04 Oct 2026 08:00:00 GMT</lastBuildDate><!-- 0.12 s -->', 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'}))
|
||||
|
|
|
|||
|
|
@ -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'<lastBuildDate>.*?</lastBuildDate>|<!--.*?-->', 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,20 +446,21 @@ 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:
|
||||
for f in feeds:
|
||||
by_url.setdefault(f.rss_url, []).append(f)
|
||||
results = {}
|
||||
if not by_url:
|
||||
return 0, 0, 0
|
||||
|
||||
def fetch(url):
|
||||
try:
|
||||
|
|
@ -446,13 +468,13 @@ def _refresh_feeds(feeds, chunk=32, full=False):
|
|||
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]
|
||||
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)
|
||||
|
|
@ -464,6 +486,10 @@ def _refresh_feeds(feeds, chunk=32, full=False):
|
|||
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
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue