Vito
66ca988d2a
* feat(trash): a freedesktop trash backend that renames without replacing or refuses (#480) * feat(images): remove a file's records and hide it with a removal mark (#480) Deletes every record of one file under every recorded spelling (image links, tags, auto-tag marks, pending review items, category, collection memberships, favorite) in one transaction. A Needs review item that is kept keeps its annotation rows (ON DELETE RESTRICT). The image_removals table hides an exact file (locator, size, mtime_ns) until it changes; linking skips a marked file, and an explicit tag, category, favorite or collection write shows it again. * feat(images): review and apply Delete images as a reviewed plan job (#480) POST /prompt_manager/images/delete/preview reviews both modes; the maintenance apply takes {token, mode}. Library mode removes the records and writes the removal mark; trash mode moves the file first and removes its records only after a successful move. The workspace listing, the output scan, the duplicate find and Tag output images skip marked files. * feat(images): Delete… in the Images selection bar, inspector and viewer (#480) The concept's Remove links… place, D159: one review per selection that asks each time (nothing preselected) whether to remove the files from the library only or also move them to the trash. The browser test server uses a fake trash. * test(images): portable delete tests, the native no-trash proof, D159 in TESTING.md (#480) * perf(images): cache the removal marks so a listing page reads no extra table (#480) The marks reload only when this process writes or clears one, or a restore replaces the database. The route snapshot gains the preview route. * test(images): the full inspector specs expect D159's Delete… in the panel and the viewer (#480) * fix(images): the delete review starts nothing without a chosen mode, and dims its primary until then (#480) An aria-disabled primary ignores every activation, a chosen one starts once, and the apply helper refuses a missing mode. Tests cover a forced click, a double click, Enter and Space on the primary, Enter in the dialog and a submit event, and the server refusing an apply without an exact mode.
721 lines
30 KiB
Python
721 lines
30 KiB
Python
"""Maintenance routes: reviewed, read-only plans (#249 System S6).
|
|
|
|
``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
|
|
one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
|
|
``missing_links``; ``database/maintenance.py``; ``duplicate_files``,
|
|
``py/api/duplicate_files.py``; ``rotated_logs``, the rotated log copies,
|
|
``py/api/log_cleanup.py``, which has its own apply job) off the event loop and
|
|
answers a ``Plan``: ``{success, token, operation, expires_at, scope:{count,
|
|
label}, protected:{count, reasons}, backup:{policy, destination_label},
|
|
media:{deletes_files, label}, details:[≤50 {label, detail}]}``. The preview
|
|
writes nothing. The plan's fingerprint, its targets and the sources it judged
|
|
stay in the server's ``PlanStore`` under the token, bound to the database the
|
|
preview read (a restore that replaces it invalidates the plan).
|
|
|
|
The operation enum is strict, ``options`` must be an empty object and the
|
|
body is capped at ``PREVIEW_BODY_MAX`` bytes. Errors are ``{success:false,
|
|
code, error}`` with ``invalid-request`` (400), ``too-large`` (413: the body,
|
|
or a plan over the store's budget), ``storage-changed`` (409: the database
|
|
was replaced while the preview read it) or a null code (500); none carries a
|
|
server path or exception text.
|
|
|
|
``POST /prompt_manager/maintenance/apply {token}`` applies a reviewed plan as
|
|
an Activity job (202 ``{job}``): kind ``maintenance``, titled after the
|
|
operation, admission ``records`` (refused while a scan, restore, LoRA import
|
|
or other record-changing job runs, and refusing them meanwhile), not
|
|
cancellable, with outcome groups ``changed``, ``protected`` and ``errors``.
|
|
The token is consumed when the job starts (on the same loop turn, before its
|
|
runner runs); a refused start keeps it. The runner:
|
|
|
|
1. takes a safety backup when the plan was reviewed with the policy on or the
|
|
policy (``maintenance.safety_backup``, default on) is on now: SQLite's
|
|
backup API into a temp file beside the database, ``verify_database_file``,
|
|
fsync, ``0o600`` on POSIX, then ``os.replace`` onto a unique
|
|
``<db>.backup_<YYYYmmdd_HHMMSS_ffffff>[_n]`` name. Any failure removes
|
|
what it wrote and ends the job ``backup-failed`` with nothing changed;
|
|
2. applies the plan, which first rechecks the reviewed fingerprint inside
|
|
``BEGIN IMMEDIATE`` (``plan-changed`` writes nothing);
|
|
3. counts the changed and protected records, and prunes old safety backups.
|
|
|
|
The job log names the backup by its file name only, and the job says "No
|
|
safety backup" when none was taken. Apply errors: ``invalid-request`` (400),
|
|
``apply-failed`` (500: nothing started), ``plan-expired`` (410: unknown,
|
|
expired or forgotten after a restart),
|
|
``plan-used`` (409, with the job that applied it), ``storage-changed`` (409:
|
|
a restore replaced the database since the review) and ``busy`` (409).
|
|
"""
|
|
|
|
import datetime
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
|
|
from aiohttp import web
|
|
|
|
from .admin import _prune_safety_backups, _scan_location, _scan_roots
|
|
from .job_routes import busy_response
|
|
from .jobs import MAINTENANCE_JOB_KIND, JobBusy
|
|
from .log_cleanup import ROTATED_LOGS
|
|
from .maintenance_plans import PlanExpired, PlanUsed, ScopeTooLarge
|
|
|
|
try:
|
|
from ...database import safety_backup as _backups
|
|
from ...database.safety_backup import ( # noqa: F401 - re-exported
|
|
SAFETY_TEMP_PREFIX,
|
|
SafetyBackupFailed,
|
|
sweep_safety_temps,
|
|
)
|
|
from ...database.workflow_storage import PRE_MIGRATION_LABEL, pre_migration_backups
|
|
from ...database.health import readonly_uri
|
|
from ...database.maintenance import MAINTENANCE_OPERATIONS, PlanChanged
|
|
except ImportError:
|
|
from database import safety_backup as _backups
|
|
from database.safety_backup import ( # noqa: F401 - re-exported
|
|
SAFETY_TEMP_PREFIX,
|
|
SafetyBackupFailed,
|
|
sweep_safety_temps,
|
|
)
|
|
from database.workflow_storage import PRE_MIGRATION_LABEL, pre_migration_backups
|
|
from database.health import readonly_uri
|
|
from database.maintenance import MAINTENANCE_OPERATIONS, PlanChanged
|
|
|
|
PREVIEW_BODY_MAX = 4096
|
|
MAINTENANCE_PREVIEW_LEASE = "Maintenance preview"
|
|
# Duplicate-file previews carry the reviewed keepers (py/api/duplicate_files.py).
|
|
DUPLICATE_PREVIEW_BODY_MAX = 1024 * 1024
|
|
DUPLICATE_FILES = "duplicate_files"
|
|
WORKFLOW_STORAGE = "workflow_storage" # applied by py/api/workflow_storage.py
|
|
# rotated_logs (#329, py/api/log_cleanup.py) reviews log files, not records.
|
|
PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES, ROTATED_LOGS)
|
|
# Room left over beyond what a step writes, so a disk is never filled to 0.
|
|
DISK_MARGIN_BYTES = 16 * 2**20
|
|
# The concept's Maintenance row titles, which name each operation's job.
|
|
MAINTENANCE_TITLES = {
|
|
"duplicate_prompts": "Duplicate prompt text",
|
|
"optimize": "Optimize database",
|
|
"unused_components": "Unused prompt components",
|
|
"missing_links": "Missing-file links",
|
|
"workflow_storage": "Store each workflow once",
|
|
}
|
|
_OPERATION_ERROR = "operation must be one of: " + ", ".join(PREVIEW_OPERATIONS)
|
|
MAINTENANCE_GROUPS = ("changed", "protected", "errors")
|
|
MAX_TOKEN_LENGTH = 128
|
|
BACKUP_FAILED = (
|
|
"The safety backup could not be taken and verified; nothing was changed."
|
|
)
|
|
PLAN_CHANGED = (
|
|
"The library changed since the review; nothing was changed. "
|
|
"Review the plan again."
|
|
)
|
|
APPLY_FAILED = "Maintenance failed; nothing was changed."
|
|
NOTHING_TO_DO = "Every image record already stores its workflow once; nothing to do."
|
|
NO_BACKUP = "No safety backup"
|
|
|
|
|
|
def backup_failed_message(error):
|
|
"""The job line for a failed backup, naming a damaged workflow document."""
|
|
reason = getattr(error, "reason", None) or ""
|
|
if reason.startswith("Workflow document "):
|
|
return f"{BACKUP_FAILED} {reason}"
|
|
return BACKUP_FAILED
|
|
|
|
|
|
def _free_bytes(path):
|
|
return shutil.disk_usage(path).free
|
|
|
|
|
|
def _device(path):
|
|
try:
|
|
return os.stat(path).st_dev
|
|
except OSError:
|
|
return path
|
|
|
|
|
|
def sqlite_temp_dir(environ=None, *, windows=None, usable=None):
|
|
"""The folder SQLite builds its temporary files in (VACUUM's copy).
|
|
|
|
Mirrors SQLite's own search, which differs from ``tempfile``'s. On Unix
|
|
(``unixTempFileDir``): ``SQLITE_TMPDIR``, ``TMPDIR``, then ``/var/tmp``,
|
|
``/usr/tmp`` and ``/tmp``, the first that is a writable directory, else
|
|
the current directory. On Windows (``GetTempPathW``): ``TMP``, ``TEMP``,
|
|
then ``USERPROFILE``, else the current directory.
|
|
"""
|
|
environ = os.environ if environ is None else environ
|
|
if windows is None:
|
|
windows = os.name == "nt"
|
|
if windows:
|
|
for name in ("TMP", "TEMP", "USERPROFILE"):
|
|
if environ.get(name):
|
|
return environ[name]
|
|
return os.path.abspath(".")
|
|
if usable is None:
|
|
|
|
def usable(path):
|
|
return os.path.isdir(path) and os.access(path, os.W_OK | os.X_OK)
|
|
|
|
candidates = [environ.get("SQLITE_TMPDIR"), environ.get("TMPDIR")]
|
|
candidates += ["/var/tmp", "/usr/tmp", "/tmp"]
|
|
for candidate in candidates:
|
|
if candidate and usable(candidate):
|
|
return candidate
|
|
return os.path.abspath(".")
|
|
|
|
|
|
def database_bytes(db_path):
|
|
"""The size a safety backup of the database will be (blocking).
|
|
|
|
SQLite's backup API copies every page, committed WAL content included, so
|
|
the copy is ``page_count * page_size`` read on one read-only snapshot.
|
|
"""
|
|
conn = sqlite3.connect(readonly_uri(db_path), uri=True)
|
|
try:
|
|
page_size = conn.execute("PRAGMA page_size").fetchone()[0]
|
|
return page_size * conn.execute("PRAGMA page_count").fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _megabytes(size):
|
|
return f"{size / 2**20:,.1f} MB"
|
|
|
|
|
|
def disk_plan(db_path, *, backup, vacuum, vacuum_bytes=None):
|
|
"""Whether the disk has room for an apply's writes (blocking).
|
|
|
|
``backup``: the safety backup, one database-sized file beside the
|
|
database. ``vacuum``: VACUUM writes the rebuilt database through the WAL
|
|
beside it and builds it in a temporary copy in SQLite's temporary folder
|
|
(``sqlite_temp_dir``; SQLite's documentation: up to twice the database
|
|
in free space). Needs
|
|
on one device add up. ``vacuum_bytes``, when given, is what VACUUM
|
|
copies (the live pages, for Reclaim space) instead of the whole file.
|
|
Returns ``(public, short)``: ``public`` is
|
|
``{free_bytes, required_bytes, sufficient}`` for the database folder,
|
|
``short`` None or ``(place, what, required, free)`` of the first place
|
|
that is short, and ``size`` the database's size (0 when nothing is
|
|
written).
|
|
"""
|
|
db_path = os.path.abspath(db_path)
|
|
folder = os.path.dirname(db_path)
|
|
size = database_bytes(db_path) if backup or vacuum else 0
|
|
needs = [] # (place, folder, bytes, what)
|
|
if backup:
|
|
needs.append(("database folder", folder, size, "the safety backup"))
|
|
if vacuum:
|
|
copy = size if vacuum_bytes is None else vacuum_bytes
|
|
what = "Optimize" if vacuum_bytes is None else "Reclaim space"
|
|
needs.append(("database folder", folder, copy, what))
|
|
needs.append(("temporary folder", sqlite_temp_dir(), copy, what))
|
|
devices = {}
|
|
for place, path, needed, what in needs:
|
|
entry = devices.setdefault(_device(path), {"place": place, "path": path})
|
|
entry["bytes"] = entry.get("bytes", 0) + needed
|
|
entry.setdefault("what", [])
|
|
if what not in entry["what"]:
|
|
entry["what"].append(what)
|
|
short = None
|
|
public = {"free_bytes": _free_bytes(folder), "required_bytes": 0}
|
|
for device, entry in devices.items():
|
|
required = entry["bytes"] + DISK_MARGIN_BYTES
|
|
free = (
|
|
public["free_bytes"]
|
|
if entry["path"] == folder
|
|
else _free_bytes(entry["path"])
|
|
)
|
|
if device == _device(folder):
|
|
public["required_bytes"] = required
|
|
if free < required and short is None:
|
|
short = (entry["place"], " and ".join(entry["what"]), required, free)
|
|
public["sufficient"] = short is None
|
|
return public, short, size
|
|
|
|
|
|
# A restore upload's hard ceiling; below it, free disk space decides.
|
|
RESTORE_MAX_BYTES = 64 * 2**30
|
|
|
|
|
|
def size_label(size):
|
|
"""``1.6 GB`` or ``120.0 MB``: a backup's size for a message."""
|
|
if size >= 2**30:
|
|
return f"{size / 2**30:,.1f} GB"
|
|
return _megabytes(size)
|
|
|
|
|
|
def restore_room(db_path):
|
|
"""The largest restore upload the disk has room for (blocking).
|
|
|
|
Everything a restore writes is in the library's folder (#318): the
|
|
upload, its staged copy (``database/restore_stage.py``) and the WAL
|
|
growth while the stage is copied over the live database (3x the upload),
|
|
plus the safety backup of the current database (1x its size), with
|
|
``DISK_MARGIN_BYTES`` to spare.
|
|
"""
|
|
db_path = os.path.abspath(db_path)
|
|
folder = os.path.dirname(db_path)
|
|
current = database_bytes(db_path) if os.path.exists(db_path) else 0
|
|
room = (_free_bytes(folder) - current - DISK_MARGIN_BYTES) // 3
|
|
return max(0, room)
|
|
|
|
|
|
def restore_space_message(size, room, *, known):
|
|
"""The truthful refusal of a restore the disks have no room for."""
|
|
if known:
|
|
return (
|
|
f"Not enough free disk space to restore a {size_label(size)} backup; "
|
|
f"there is room for about {size_label(room)}. Nothing was changed."
|
|
)
|
|
return (
|
|
"Not enough free disk space to restore this backup: it is larger than "
|
|
f"the {size_label(room)} there is room for. Nothing was changed."
|
|
)
|
|
|
|
|
|
def short_space_message(short):
|
|
"""The truthful refusal for ``disk_plan``'s ``short``."""
|
|
place, what, required, free = short
|
|
verb = "need" if " and " in what else "needs"
|
|
return (
|
|
f"Not enough free disk space: {what} {verb} about {_megabytes(required)} "
|
|
f"in the {place} and {_megabytes(free)} is free. Nothing was changed."
|
|
)
|
|
|
|
|
|
def _changed_line(operation, count):
|
|
if operation == "optimize":
|
|
return "Optimized the database"
|
|
if operation == "duplicate_prompts":
|
|
noun = "duplicate prompt" if count == 1 else "duplicate prompts"
|
|
return f"Merged {count:,} {noun}"
|
|
if operation == "unused_components":
|
|
noun = "unused prompt component" if count == 1 else "unused prompt components"
|
|
return f"Removed {count:,} {noun}"
|
|
noun = "link to a missing file" if count == 1 else "links to missing files"
|
|
return f"Removed {count:,} {noun}"
|
|
|
|
|
|
def take_safety_backup(model, label="backup"):
|
|
"""Write and verify a safety backup; return its file name (blocking).
|
|
|
|
See ``database/safety_backup.py``: a backup file on disk is always a
|
|
verified one. Raises SafetyBackupFailed, leaving no file.
|
|
"""
|
|
return os.path.basename(_backups.take_safety_backup(model, label))
|
|
|
|
|
|
def _failure(message, status, code=None, **extra):
|
|
return web.json_response(
|
|
{"success": False, "code": code, "error": message, **extra}, status=status
|
|
)
|
|
|
|
|
|
def _invalid(message):
|
|
return _failure(message, 400, "invalid-request")
|
|
|
|
|
|
def _iso(timestamp):
|
|
return datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc).isoformat()
|
|
|
|
|
|
class MaintenanceRoutesMixin:
|
|
"""Mixin providing the /prompt_manager/maintenance plan endpoints."""
|
|
|
|
def _register_maintenance_routes(self, routes):
|
|
@routes.post("/prompt_manager/maintenance/preview")
|
|
async def maintenance_preview_route(request):
|
|
return await self.preview_maintenance(request)
|
|
|
|
@routes.post("/prompt_manager/maintenance/apply")
|
|
async def maintenance_apply_route(request):
|
|
return await self.apply_maintenance_plan(request)
|
|
|
|
def _database_identity(self, db):
|
|
"""What binds a plan to one database: the swap generation and its file."""
|
|
path = os.path.normcase(os.path.realpath(db.model.db_path))
|
|
return (self._db_generation, path)
|
|
|
|
@staticmethod
|
|
def _safety_backup_policy():
|
|
"""``on`` or ``off``: whether an apply takes a safety backup first."""
|
|
from ..config import PromptManagerConfig
|
|
|
|
return "on" if PromptManagerConfig.SAFETY_BACKUP else "off"
|
|
|
|
async def _read_maintenance_body(self, request, allowed, cap=None):
|
|
"""``(data, None)`` for a capped JSON object with only ``allowed`` keys.
|
|
|
|
``cap(data)``, when given, is the body's byte limit for what it asks
|
|
(read up to ``DUPLICATE_PREVIEW_BODY_MAX``); else PREVIEW_BODY_MAX.
|
|
"""
|
|
limit = PREVIEW_BODY_MAX if cap is None else DUPLICATE_PREVIEW_BODY_MAX
|
|
data, error = await self._read_json_body(request, max_bytes=limit)
|
|
if error is not None:
|
|
if error.status == 413:
|
|
return None, _failure("Request body too large", 413, "too-large")
|
|
return None, _invalid("Request body must be a JSON object")
|
|
if cap is not None:
|
|
size = len(json.dumps(data, separators=(",", ":")).encode("utf-8"))
|
|
if size > cap(data):
|
|
return None, _failure("Request body too large", 413, "too-large")
|
|
unknown = sorted(set(data) - set(allowed))
|
|
if unknown:
|
|
return None, _invalid("Unknown field: " + ", ".join(unknown)[:200])
|
|
return data, None
|
|
|
|
async def preview_maintenance(self, request):
|
|
"""Review one operation under a registry lease (#249 S8).
|
|
|
|
The preview's read-only connections must not be open while a restore
|
|
switches the database: it is refused (409 ``busy``) while an
|
|
``exclusive`` job runs, and a restore waits for running previews.
|
|
"""
|
|
try:
|
|
async with self._jobs.lease(MAINTENANCE_PREVIEW_LEASE):
|
|
return await self._preview_maintenance(request)
|
|
except JobBusy as busy:
|
|
return busy_response(busy)
|
|
|
|
async def _preview_maintenance(self, request):
|
|
data, error = await self._read_maintenance_body(
|
|
request,
|
|
("operation", "options"),
|
|
cap=lambda body: (
|
|
DUPLICATE_PREVIEW_BODY_MAX
|
|
if body.get("operation") == DUPLICATE_FILES
|
|
else PREVIEW_BODY_MAX
|
|
),
|
|
)
|
|
if error is not None:
|
|
return error
|
|
operation = data.get("operation")
|
|
options = data.get("options", {})
|
|
if not isinstance(operation, str) or operation not in PREVIEW_OPERATIONS:
|
|
return _invalid(_OPERATION_ERROR)
|
|
if operation == DUPLICATE_FILES:
|
|
return await self.preview_duplicate_files(options)
|
|
if not isinstance(options, dict) or options:
|
|
return _invalid(f"{operation} takes no options")
|
|
if operation == ROTATED_LOGS:
|
|
return await self.preview_log_cleanup()
|
|
db = self.db
|
|
identity = self._database_identity(db)
|
|
try:
|
|
roots = (
|
|
await self._run_in_executor(self._health_roots)
|
|
if operation == "missing_links"
|
|
else []
|
|
)
|
|
draft = await self._run_in_executor(
|
|
db.preview_maintenance, operation, roots
|
|
)
|
|
except Exception:
|
|
self.logger.exception(f"Maintenance preview of {operation} failed")
|
|
return _failure("Failed to review maintenance", 500)
|
|
return await self._plan_response(
|
|
db,
|
|
identity,
|
|
draft.operation,
|
|
draft.fingerprint,
|
|
draft.public(),
|
|
{
|
|
"targets": draft.targets,
|
|
"roots": roots,
|
|
"protected": draft.protected["count"],
|
|
"rows": draft.scope["count"],
|
|
},
|
|
)
|
|
|
|
async def _plan_response(
|
|
self, db, identity, operation, fingerprint, public, payload
|
|
):
|
|
"""Issue the reviewed plan's token and answer the ``Plan``.
|
|
|
|
Refused with ``storage-changed`` when the database was replaced while
|
|
it was reviewed. The plan reports the safety backup's expected size
|
|
and the disk space the apply needs (``disk_plan``).
|
|
"""
|
|
if identity != self._database_identity(self.db):
|
|
return _failure(
|
|
"The database was replaced while it was reviewed. Review again.",
|
|
409,
|
|
"storage-changed",
|
|
)
|
|
policy = self._safety_backup_policy()
|
|
destination = "backup"
|
|
if operation == WORKFLOW_STORAGE:
|
|
# Always backed up first, whatever the policy (the way back);
|
|
# one kept pre-migration backup ever, pooled ones after it.
|
|
policy = "required"
|
|
kept = await self._run_in_executor(pre_migration_backups, db.model.db_path)
|
|
destination = "backup" if kept else PRE_MIGRATION_LABEL
|
|
try:
|
|
disk, _short, size = await self._run_in_executor(
|
|
disk_plan,
|
|
db.model.db_path,
|
|
backup=policy != "off",
|
|
vacuum=operation == "optimize",
|
|
)
|
|
except (OSError, sqlite3.Error):
|
|
self.logger.exception("Could not measure the disk space for a plan")
|
|
return _failure("Failed to review maintenance", 500)
|
|
store = self._maintenance_plans
|
|
try:
|
|
token = store.issue(
|
|
operation, fingerprint, {**payload, "backup": policy}, identity
|
|
)
|
|
except ScopeTooLarge:
|
|
return _failure(
|
|
"This plan is too large to review at once.", 413, "too-large"
|
|
)
|
|
plan = store.peek(token, operation)
|
|
name = os.path.basename(db.model.db_path)
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"token": token,
|
|
"expires_at": _iso(plan.expires_at),
|
|
**public,
|
|
"backup": {
|
|
"policy": policy,
|
|
"destination_label": f"{name}.{destination}_<date and time>",
|
|
"expected_bytes": size if policy != "off" else None,
|
|
},
|
|
"disk": disk,
|
|
},
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
# ── apply ─────────────────────────────────────────────────────────
|
|
|
|
async def apply_maintenance_plan(self, request):
|
|
from .image_delete import IMAGE_DELETE, parse_mode
|
|
|
|
data, error = await self._read_maintenance_body(request, ("token", "mode"))
|
|
if error is not None:
|
|
return error
|
|
token = data.get("token")
|
|
if not isinstance(token, str) or not 0 < len(token) <= MAX_TOKEN_LENGTH:
|
|
return _invalid("token must be the plan's token")
|
|
plan, refused = self._peek_plan(token)
|
|
if refused is not None:
|
|
return refused
|
|
mode = None
|
|
if plan.operation == IMAGE_DELETE:
|
|
mode = parse_mode(data.get("mode"))
|
|
if mode is None:
|
|
return _invalid('mode must be "library" or "trash"')
|
|
elif "mode" in data:
|
|
return _invalid("Unknown field: mode")
|
|
store = self._maintenance_plans
|
|
if plan.operation == ROTATED_LOGS:
|
|
# Log files only: no database identity, backup or disk check.
|
|
try:
|
|
job = self._start_log_cleanup(plan)
|
|
except JobBusy as busy:
|
|
return busy_response(busy) # the token stays usable
|
|
# Same loop turn as the peek and the start: the runner has not run.
|
|
store.consume(token, plan.operation, job_id=job.id)
|
|
return web.json_response(
|
|
{"success": True, "job": job.summary()}, status=202
|
|
)
|
|
db = self.db
|
|
if plan.db_identity != self._database_identity(db):
|
|
store.discard(token)
|
|
return _failure(
|
|
"The database was replaced since the review. Review again.",
|
|
409,
|
|
"storage-changed",
|
|
)
|
|
if plan.operation == WORKFLOW_STORAGE and not plan.payload.get("rows"):
|
|
store.discard(token) # no job, no backup: nothing to rewrite
|
|
return _failure(NOTHING_TO_DO, 409, "nothing-to-do")
|
|
backup = self._backup_promised(plan)
|
|
try:
|
|
_disk, short, _size = await self._run_in_executor(
|
|
disk_plan,
|
|
db.model.db_path,
|
|
backup=backup,
|
|
vacuum=plan.operation == "optimize",
|
|
)
|
|
except (OSError, sqlite3.Error):
|
|
self.logger.exception("Could not measure the disk space for an apply")
|
|
# Coded: the workspace tells the server's own refusal (nothing
|
|
# started) from a proxy's 5xx (the job may have started).
|
|
return _failure(APPLY_FAILED, 500, "apply-failed")
|
|
if short is not None:
|
|
# Refused before any job starts: nothing fails halfway for space.
|
|
return _failure(short_space_message(short), 507, "insufficient-space")
|
|
# The token may have been used, expired or dropped while the disk was
|
|
# measured: look again with no await before the job starts.
|
|
plan, refused = self._peek_plan(token)
|
|
if refused is not None:
|
|
return refused
|
|
try:
|
|
if plan.operation == DUPLICATE_FILES:
|
|
job = self._start_duplicate_cleanup(db, plan, backup)
|
|
elif plan.operation == IMAGE_DELETE:
|
|
job = self._start_image_delete(db, plan, mode, backup)
|
|
elif plan.operation == WORKFLOW_STORAGE:
|
|
job = self._start_workflow_storage(db, backup)
|
|
else:
|
|
job = self._jobs.start(
|
|
MAINTENANCE_JOB_KIND,
|
|
MAINTENANCE_TITLES[plan.operation],
|
|
# VACUUM rewrites the whole file under the write lock: it
|
|
# runs alone, beside no other job of any class.
|
|
admission=(
|
|
"exclusive" if plan.operation == "optimize" else "records"
|
|
),
|
|
cancellable=False,
|
|
scope={"operation": plan.operation},
|
|
groups=MAINTENANCE_GROUPS,
|
|
runner=lambda job: self._run_maintenance_plan(
|
|
job, db, plan, backup
|
|
),
|
|
)
|
|
except JobBusy as busy:
|
|
return busy_response(busy) # the token stays usable
|
|
# Same loop turn as the peek and the start: the runner has not run.
|
|
store.consume(token, plan.operation, job_id=job.id)
|
|
return web.json_response({"success": True, "job": job.summary()}, status=202)
|
|
|
|
def _peek_plan(self, token):
|
|
"""``(plan, None)`` for a usable token, else ``(None, refusal)``."""
|
|
try:
|
|
return self._maintenance_plans.peek(token), None
|
|
except PlanUsed as used:
|
|
job = self._jobs.get(used.job_id)
|
|
return None, _failure(
|
|
"This plan was already applied.",
|
|
409,
|
|
"plan-used",
|
|
job={"id": used.job_id, "title": job.title if job else None},
|
|
)
|
|
except PlanExpired:
|
|
return None, _failure(
|
|
"This review expired. Review the plan again.", 410, "plan-expired"
|
|
)
|
|
|
|
def _backup_promised(self, plan):
|
|
"""A plan reviewed with the policy on keeps that promise; so does on now.
|
|
|
|
Decided once, when the apply starts (the disk check measured it), and
|
|
handed to the job.
|
|
"""
|
|
if plan.operation == WORKFLOW_STORAGE:
|
|
return True # never without its pre-migration backup
|
|
return plan.payload["backup"] == "on" or self._safety_backup_policy() == "on"
|
|
|
|
async def _safety_backup_step(self, job, db, backup_on):
|
|
"""``(ok, name)``: take the promised safety backup, or report none.
|
|
|
|
``name`` is the backup's file name, or None when no backup was
|
|
promised. A failed backup ends the job ``backup-failed`` (``ok``
|
|
False).
|
|
"""
|
|
if not backup_on:
|
|
job.log_line("No safety backup (the policy is off)")
|
|
return True, None
|
|
job.update(status_line="Taking a safety backup")
|
|
job.log_line("Taking a safety backup of the database")
|
|
try:
|
|
backup = await self._run_in_executor(take_safety_backup, db.model)
|
|
except SafetyBackupFailed as error:
|
|
self.logger.error("Safety backup failed", exc_info=True)
|
|
_fail_job(job, "backup-failed", backup_failed_message(error))
|
|
return False, None
|
|
job.log_line(f"Safety backup {backup} verified")
|
|
return True, backup
|
|
|
|
async def _prune_backup_pool(self, job, db):
|
|
"""Keep the one ``<db>.backup_*`` pool (restore's too) at its cap."""
|
|
removed = await self._run_in_executor(_prune_safety_backups, db.model.db_path)
|
|
if removed:
|
|
job.log_line(f"Removed {len(removed)} older safety backups")
|
|
|
|
async def _run_maintenance_plan(self, job, db, plan, backup_on):
|
|
ok, backup = await self._safety_backup_step(job, db, backup_on)
|
|
if not ok:
|
|
return
|
|
try:
|
|
await self._apply_maintenance_plan(job, db, plan, backup)
|
|
finally:
|
|
if backup:
|
|
await self._prune_backup_pool(job, db)
|
|
|
|
async def _apply_maintenance_plan(self, job, db, plan, backup):
|
|
payload = plan.payload
|
|
job.update(status_line=f"Applying: {job.title}")
|
|
try:
|
|
result = await self._run_in_executor(
|
|
db.apply_maintenance,
|
|
plan.operation,
|
|
plan.fingerprint,
|
|
payload["targets"],
|
|
payload["roots"],
|
|
)
|
|
except PlanChanged:
|
|
return _fail_job(job, "plan-changed", PLAN_CHANGED)
|
|
except Exception:
|
|
self.logger.exception(f"Maintenance {plan.operation} failed")
|
|
return _fail_job(job, None, APPLY_FAILED)
|
|
rows = await self._run_in_executor(
|
|
self._maintenance_rows, plan.operation, result["rows"]
|
|
)
|
|
job.record_counted("changed", result["changed"], rows)
|
|
job.record_counted("protected", payload["protected"])
|
|
line = _changed_line(plan.operation, result["changed"])
|
|
job.log_line(line)
|
|
job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
|
|
|
|
def _maintenance_rows(self, operation, described):
|
|
"""Outcome rows for the changed records (root-relative paths only)."""
|
|
if operation == "missing_links":
|
|
roots = _scan_roots(self._get_all_output_dirs())
|
|
rows = []
|
|
for item in described:
|
|
path, label, index = _scan_location(item["image_path"], roots)
|
|
rows.append(
|
|
{
|
|
"path": path,
|
|
"root": label,
|
|
"root_index": index,
|
|
"detail": "Link removed; the file was already missing",
|
|
}
|
|
)
|
|
return rows
|
|
if operation == "duplicate_prompts":
|
|
return [
|
|
{
|
|
"path": "",
|
|
"detail": (
|
|
f"Prompt {item['prompt_id']} merged into "
|
|
f"prompt {item['kept_id']}"
|
|
),
|
|
}
|
|
for item in described
|
|
]
|
|
return [
|
|
{"path": "", "detail": f"Removed: {item.get('text') or item['prompt_id']}"}
|
|
for item in described
|
|
]
|
|
|
|
|
|
def backup_note(backup):
|
|
"""The status line's backup part: the file name, or that there is none."""
|
|
return f"Safety backup {backup}" if backup else NO_BACKUP
|
|
|
|
|
|
def _fail_job(job, code, message):
|
|
"""End ``job`` failed with ``message`` (one error row) and its code.
|
|
|
|
The code is also the job's ``scope["failure"]``, so Activity can offer
|
|
Review again for a plan the library overtook.
|
|
"""
|
|
if code:
|
|
job.scope["failure"] = code
|
|
job.record("errors", {"path": "", "detail": message}, f"! {message}")
|
|
job.update(status_line=message)
|
|
job.publish({"type": "error", "code": code, "message": message})
|
|
job.finish("failed")
|