subplz-web/backend/retention.py

backend/retention.py (2958 bytes)
1 """What the server keeps, and for how long.
2 
3 A server job brings an audiobook and a book onto this machine. They are the
4 visitor's files, often bought ones, and the disk is small. So nothing stays
5 longer than it is of use:
6 
7   uploaded files   deleted when the job succeeds (runner.py). A job that was
8                    never started, failed or was cancelled keeps them for
9                    `input_retention_hours`, so the visitor can fix the language
10                    and try again; then they go too.
11   results          subtitles and videos stay for `artifact_retention_days`,
12                    then they are deleted. The row of the job stays, so the list
13                    still says what was converted.
14 
15 Jobs that ran in the visitor's browser have no files here, only the subtitles
16 the browser sent back; those follow the rule for results.
17 """
18 
19 from __future__ import annotations
20 
21 import logging
22 import shutil
23 from datetime import timedelta
24 
25 from sqlalchemy.orm import Session
26 
27 from .db import Artifact, Job, JobStatus, utcnow
28 from .runner import Paths
29 from .settings import settings
30 from .storage import storage
31 
32 log = logging.getLogger("subplz-web.retention")
33 
34 _IDLE = [JobStatus.draft, JobStatus.failed, JobStatus.canceled]
35 
36 
37 def sweep(session: Session) -> dict[str, int]:
38     """Delete what is past its time. Safe to run at any moment, and again."""
39     now = utcnow()
40     done = {"inputs": 0, "drafts": 0, "results": 0}
41 
42     old_idle = (
43         session.query(Job)
44         .filter(Job.local == 0, Job.status.in_(_IDLE),
45                 Job.created_at < now - timedelta(hours=settings.input_retention_hours))
46         .all()
47     )
48     for job in old_idle:
49         root = Paths.for_job(job.id).root
50         if root.exists():
51             shutil.rmtree(root, ignore_errors=True)
52             done["inputs"] += 1
53         _delete_stored(job.id)
54         if job.status == JobStatus.draft:
55             # Never started: nothing to show for it.
56             session.delete(job)
57             done["drafts"] += 1
58 
59     old_results = (
60         session.query(Job)
61         .join(Artifact, Artifact.job_id == Job.id)
62         .filter(Job.status == JobStatus.succeeded,
63                 Job.finished_at < now - timedelta(days=settings.artifact_retention_days))
64         .distinct()
65         .all()
66     )
67     for job in old_results:
68         _delete_stored(job.id)
69         shutil.rmtree(Paths.for_job(job.id).root, ignore_errors=True)
70         session.query(Artifact).filter(Artifact.job_id == job.id).delete()
71         job.stage = f"Files deleted after {settings.artifact_retention_days} days"
72         done["results"] += 1
73 
74     session.commit()
75     if any(done.values()):
76         log.info("retention sweep: %s", done)
77     return done
78 
79 
80 def _delete_stored(job_id: str) -> None:
81     try:
82         storage.delete_prefix(job_id)
83     except Exception as exc:  # noqa: BLE001 - the next sweep tries again
84         log.warning("could not delete stored files of %s: %s", job_id, exc)