fix(sxyprn): reaper porcjami, z widocznym postepem i odporny na restart workera

Przebieg po 151 tys. sprawdzonych zrodel przepadl, bo sam ubilem go wlasnym
`docker compose restart worker` przy wdrazaniu poprawek seekplayera. Log urwal sie
o 06:54, kontener wystartowal o 06:58. Proces zostal jako pusta skorupa (jeden watek,
zero postepu), a wygladalo to na zawieszenie: sxyprn odpowiadal normalnie w 0,2 s,
baza byla bez blokad.

Trzy zmiany, kazda usuwa jedna slabosc, ktora to ukrywala albo pogarszala:
* PORCJAMI po 500, ze swieza pula na kazda porcje. `pool.map` na calej liscie
  kolejkowal 188 tys. zadan naraz, wiec przerwanie w polowie nie zostawialo zadnego
  punktu zaczepienia, a pamiec rosla bez potrzeby.
* JEDEN `UPDATE` na porcje zamiast jednego na wiersz.
* `flush()` po kazdej porcji. Log do pliku byl buforowany i milczal godzinami, przez co
  martwy przebieg wygladal identycznie jak pracujacy.

W docstringu jawne ostrzezenie: to zadanie NIE MOZE chodzic w kontenerze workera, bo
kazdy deploy backendu je ubija. Odpalac przez `docker compose run -d --rm worker`,
czyli w osobnym kontenerze. Wznawianie jest darmowe, bo bierzemy tylko zywe zrodla.

Stan: 52 998 zrodel juz oznaczonych martwymi, wznowione na pozostalych 138 154.
This commit is contained in:
goon-foss 2026-08-27 12:04:25 +02:00
parent 1fd1a750e3
commit 9219c94c5a

View file

@ -22,7 +22,13 @@ miesięcznie). Dlatego niska równoległość i odstęp między pobraniami, żeb
rate-limitem ani banem IP. To celowo wolne. Wznawialne za darmo: bierzemy tylko żywe rate-limitem ani banem IP. To celowo wolne. Wznawialne za darmo: bierzemy tylko żywe
źródła, więc ponowny przebieg pomija już oznaczone. źródła, więc ponowny przebieg pomija już oznaczone.
Użycie (w kontenerze workera): UWAGA NA RESTART: `docker compose restart worker` UBIJA to zadanie, jesli chodzi w
kontenerze workera (2026-08-27: stracilem tak przebieg po 151 tys. sprawdzonych, log
urwal sie 4 minuty przed restartem). Odpalaj przez `docker compose run -d --rm worker`,
bo to OSOBNY kontener, ktorego restart workera nie dotyka. Wznawianie jest darmowe:
bierzemy tylko zywe zrodla, wiec ponowne odpalenie samo pomija juz oznaczone.
Użycie:
python3 /srv/scripts/sxyprn_reap_dead.py --limit 300 --dry-run python3 /srv/scripts/sxyprn_reap_dead.py --limit 300 --dry-run
python3 /srv/scripts/sxyprn_reap_dead.py --limit 5000 python3 /srv/scripts/sxyprn_reap_dead.py --limit 5000
""" """
@ -100,25 +106,36 @@ def main(argv: list[str] | None = None) -> int:
sid, page_url = row sid, page_url = row
return sid, _probe(page_url) return sid, _probe(page_url)
# Porcjami, ze SWIEZA pula na kazda porcje. `pool.map` na calej liscie kolejkowal
# 188 tys. zadan naraz, wiec postep byl widoczny dopiero na koncu, a przerwanie w
# polowie nie zostawialo zadnego punktu zaczepienia.
_PORCJA = 500
for start in range(0, len(rows), _PORCJA):
porcja = rows[start : start + _PORCJA]
do_ubicia: list = []
with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool: with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
for i, (sid, verdict) in enumerate(pool.map(work, rows), 1): for sid, verdict in pool.map(work, porcja):
cnt[verdict] += 1 cnt[verdict] += 1
if verdict == _DEAD and not a.dry_run: if verdict == _DEAD:
do_ubicia.append(sid)
if do_ubicia and not a.dry_run:
# Jeden UPDATE na porcje zamiast jednego na wiersz.
with session_scope() as s: with session_scope() as s:
ubite += s.execute( ubite += s.execute(
text( text(
"UPDATE playback_sources SET dead_at = now(), " "UPDATE playback_sources SET dead_at = now(), "
"dead_reason = 'sxyprn-reap: Post Not Found' " "dead_reason = 'sxyprn-reap: Post Not Found' "
"WHERE id = :id AND dead_at IS NULL" "WHERE id = ANY(:ids) AND dead_at IS NULL"
), ),
{"id": sid}, {"ids": do_ubicia},
).rowcount ).rowcount
if i % 200 == 0:
log.info( log.info(
"[%d/%d] żywe=%d martwe=%d nieznane=%d, ubite=%d (%.0f min)", "[%d/%d] żywe=%d martwe=%d nieznane=%d, ubite=%d (%.0f min)",
i, len(rows), cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN], ubite, min(start + _PORCJA, len(rows)), len(rows),
(time.time() - t0) / 60, cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN], ubite, (time.time() - t0) / 60,
) )
for h in logging.getLogger().handlers:
h.flush() # bez tego log do pliku milczal godzinami (buforowanie)
total = sum(cnt.values()) total = sum(cnt.values())
log.info( log.info(