102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
"""Reachability GC. Dry-run by default; exclusive lease covers mark AND sweep.
|
|
|
|
python -m app.gc [--execute] [--interval 86400]
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
|
|
from sqlalchemy import delete, select
|
|
|
|
from app.blobs import BLOB_ID_RE, blob_path
|
|
from app.config import get_settings
|
|
from app.db import SessionLocal
|
|
from app.maintenance import storage_access
|
|
from app.models import Game, GameVersion, Revision, VersionHistory
|
|
|
|
|
|
def hashes(manifest):
|
|
return {entry["hash"] for entry in (manifest or {}).get("files", {}).values()}
|
|
|
|
|
|
def reachable_revisions(graph, roots):
|
|
live, pending = set(), list(roots)
|
|
while pending:
|
|
key = pending.pop()
|
|
if key in live or key not in graph:
|
|
continue
|
|
live.add(key)
|
|
pending.extend(graph[key].parents)
|
|
return live
|
|
|
|
|
|
def candidates(root, live):
|
|
root = root.resolve()
|
|
result = []
|
|
for path in root.glob("*/*/*"):
|
|
if not BLOB_ID_RE.fullmatch(path.name) or path.name in live:
|
|
continue
|
|
# Never follow links out of the blob store (including linked directories).
|
|
if path.is_symlink() or path.parent.is_symlink() or path.parent.parent.is_symlink():
|
|
continue
|
|
resolved = path.resolve()
|
|
if resolved.is_relative_to(root) and path.is_file() and path == blob_path(path.name):
|
|
result.append(path)
|
|
return result
|
|
|
|
|
|
async def collect(execute=False):
|
|
async with storage_access(exclusive=True):
|
|
async with SessionLocal() as db:
|
|
versions = list((await db.scalars(select(GameVersion))).all())
|
|
graph = {r.id: r for r in (await db.scalars(select(Revision))).all()}
|
|
roots = {rid for v in versions for rid in (v.revision_id, v.base_revision_id) if rid}
|
|
if roots - set(graph):
|
|
raise RuntimeError("Missing root revisions; GC aborted")
|
|
live_revisions = reachable_revisions(graph, roots)
|
|
if any(p not in graph for key in live_revisions for p in graph[key].parents):
|
|
raise RuntimeError("Incomplete revision graph; GC aborted")
|
|
live = set()
|
|
for v in versions:
|
|
live.update(hashes(v.manifest))
|
|
for key in live_revisions:
|
|
live.update(hashes(graph[key].manifest))
|
|
for manifest in (await db.scalars(select(VersionHistory.manifest))).all():
|
|
live.update(hashes(manifest))
|
|
live.update(h for h in (await db.scalars(select(Game.cover_hash))).all() if h)
|
|
for gallery in (await db.scalars(select(Game.gallery_hashes))).all():
|
|
live.update(gallery or [])
|
|
for key in live:
|
|
if not BLOB_ID_RE.fullmatch(key) or not blob_path(key).is_file():
|
|
raise RuntimeError(f"Missing/invalid referenced blob {key}; GC aborted")
|
|
garbage = candidates(get_settings().blobs_dir, live)
|
|
orphan_revisions = set(graph) - live_revisions
|
|
report = {"execute": execute, "files": len(garbage),
|
|
"bytes": sum(p.stat().st_size for p in garbage),
|
|
"orphanRevisions": len(orphan_revisions), "retainedBlobs": len(live)}
|
|
if execute:
|
|
if orphan_revisions:
|
|
await db.execute(delete(Revision).where(Revision.id.in_(orphan_revisions)))
|
|
await db.commit()
|
|
for path in garbage:
|
|
path.unlink(missing_ok=True)
|
|
return report
|
|
|
|
|
|
async def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--execute", action="store_true", help="Delete unreachable files and orphan revision rows")
|
|
parser.add_argument("--interval", type=int, default=0, help="Repeat interval in seconds (0: one run)")
|
|
args = parser.parse_args()
|
|
if args.interval and args.interval < 60:
|
|
parser.error("interval must be at least 60 seconds")
|
|
while True:
|
|
print(json.dumps(await collect(args.execute)), flush=True)
|
|
if not args.interval:
|
|
break
|
|
await asyncio.sleep(args.interval)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|