nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge origin/main (System S8) into feat/327-system-s10-integrations

Browse Source
This commit is contained in:
Vito Sansevero
2026-10-25 10:25:57 -07:00
parent a073dc613e 28ad3a1a1f
commit 4d21f48a8a
24 changed files with 3890 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/api/__init__.py
+6
View File
@@ -57,6 +57,7 @@ from .thumbnail_auto import AutoThumbnailMixin
from .system import SystemRoutesMixin
from .library_location import LibraryLocationRoutesMixin
from .maintenance import MaintenanceRoutesMixin
from .backup_restore import BackupRestoreRoutesMixin
from .workflow_storage import WorkflowStorageRoutesMixin
from .duplicate_files import DuplicateFilesMixin
from .maintenance_plans import PlanStore
@@ -304,6 +305,7 @@ class PromptManagerAPI(
SystemRoutesMixin,
LibraryLocationRoutesMixin,
MaintenanceRoutesMixin,
BackupRestoreRoutesMixin,
WorkflowStorageRoutesMixin,
DuplicateFilesMixin,
LoggingRoutesMixin,
@@ -379,7 +381,10 @@ class PromptManagerAPI(
try:
from .maintenance import sweep_safety_temps
from .backup_restore import restore_stage
swept = sweep_safety_temps(self.db.model.db_path)
swept += restore_stage.sweep_restore_temps(self.db.model.db_path)
if swept:
self.logger.info(f"Removed {len(swept)} interrupted backup copies")
except Exception as e:
@@ -550,6 +555,7 @@ class PromptManagerAPI(
self._register_system_routes(routes)
self._register_library_location_routes(routes)
self._register_maintenance_routes(routes)
self._register_backup_restore_routes(routes)
self._register_duplicate_routes(routes)
self._register_logging_routes(routes)
self._register_autotag_routes(routes)
py/api/admin.py
-215
View File
@@ -3,14 +3,12 @@
import asyncio
import contextlib
from contextlib import closing
import datetime
import functools
import hashlib
import json
import math
import os
import sqlite3
import tempfile
import threading
import time
import urllib.parse
@@ -28,7 +26,6 @@ from .jobs import (
DUPLICATE_CLEANUP_JOB_KIND,
DUPLICATES_JOB_KIND,
MAINTENANCE_JOB_KIND,
RESTORE_JOB_KIND,
JobBusy,
)
@@ -38,7 +35,6 @@ except ImportError:
from utils.parallel import map_parallel
try:
from ...database.operations import PromptDatabase
from ...database.pair_ingestion import ingest_resolved_pair
from ...utils.generation_params import (
PLACEHOLDERS,
@@ -49,7 +45,6 @@ try:
from ...utils.prompt_pair_resolver import resolve_prompt_pair
from ...utils.review_capture import reviewable
except ImportError:
from database.operations import PromptDatabase
from database.pair_ingestion import ingest_resolved_pair
from utils.generation_params import (
PLACEHOLDERS,
@@ -1259,216 +1254,6 @@ class AdminRoutesMixin:
"message": f"Found {len(issues)} consistency issues",
}
async def backup_database(self, request):
"""Download a consistent copy of the prompts database (WAL included)."""
try:
model = self.db.model
db_path = model.db_path
if not os.path.exists(db_path):
return web.json_response(
{"success": False, "error": "Database file not found"}, status=404
)
fd, temp_path = tempfile.mkstemp(suffix=".db")
os.close(fd)
try:
ok = await self._run_in_executor(model.backup_database, temp_path)
if not ok:
return web.json_response(
{"success": False, "error": "Failed to create database backup"},
status=500,
)
# Read once (the file is deleted right after) rather than
# streaming so the temp file never outlives the request.
with open(temp_path, "rb") as fh:
file_data = fh.read()
finally:
if os.path.exists(temp_path):
os.unlink(temp_path)
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"prompts_backup_{timestamp}.db"
return web.Response(
body=file_data,
content_type="application/octet-stream",
headers={
"Content-Disposition": f'attachment; filename="{filename}"',
"Content-Length": str(len(file_data)),
},
)
except Exception as e:
self.logger.error(f"Backup error: {e}", exc_info=True)
return web.json_response(
{
"success": False,
"error": f"Failed to backup database: {self._public_error(e)}",
},
status=500,
)
async def restore_database(self, request):
"""Restore the prompts database from an uploaded SQLite file.
The upload is streamed to a temp file. Free disk space decides how
large it may be (``maintenance.restore_room``), under a 64 GiB ceiling
(``restore_max_bytes``): a request whose Content-Length does not fit
is refused before anything is written, and one without is refused as
soon as its running size does not. The file is
verified with PromptModel.verify_database_file and only then copied
over the live database, which is backed up first. The whole request
holds the registry (records): it is refused while a scan or another
record-changing job runs, and those are refused while it runs.
"""
return await self._run_held(
RESTORE_JOB_KIND,
"Database restore (admin page)",
"records",
lambda: self._restore_database(request),
)
def _restore_refusal(self, size, ceiling, room, *, known):
"""``400 too-large`` past the ceiling, else ``507 insufficient-space``."""
from .maintenance import restore_space_message
if size > ceiling:
return web.json_response(
{
"success": False,
"code": "too-large",
"error": (
"File too large. Maximum size is "
f"{ceiling // (1024 * 1024)}MB"
),
},
status=400,
)
return web.json_response(
{
"success": False,
"code": "insufficient-space",
"error": restore_space_message(size, room, known=known),
},
status=507,
)
async def _restore_database(self, request):
from .maintenance import RESTORE_MAX_BYTES, restore_room
ceiling = getattr(self, "restore_max_bytes", RESTORE_MAX_BYTES)
temp_path = None
try:
# Free disk space decides how large a backup may be (#313): the
# upload, its staged copy and the swap must all fit.
room = await self._run_in_executor(restore_room, self.db.model.db_path)
max_bytes = min(ceiling, room)
length = request.content_length
if length is not None and length > max_bytes:
return self._restore_refusal(length, ceiling, room, known=True)
reader = await request.multipart()
field = await reader.next()
if not field or field.name != "database_file":
return web.json_response(
{
"success": False,
"error": (
"No database file uploaded. "
"Expected field name: database_file"
),
},
status=400,
)
fd, temp_path = tempfile.mkstemp(suffix=".db")
os.close(fd)
size = 0
with open(temp_path, "wb") as out:
while True:
chunk = await field.read_chunk()
if not chunk:
break
size += len(chunk)
if size > max_bytes:
return self._restore_refusal(size, ceiling, room, known=False)
out.write(chunk)
if size == 0:
return web.json_response(
{"success": False, "error": "Uploaded file is empty"}, status=400
)
model = self.db.model
ok, reason = await self._run_in_executor(
model.verify_database_file, temp_path
)
if ok:
reason = await self._run_in_executor(_inspect_restore_upload, temp_path)
ok = reason is None
if not ok:
return web.json_response(
{"success": False, "error": reason}, status=400
)
try:
backup_path = await self._run_in_executor(
model.restore_from_file, temp_path
)
except sqlite3.OperationalError as exc:
if (
getattr(exc, "_prompt_manager_restore_backup_created", False)
is True
):
try:
await self._run_in_executor(
_prune_safety_backups, model.db_path
)
except Exception:
self.logger.warning(
"Could not prune safety backups after failed restore"
)
raise
removed = await self._run_in_executor(_prune_safety_backups, model.db_path)
if removed:
self.logger.info(f"Removed {len(removed)} old safety backups")
previous = self.db
self.db = PromptDatabase(model.db_path)
self._database_replaced() # a health report described the old one
# Its connections would otherwise linger (a file lock on Windows)
previous.close_all()
prompt_count = self.db.model.get_database_info().get("total_prompts", 0)
return web.json_response(
{
"success": True,
"message": (
"Database restored successfully. "
f"Found {prompt_count} prompts."
),
"prompt_count": prompt_count,
"backup_created": (
self._public_path(backup_path) if backup_path else None
),
}
)
except Exception as e:
self.logger.error(f"Restore error: {e}", exc_info=True)
return web.json_response(
{
"success": False,
"error": f"Failed to restore database: {self._public_error(e)}",
},
status=500,
)
finally:
if temp_path:
for suffix in ("", "-wal", "-shm", "-journal"):
if os.path.exists(temp_path + suffix):
os.unlink(temp_path + suffix)
async def scan_images(self, request):
"""Scan ComfyUI output images for prompt metadata and add them to the database.
py/api/backup_restore.py
+937
View File
@@ -0,0 +1,937 @@
"""Database backup and restore routes (#249 System S8).
Legacy routes, responses unchanged:
- ``GET /prompt_manager/backup`` downloads a consistent copy of the library.
While the copy is made it holds the registry as ``backup`` ("Database
backup", ``guarded``): it is refused while a restore, maintenance apply or
other record-changing job runs. The copy is written in the library's
folder and made one standalone file; the hold ends there, so a slow or
stalled download never keeps other work waiting. A copy that fails its
integrity check is a 500 with no bytes. One that fails only a semantic
check (``verify_database_file``) is still sent: a user can always get a
backup. The reason is logged and sent as ``X-PromptManager-Verify``
(``ok`` otherwise), and the status records it as not verified. The file
is streamed, then closed and removed.
- ``POST /prompt_manager/restore`` (multipart ``database_file``) restores in
one request through the same stage and publish as the restore job, holding
the registry as ``restore`` (``exclusive``).
Reviewed restore (new routes answer errors as ``{success:false, code, error}``):
- ``POST /prompt_manager/restore/preview`` (multipart ``database_file``, the
same disk-space cap) stages the upload in the library's folder
(``database/restore_stage.py``) and answers what it holds beside the live
library: ``{token, expires_at, summary, schema:{migrated, library_state},
integrity:"ok", documents:{repaired, cleared, images}, live, size,
safety_backup:{destination_label}}``. Nothing live is written or backed
up. At most two previews are held (``too-large``, 429).
- ``DELETE /prompt_manager/restore/preview/{token}`` discards a stage.
- ``POST /prompt_manager/restore/apply {token}`` starts the restore job (202
``{job}``): kind ``restore``, "Database restore", ``exclusive`` (it runs
alone, like Move library), not cancellable, indeterminate. The runner waits
for request leases to end, rechecks the live fingerprint (``plan-changed``),
publishes the exact staged bytes, swaps ``self.db``, prunes old safety
backups and logs what was restored. Once the copy replaced the library,
the restore is done: the later steps only log their failures.
- ``GET /prompt_manager/backup/status``: the safety backups on disk
(``<db>.backup_*``, newest first, names only) and this session's last
verified download (Q9: no sidecar file).
"""
import asyncio
import datetime
import os
import sqlite3
import tempfile
from contextlib import closing
from pathlib import Path
from aiohttp import web
from . import admin as _admin
from .job_routes import busy_response
from .jobs import RESTORE_JOB_KIND, JobBusy
from .maintenance_plans import PlanExpired, PlanUsed
from .restore_plans import RestorePlanStore
try:
from ...database import restore_stage
from ...database.operations import PromptDatabase
from ...database.safety_backup import (
SafetyBackupFailed,
fsync_file,
make_standalone,
remove_quietly,
safety_temp,
)
except ImportError:
from database import restore_stage
from database.operations import PromptDatabase
from database.safety_backup import (
SafetyBackupFailed,
fsync_file,
make_standalone,
remove_quietly,
safety_temp,
)
BACKUP_JOB_KIND = "backup"
BACKUP_HOLD_TITLE = "Database backup"
LEGACY_RESTORE_TITLE = "Database restore (admin page)"
RESTORE_JOB_TITLE = "Database restore"
RESTORE_GROUPS = ("errors",)
MAX_TOKEN_LENGTH = 128
APPLY_BODY_MAX = 1024
STREAM_CHUNK = 1024 * 1024
MAX_LISTED_BACKUPS = 10
SIDECARS = ("", "-wal", "-shm", "-journal")
REQUESTS_BUSY = (
"Other requests are still reading the library; nothing was changed. "
"Try again in a moment."
)
REVIEW_KEPT = (
"Other requests are still reading the library; nothing was changed. "
"The review is kept: start the restore again in a moment."
)
VERIFY_HEADER = "X-PromptManager-Verify"
SQLITE_HEADER = b"SQLite format 3\x00"
PLAN_CHANGED = (
"The library changed since the review; nothing was changed. "
"Review the restore again."
)
NOT_RESTORED = "The restore failed; the library was not changed."
def _failure(message, status, code=None, **extra):
return web.json_response(
{"success": False, "code": code, "error": message, **extra}, status=status
)
def _iso(timestamp):
return datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc).isoformat()
def _plural(count, word):
return f"{count:,} {word}" + ("" if count == 1 else "s")
def documents_line(documents):
"""The repair report's sentence, or "" when nothing was damaged."""
repaired, cleared = documents.get("repaired", 0), documents.get("cleared", 0)
parts = []
if repaired:
parts.append(f"repaired {_plural(repaired, 'damaged workflow document')}")
if cleared:
images = documents.get("images", 0)
parts.append(
f"could not repair {_plural(cleared, 'workflow document')}: "
f"{_plural(images, 'image')} {'shows' if images == 1 else 'show'} "
"no workflow"
)
if not parts:
return ""
text = "; ".join(parts)
return text[0].upper() + text[1:] + "."
def restored_line(summary):
return (
f"Restored {_plural(summary['prompts'], 'prompt')}, "
f"{_plural(summary['image_links'], 'image link')}, "
f"{_plural(summary['pairs'], 'saved pair')} and "
f"{_plural(summary['runs'], 'run')}"
)
def safety_backups(db_path, limit=MAX_LISTED_BACKUPS):
"""``<db>.backup_*`` files beside the database, newest first (blocking)."""
db_path = os.path.abspath(db_path)
folder = os.path.dirname(db_path)
prefix = os.path.basename(db_path) + ".backup_"
found = []
try:
names = os.listdir(folder)
except OSError:
return []
for name in names:
if not name.startswith(prefix):
continue
try:
stat = os.stat(os.path.join(folder, name))
except OSError:
continue
if not os.path.isfile(os.path.join(folder, name)):
continue
found.append((stat.st_mtime, name, stat.st_size))
found.sort(key=lambda item: (item[0], item[1]), reverse=True)
return [
{"name": name, "bytes": size, "modified_at": _iso(mtime)}
for mtime, name, size in found[:limit]
]
def snapshot_corruption(path):
"""Why the snapshot file is corrupt, or None (blocking).
Only physical damage refuses a download: not a SQLite file, or a failed
``PRAGMA integrity_check``.
"""
try:
with open(path, "rb") as handle:
if handle.read(len(SQLITE_HEADER)) != SQLITE_HEADER:
return "Not a SQLite database file"
uri = Path(os.path.abspath(path)).as_uri() + "?mode=ro"
with closing(sqlite3.connect(uri, uri=True)) as conn:
row = conn.execute("PRAGMA integrity_check").fetchone()
except (OSError, sqlite3.Error) as error:
return f"Invalid SQLite database: {error}"
verdict = row[0] if row else "no result"
return None if verdict == "ok" else f"Integrity check failed: {verdict}"
def write_backup_snapshot(model):
"""A standalone copy of the library beside it: ``(path, warning)`` (blocking).
Raises SafetyBackupFailed with nothing left on disk when the copy cannot
be written or is corrupt (``snapshot_corruption``). ``warning`` is the
reason a semantic check (``verify_database_file``, workflow documents
not read) refused the copy, or None. The caller removes the file.
"""
temp = safety_temp(model.db_path)
try:
if not model.backup_database(temp):
raise SafetyBackupFailed("The backup could not be written")
make_standalone(temp)
corrupt = snapshot_corruption(temp)
if corrupt is not None:
raise SafetyBackupFailed("The backup is corrupt", corrupt)
ok, reason = model.verify_database_file(temp, documents=False)
fsync_file(temp)
return temp, (None if ok else reason)
except BaseException:
remove_quietly(*(temp + suffix for suffix in SIDECARS))
raise
def _header_text(text):
"""A reason as a one-line, latin-1 header value."""
flat = " ".join(str(text).split())[:300]
return flat.encode("latin-1", "replace").decode("latin-1")
class BackupRestoreRoutesMixin:
"""Mixin providing the backup and restore endpoints."""
def _register_backup_restore_routes(self, routes):
@routes.post("/prompt_manager/restore/preview")
async def restore_preview_route(request):
return await self.preview_restore(request)
@routes.delete("/prompt_manager/restore/preview/{token}")
async def restore_discard_route(request):
return await self.discard_restore_preview(request)
@routes.post("/prompt_manager/restore/apply")
async def restore_apply_route(request):
return await self.apply_restore(request)
@routes.get("/prompt_manager/backup/status")
async def backup_status_route(request):
return await self.backup_status(request)
@property
def _restore_plans(self):
store = getattr(self, "_restore_plan_store", None)
if store is None:
store = self._restore_plan_store = RestorePlanStore()
return store
# ── backup ────────────────────────────────────────────────────────
async def backup_database(self, request):
"""Download a copy of the prompts database (WAL included)."""
return await self._stream_backup(request)
async def _stream_backup(self, request):
model = self.db.model
if not os.path.exists(model.db_path):
return web.json_response(
{"success": False, "error": "Database file not found"}, status=404
)
made = await self._run_held(
BACKUP_JOB_KIND,
BACKUP_HOLD_TITLE,
"guarded",
lambda: self._make_backup_snapshot(model),
)
if isinstance(made, web.StreamResponse):
return made # refused, or the snapshot failed
temp, warning = made
return await self._send_snapshot(request, temp, warning)
async def _make_backup_snapshot(self, model):
"""``(path, warning)``, or the error response (inside the hold)."""
loop = asyncio.get_running_loop()
future = loop.run_in_executor(None, write_backup_snapshot, model)
try:
temp, warning = await asyncio.shield(future)
except asyncio.CancelledError:
# The worker cannot be stopped: remove what it writes once it is
# done (#335 re-review R4).
future.add_done_callback(
lambda done: loop.create_task(self._remove_finished_snapshot(done))
)
raise
except SafetyBackupFailed as failed:
self.logger.error(f"Backup failed: {failed} {failed.reason or ''}")
return web.json_response(
{"success": False, "error": "Failed to create database backup"},
status=500,
)
except Exception as e:
self.logger.error(f"Backup error: {e}", exc_info=True)
return web.json_response(
{
"success": False,
"error": f"Failed to backup database: {self._public_error(e)}",
},
status=500,
)
if warning is not None:
self.logger.warning(f"Backup sent with a failed check: {warning}")
return temp, warning
async def _remove_finished_snapshot(self, future):
"""Remove the snapshot a cancelled download's worker finished."""
if future.cancelled() or future.exception() is not None:
return # nothing was left on disk
temp, _warning = future.result()
await self._run_in_executor(
remove_quietly, *(temp + suffix for suffix in SIDECARS)
)
async def _send_snapshot(self, request, temp, warning):
"""Stream the snapshot (outside the hold), then close and remove it."""
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
handle = None
try:
size = os.path.getsize(temp)
handle = open(temp, "rb")
response = web.StreamResponse(
headers={
"Content-Type": "application/octet-stream",
"Content-Disposition": (
f'attachment; filename="prompts_backup_{timestamp}.db"'
),
"Content-Length": str(size),
VERIFY_HEADER: "ok" if warning is None else _header_text(warning),
}
)
await response.prepare(request)
while True:
chunk = await self._run_in_executor(handle.read, STREAM_CHUNK)
if not chunk:
break
await response.write(chunk)
await response.write_eof()
self._last_backup_download = {
"at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"bytes": size,
"verified": warning is None,
}
return response
finally:
if handle is not None:
handle.close() # before the unlink (Windows)
await self._run_in_executor(
remove_quietly, *(temp + suffix for suffix in SIDECARS)
)
async def backup_status(self, request):
try:
backups = await self._run_in_executor(safety_backups, self.db.model.db_path)
except Exception:
self.logger.exception("Backup status failed")
return _failure("Failed to read the backup status", 500)
return web.json_response(
{
"success": True,
"safety_backups": backups,
"last_download": getattr(self, "_last_backup_download", None),
},
headers={"Cache-Control": "no-store"},
)
# ── the upload ────────────────────────────────────────────────────
def _restore_refusal(self, size, ceiling, room, *, known):
"""``400 too-large`` past the ceiling, else ``507 insufficient-space``."""
from .maintenance import restore_space_message
if size > ceiling:
return web.json_response(
{
"success": False,
"code": "too-large",
"error": (
"File too large. Maximum size is "
f"{ceiling // (1024 * 1024)}MB"
),
},
status=400,
)
return web.json_response(
{
"success": False,
"code": "insufficient-space",
"error": restore_space_message(size, room, known=known),
},
status=507,
)
async def _receive_restore_upload(self, request, uploads):
"""Stream ``database_file`` into the library's folder.
Returns ``(path, None)`` or ``(None, refusal)``. Each temp path is
appended to ``uploads`` as soon as it exists, so the caller removes
it (and its sidecars) whatever happens.
"""
from .maintenance import RESTORE_MAX_BYTES, restore_room
ceiling = getattr(self, "restore_max_bytes", RESTORE_MAX_BYTES)
db_path = self.db.model.db_path
# Free disk space decides how large a backup may be (#313): the
# upload, its staged copy and the swap must all fit.
room = await self._run_in_executor(restore_room, db_path)
max_bytes = min(ceiling, room)
length = request.content_length
if length is not None and length > max_bytes:
return None, self._restore_refusal(length, ceiling, room, known=True)
reader = await request.multipart()
field = await reader.next()
if not field or field.name != "database_file":
return None, web.json_response(
{
"success": False,
"error": (
"No database file uploaded. "
"Expected field name: database_file"
),
},
status=400,
)
fd, temp_path = tempfile.mkstemp(
prefix=restore_stage.UPLOAD_PREFIX,
suffix=".tmp",
dir=os.path.dirname(os.path.abspath(db_path)),
)
uploads.append(temp_path)
size = 0
with os.fdopen(fd, "wb") as out:
while True:
chunk = await field.read_chunk()
if not chunk:
break
size += len(chunk)
if size > max_bytes:
return None, self._restore_refusal(size, ceiling, room, known=False)
out.write(chunk)
if size == 0:
return None, web.json_response(
{"success": False, "error": "Uploaded file is empty"}, status=400
)
return temp_path, None
async def _upload_problem(self, path):
"""None when the upload may be staged, else a readable reason."""
ok, reason = await self._run_in_executor(
self.db.model.verify_database_file, path, documents=False
)
if ok:
reason = await self._run_in_executor(_admin._inspect_restore_upload, path)
ok = reason is None
return None if ok else reason
async def _remove_uploads(self, uploads):
for path in uploads:
await self._run_in_executor(
remove_quietly, *(path + suffix for suffix in SIDECARS)
)
async def _prune_after(self, db_path, *, failed=False):
"""Prune old safety backups; a failure is only logged (never raises).
It runs after a restore replaced the library, or after one failed:
either way the restore's outcome stands.
"""
try:
removed = await self._run_in_executor(_admin._prune_safety_backups, db_path)
except Exception as error:
when = "failed" if failed else "completed"
self.logger.warning(
f"Could not prune safety backups after a {when} restore: "
f"{type(error).__name__}"
)
return
if removed:
self.logger.info(f"Removed {len(removed)} old safety backups")
async def _publish(self, model, staged):
"""Publish ``staged``; on a page-copy failure prune and re-raise."""
try:
return await self._run_in_executor(model.publish_staged, staged)
except sqlite3.OperationalError as exc:
if getattr(exc, "_prompt_manager_restore_backup_created", False) is True:
await self._prune_after(model.db_path, failed=True)
raise
async def _after_publish(self, db_path):
"""Everything after the copy replaced the library: best effort, logged.
The restore is done once the copy succeeded, so nothing here fails it
(#335 review I1). The library is adopted as ``self.db`` (the previous
object, which points at the same file, stays when a new one cannot
be opened), the generation is bumped (so the health report and
reviewed plans of the old library are dropped), held restore
previews are dropped and the previous object's connections closed.
"""
await self._prune_after(db_path)
previous = self.db
try:
self.db = PromptDatabase(db_path)
except Exception:
self.logger.exception("Could not reopen the restored library")
try:
self._database_replaced()
except Exception:
self.logger.exception("Could not drop the old library's reports")
try:
await self._discard_held(self._restore_plans.pop_all())
except Exception:
self.logger.exception("Could not remove the old restore previews")
if previous is not self.db:
try:
# Its connections would otherwise linger (a lock on Windows).
await self._run_in_executor(previous.close_all)
except Exception:
self.logger.exception("Could not close the previous library")
async def _discard_held(self, held):
"""Remove the stage folders of dropped store entries (off the loop)."""
for entry in held:
try:
await self._run_in_executor(entry.staged.discard)
except Exception:
self.logger.exception("Could not remove a restore stage")
async def _discard_popped(self, token):
"""Drop ``token`` and remove its stage, only if the store still held it.
A stage another apply's job has taken is not this request's to remove
(#335 re-review R2).
"""
popped = self._restore_plans.pop(token)
if popped is not None:
await self._discard_held([popped])
async def _drop_expired_restores(self):
"""Drop expired previews from the store (on the loop), then their files."""
await self._discard_held(self._restore_plans.pop_expired())
def _schedule_restore_expiry(self):
"""Remove each preview's stage when it expires, without a request.
A loop timer (``call_later``), so a server that stops with previews
held leaves nothing pending; the startup sweep removes the folders.
"""
store = self._restore_plans
expires = store.next_expiry()
if expires is None:
return
handle = getattr(self, "_restore_expiry_handle", None)
if handle is not None:
handle.cancel()
loop = asyncio.get_running_loop()
def fire():
self._restore_expiry_handle = None
task = loop.create_task(self._expire_restores())
self._restore_expiry_task = task
delay = max(0.0, expires - store.now())
self._restore_expiry_handle = loop.call_later(delay, fire)
async def _expire_restores(self):
await self._drop_expired_restores()
self._schedule_restore_expiry()
# ── legacy one-step restore ───────────────────────────────────────
async def restore_database(self, request):
"""Restore the prompts database from an uploaded SQLite file.
The upload is streamed to a temp file in the library's folder. Free
disk space decides how large it may be (``maintenance.restore_room``),
under a 64 GiB ceiling (``restore_max_bytes``): a request whose
Content-Length does not fit is refused before anything is written,
and one without is refused as soon as its running size does not. The
file is checked, staged, repaired and verified, then published over
the live database, which is backed up first. The whole request holds
the registry as ``restore`` (``exclusive``): it is refused while any
job runs, and every job and request lease is refused while it runs.
"""
return await self._run_held(
RESTORE_JOB_KIND,
LEGACY_RESTORE_TITLE,
"exclusive",
lambda: self._restore_database(request),
)
async def _restore_database(self, request):
uploads = []
try:
temp_path, refused = await self._receive_restore_upload(request, uploads)
if refused is not None:
return refused
reason = await self._upload_problem(temp_path)
if reason is not None:
return web.json_response(
{"success": False, "error": reason}, status=400
)
if not await self._jobs.drain_leases():
return web.json_response(
{"success": False, "error": REQUESTS_BUSY}, status=409
)
model = self.db.model
staged = await self._run_in_executor(model.stage_restore, temp_path)
try:
backup_path = await self._publish(model, staged)
finally:
await self._run_in_executor(staged.discard)
# The library is replaced: from here on the restore succeeded.
await self._after_publish(model.db_path)
try:
prompt_count = await self._run_in_executor(
lambda: self.db.model.get_database_info().get("total_prompts", 0)
)
except Exception:
self.logger.exception("Could not count the restored prompts")
prompt_count = staged.summary.get("prompts", 0)
message = f"Database restored successfully. Found {prompt_count} prompts."
body = {
"success": True,
"message": message,
"prompt_count": prompt_count,
"backup_created": (
self._public_path(backup_path) if backup_path else None
),
}
repaired = documents_line(staged.documents)
if repaired:
body["message"] = f"{message} {repaired}"
body["workflow_documents"] = dict(staged.documents)
return web.json_response(body)
except Exception as e:
self.logger.error(f"Restore error: {e}", exc_info=True)
return web.json_response(
{
"success": False,
"error": f"Failed to restore database: {self._public_error(e)}",
},
status=500,
)
finally:
await self._remove_uploads(uploads)
# ── reviewed restore ──────────────────────────────────────────────
def _live_fingerprint(self, db):
return restore_stage.live_fingerprint(db.model.get_connection())
async def preview_restore(self, request):
"""Stage an upload for review; nothing live is written (see module)."""
await self._drop_expired_restores()
if self._restore_plans.full():
return _too_many_reviews()
uploads = []
try:
db = self.db
identity = self._database_identity(db)
temp_path, refused = await self._receive_restore_upload(request, uploads)
if refused is not None:
return refused
return await self._review_upload(db, identity, temp_path, uploads)
except Exception:
self.logger.exception("Restore preview failed")
return _failure("Failed to review the restore", 500)
finally:
await self._remove_uploads(uploads)
async def _stage_for_review(self, db, temp_path, uploads):
"""Stage ``temp_path`` in the executor; a cancel discards the result.
A cancelled handler (shutdown, a dropped client) cannot stop the
worker thread: the stage it finishes is removed when it is done, and
so is the upload it reads (``_orphan_stage_cleanup``).
"""
loop = asyncio.get_running_loop()
future = loop.run_in_executor(None, db.model.stage_restore, temp_path)
try:
return await asyncio.shield(future)
except asyncio.CancelledError:
orphans = getattr(self, "_orphan_stages", None)
if orphans is None:
orphans = self._orphan_stages = set()
entry = (future, tuple(uploads))
orphans.add(entry)
uploads.clear() # still being read: removed with the stage
future.add_done_callback(
lambda _f: loop.create_task(self._orphan_stage_cleanup())
)
raise
async def _orphan_stage_cleanup(self):
"""Remove what cancelled previews staged, once their workers are done."""
for entry in list(getattr(self, "_orphan_stages", ()) or ()):
future, uploads = entry
if not future.done():
continue
self._orphan_stages.discard(entry)
if not future.cancelled() and future.exception() is None:
await self._discard_held([_Orphan(future.result())])
await self._remove_uploads(uploads)
async def _review_upload(self, db, identity, temp_path, uploads=None):
"""Check, stage and compare an upload; the preview response.
Only the read of the live library takes a lease (#335 review M2):
the upload and the staging never touch it, and a lease held for a
long staging would make a restore's drain fail. A stage that is not
handed to the store (a refusal, an error, a cancel) is removed.
"""
uploads = [] if uploads is None else uploads # the caller removes them
reason = await self._upload_problem(temp_path)
if reason is not None:
return _failure(reason, 400, "invalid-backup")
staged = None
issued = False
try:
try:
staged = await self._stage_for_review(db, temp_path, uploads)
except ValueError as invalid:
return _failure(str(invalid), 400, "invalid-backup")
try:
async with self._jobs.lease("Restore preview"):
live, fingerprint = await self._run_in_executor(
self._live_fingerprint, db
)
except JobBusy as busy:
return busy_response(busy)
store = self._restore_plans
if identity != self._database_identity(self.db):
return _failure(
"The database was replaced while the restore was reviewed. "
"Review again.",
409,
"storage-changed",
)
if store.full():
return _too_many_reviews()
token = store.issue(staged, fingerprint, identity, live)
issued = True
self._schedule_restore_expiry()
finally:
if staged is not None and not issued:
await self._discard_held([_Orphan(staged)])
held = store.entry(token)
name = os.path.basename(db.model.db_path)
return web.json_response(
{
"success": True,
"token": token,
"expires_at": _iso(held.expires_at),
"summary": dict(staged.summary),
"schema": dict(staged.schema),
"integrity": "ok",
"documents": dict(staged.documents),
"live": dict(live),
"size": staged.size,
"safety_backup": {
"destination_label": f"{name}.backup_<date and time>"
},
},
headers={"Cache-Control": "no-store"},
)
async def discard_restore_preview(self, request):
token = request.match_info.get("token", "")
if not 0 < len(token) <= MAX_TOKEN_LENGTH:
return _failure("Unknown restore review.", 410, "plan-expired")
held = self._restore_plans.pop(token)
if held is None:
await self._drop_expired_restores()
return _failure("Unknown restore review.", 410, "plan-expired")
await self._discard_held([held])
return web.json_response({"success": True})
def _peek_restore(self, token):
try:
return self._restore_plans.peek(token), None
except PlanUsed as used:
job = self._jobs.get(used.job_id)
return None, _failure(
"This restore was already started.",
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 restore again.",
410,
"plan-expired",
)
async def apply_restore(self, request):
data, error = await self._read_json_body(request, max_bytes=APPLY_BODY_MAX)
if error is not None:
if error.status == 413:
return _failure("Request body too large", 413, "too-large")
return _failure(
"Request body must be a JSON object", 400, "invalid-request"
)
token = data.get("token")
if set(data) != {"token"} or not isinstance(token, str):
return _failure("token must be the review's token", 400, "invalid-request")
if not 0 < len(token) <= MAX_TOKEN_LENGTH:
return _failure("token must be the review's token", 400, "invalid-request")
await self._drop_expired_restores()
held, refused = self._peek_restore(token)
if refused is not None:
await self._drop_expired_restores() # one peek found expired
return refused
db = self.db
if held.db_identity != self._database_identity(db):
await self._discard_popped(token)
return _failure(
"The database was replaced since the review. Review again.",
409,
"storage-changed",
)
try:
async with self._jobs.lease("Restore review check"):
_live, fingerprint = await self._run_in_executor(
self._live_fingerprint, db
)
except JobBusy as busy:
return busy_response(busy)
if fingerprint != held.fingerprint:
# It can never match again: free the stage for a new review.
await self._discard_popped(token)
return _failure(PLAN_CHANGED, 409, "plan-changed")
# The token may have been used, expired or discarded meanwhile.
held, refused = self._peek_restore(token)
if refused is not None:
await self._drop_expired_restores()
return refused
if db is not self.db:
return _failure(
"The database was replaced since the review. Review again.",
409,
"storage-changed",
)
try:
job = self._jobs.start(
RESTORE_JOB_KIND,
RESTORE_JOB_TITLE,
# It replaces the whole database: it runs alone, beside no
# job of any class, like Move library.
admission="exclusive",
cancellable=False,
scope={},
groups=RESTORE_GROUPS,
runner=lambda job: self._run_restore(job, db, held, token),
)
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.
self._restore_plans.take(token, job.id)
return web.json_response({"success": True, "job": job.summary()}, status=202)
async def _run_restore(self, job, db, held, token=None):
from .maintenance import _fail_job
staged = held.staged
keep_stage = False
try:
job.update(status_line="Waiting for other requests to finish")
if not await self._jobs.drain_leases():
# Nothing ran: the reviewed stage stays usable (#335 M2),
# unless it expired meanwhile or its slot was reused.
if token is not None and self._restore_plans.give_back(token, held):
keep_stage = True
self._schedule_restore_expiry()
return _fail_job(job, "busy", REVIEW_KEPT)
return _fail_job(job, "busy", REQUESTS_BUSY)
job.update(status_line="Checking the library")
_live, fingerprint = await self._run_in_executor(self._live_fingerprint, db)
if db is not self.db or fingerprint != held.fingerprint:
return _fail_job(job, "plan-changed", PLAN_CHANGED)
job.update(status_line="Restoring the database")
job.log_line("Restoring the reviewed backup")
model = db.model
try:
backup_path = await self._publish(model, staged)
except restore_stage.StageChanged:
return _fail_job(job, "plan-changed", PLAN_CHANGED)
except SafetyBackupFailed:
return _fail_job(
job,
"backup-failed",
"The current library could not be backed up; nothing was "
"changed.",
)
except sqlite3.OperationalError as error:
kept = (
" The safety backup was kept."
if getattr(error, "_prompt_manager_restore_backup_created", False)
else ""
)
self.logger.error(f"Restore failed: {error}")
return _fail_job(job, "restore-failed", NOT_RESTORED + kept)
except Exception:
self.logger.exception("Restore job failed")
return _fail_job(job, None, NOT_RESTORED)
finally:
if not keep_stage:
await self._discard_held([_Orphan(staged)])
# The copy replaced the library: the restore is done (#335 review I1).
# Every later step only logs a failure.
await self._after_publish(model.db_path)
if backup_path:
job.log_line(f"Safety backup {os.path.basename(backup_path)}")
else:
job.log_line("No safety backup: there was no library to keep")
repaired = documents_line(staged.documents)
if repaired:
job.log_line(repaired)
line = restored_line(staged.summary)
job.log_line(line)
job.update(progress=100, status_line=line)
class _Orphan:
"""A stage with no store entry, in the shape ``_discard_held`` takes."""
def __init__(self, staged):
self.staged = staged
def _too_many_reviews():
return _failure(
"Two restores are already being reviewed. Discard one first.",
429,
"too-large",
)
py/api/jobs.py
+100
View File
@@ -28,6 +28,15 @@ ones. One job of a kind runs at a time, and each job has an admission class
A short legacy-page operation takes a ``hold``: a running, non-cancellable
job that lasts as long as the request, so the Activity page lists it and a
workspace job is refused while it runs (and the other way round).
Short request-scoped work that reads or writes the library beside any
ordinary job (a Needs review decision, a maintenance or restore preview)
takes a ``lease`` instead: it is not listed (so quick requests never push
real jobs out of the finished history, and two of them never refuse each
other), but it is refused while an ``exclusive`` job runs, and an
``exclusive`` job that replaces the database (a restore) awaits
``drain_leases`` before it switches the file, so no lease's connection is
open across the switch (#249 S8).
"""
import asyncio
@@ -48,6 +57,9 @@ MAX_OUTCOME_ROWS = 200 # rows listed per outcome group; counters stay exact
MAX_LOG_LINES = 2000
LOG_CAP_LINE = "[Log capped at 2,000 lines]"
MAX_FINISHED_JOBS = 20 # finished jobs this session keeps
# The longest an exclusive job waits for running leases to end (a ceiling a
# hung request reaches, not a pacing timer: leases end with their request).
LEASE_DRAIN_SECONDS = 60
# The metadata scan's outcome groups, in counter-tile order of the spec's
# mapping, and the per-file scan status each group collects. Files without
@@ -403,6 +415,74 @@ class JobRegistry:
def __init__(self, logger_=None):
self._logger = logger_ or logger
self._jobs = {}
self._leases = {} # title -> running count
self._drain_waiters = []
# ── leases ────────────────────────────────────────────────────────
def lease(self, title):
"""An async context manager for short request-scoped library work.
Entering raises JobBusy, naming the job, while an ``exclusive`` job
runs; otherwise the lease is counted until the block exits. Leases
are never listed and never refuse one another or any job.
"""
return _Lease(self, title)
def lease_blocker(self):
"""The running ``exclusive`` job a new lease is refused for, or None."""
for job in self.running():
if job.admission == "exclusive":
return job
return None
@property
def leases(self):
"""Running leases by title (a copy)."""
return dict(self._leases)
def _take_lease(self, title):
blocker = self.lease_blocker()
if blocker is not None:
raise JobBusy(blocker)
self._leases[title] = self._leases.get(title, 0) + 1
def _release_lease(self, title):
count = self._leases.get(title, 0) - 1
if count > 0:
self._leases[title] = count
else:
self._leases.pop(title, None)
if not self._leases:
waiters, self._drain_waiters = self._drain_waiters, []
for waiter in waiters:
if not waiter.done():
waiter.set_result(True)
async def drain_leases(self, timeout=None):
"""Wait until no lease runs; False when ``timeout`` ended the wait.
For an ``exclusive`` job about to switch the database: once it runs
no new lease is admitted, so the count only goes down.
"""
if not self._leases:
return True
waiter = asyncio.get_running_loop().create_future()
self._drain_waiters.append(waiter)
try:
await asyncio.wait_for(
asyncio.shield(waiter),
LEASE_DRAIN_SECONDS if timeout is None else timeout,
)
return True
except asyncio.TimeoutError:
self._logger.warning(
"Still running after the wait: " + ", ".join(sorted(self._leases))
)
return False
finally:
if waiter in self._drain_waiters:
self._drain_waiters.remove(waiter)
def start(
self,
@@ -565,3 +645,23 @@ class _Hold:
if self._job.running:
self._job.finish("finished" if exc_type is None else "failed")
return False
class _Lease:
"""The async context manager ``JobRegistry.lease`` returns."""
def __init__(self, registry, title):
self._registry = registry
self._title = title
self._held = False
async def __aenter__(self):
self._registry._take_lease(self._title)
self._held = True
return self
async def __aexit__(self, exc_type, exc, tb):
if self._held:
self._held = False
self._registry._release_lease(self._title)
return False
py/api/maintenance.py
+36 -168
View File
@@ -48,8 +48,6 @@ import json
import os
import shutil
import sqlite3
import tempfile
from contextlib import closing
from aiohttp import web
@@ -59,15 +57,28 @@ from .jobs import MAINTENANCE_JOB_KIND, JobBusy
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"
@@ -96,18 +107,6 @@ PLAN_CHANGED = (
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"
_NAME_ATTEMPTS = 1000
class SafetyBackupFailed(Exception):
"""The safety backup could not be written, verified or put in place.
``reason`` is the verify reason when the copy did not verify.
"""
def __init__(self, message, reason=None):
super().__init__(message)
self.reason = reason
def backup_failed_message(error):
@@ -118,10 +117,6 @@ def backup_failed_message(error):
return BACKUP_FAILED
def _now():
return datetime.datetime.now()
def _free_bytes(path):
return shutil.disk_usage(path).free
@@ -244,29 +239,18 @@ def size_label(size):
def restore_room(db_path):
"""The largest restore upload the disks have room for (blocking).
"""The largest restore upload the disk has room for (blocking).
The upload is written to a temp file and staged once more in the
temporary folder (2x), then copied over the live database beside it,
whose current contents are first kept as a safety backup (1x the upload
for the WAL, plus the current database). Needs on one device add up,
each with ``DISK_MARGIN_BYTES``.
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
devices = {}
for path, per_upload, fixed in (
(tempfile.gettempdir(), 2, 0),
(folder, 1, current),
):
entry = devices.setdefault(_device(path), {"path": path, "per": 0, "fixed": 0})
entry["per"] += per_upload
entry["fixed"] += fixed
room = min(
(_free_bytes(e["path"]) - e["fixed"] - DISK_MARGIN_BYTES) // e["per"]
for e in devices.values()
)
room = (_free_bytes(folder) - current - DISK_MARGIN_BYTES) // 3
return max(0, room)
@@ -306,142 +290,13 @@ def _changed_line(operation, count):
return f"Removed {count:,} {noun}"
def _remove_quietly(*paths):
for path in paths:
try:
if path and os.path.exists(path):
os.remove(path)
except OSError:
pass
def _reserve_backup_name(db_path, label="backup"):
"""Create and return an unused ``<db>.<label>_<stamp>[_n]`` (O_EXCL)."""
for candidate in _backup_names(db_path, label):
try:
fd = os.open(candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
continue
os.close(fd)
return candidate
raise SafetyBackupFailed("No free backup name")
def _backup_names(db_path, label="backup"):
base = f"{db_path}.{label}_{_now().strftime('%Y%m%d_%H%M%S_%f')}"
for attempt in range(_NAME_ATTEMPTS):
yield base if attempt == 0 else f"{base}_{attempt}"
def _place_backup(temp, db_path, label="backup"):
"""Give the verified ``temp`` its unique backup name; return that path.
``os.link`` is atomic and refuses an existing name, so no empty or half
file ever carries a backup name, even across a crash. A filesystem
without hard links falls back to reserving the name (``O_EXCL``) and
``os.replace``. The temp file is removed by the caller.
"""
for candidate in _backup_names(db_path, label):
try:
os.link(temp, candidate)
except FileExistsError:
continue
except (OSError, NotImplementedError):
break # no hard links here
else:
return candidate
else:
raise SafetyBackupFailed("No free backup name")
final = _reserve_backup_name(db_path, label)
try:
os.replace(temp, final)
except OSError:
_remove_quietly(final)
raise
return final
SAFETY_TEMP_PREFIX = ".pm-safety-"
def sweep_safety_temps(db_path):
"""Remove the temp copies an interrupted safety backup left (blocking).
Called once at startup, before any job can take a backup. Returns the
names removed.
"""
folder = os.path.dirname(os.path.abspath(db_path))
removed = []
try:
names = os.listdir(folder)
except OSError:
return removed
for name in names:
if not (name.startswith(SAFETY_TEMP_PREFIX) and ".tmp" in name):
continue
try:
os.remove(os.path.join(folder, name))
removed.append(name)
except OSError:
continue
return removed
def _standalone(path):
"""Make the copy one self-contained file (rollback journal, not WAL).
The backup API copies the live database's WAL flag; a WAL file opened
read-only later (a verify, a restore preview) would leave ``-wal`` and
``-shm`` sidecars beside it.
"""
try:
with closing(sqlite3.connect(path)) as conn:
conn.execute("PRAGMA journal_mode=DELETE").fetchone()
except sqlite3.Error as exc:
raise SafetyBackupFailed("The backup could not be finished") from exc
def _fsync(path):
with open(path, "rb+") as handle:
os.fsync(handle.fileno())
def take_safety_backup(model, label="backup"):
"""Write and verify a safety backup; return its file name (blocking).
The copy is made in a temp file beside the database and only renamed
onto its final, unique name ``<db>.<label>_<stamp>`` once it verified,
so a backup file on disk is always a verified one. ``label`` "backup"
joins the pruned pool; another label (the workflow storage migration's
``pre-workflow-storage``) is kept until the user removes it. Raises
SafetyBackupFailed, leaving no file.
See ``database/safety_backup.py``: a backup file on disk is always a
verified one. Raises SafetyBackupFailed, leaving no file.
"""
db_path = os.path.abspath(model.db_path)
fd, temp = tempfile.mkstemp(
dir=os.path.dirname(db_path), prefix=SAFETY_TEMP_PREFIX, suffix=".tmp"
)
os.close(fd)
final = None
try:
if not model.backup_database(temp):
raise SafetyBackupFailed("The backup could not be written")
_standalone(temp)
ok, reason = model.verify_database_file(temp)
if not ok:
raise SafetyBackupFailed("The backup did not verify", reason)
_fsync(temp)
if os.name != "nt":
os.chmod(temp, 0o600)
final = _place_backup(temp, db_path, label)
return os.path.basename(final)
except OSError as exc:
_remove_quietly(final)
raise SafetyBackupFailed("The backup could not be put in place") from exc
except SafetyBackupFailed:
_remove_quietly(final)
raise
finally:
_remove_quietly(temp, temp + "-journal", temp + "-wal", temp + "-shm")
return os.path.basename(_backups.take_safety_backup(model, label))
def _failure(message, status, code=None, **extra):
@@ -504,6 +359,19 @@ class MaintenanceRoutesMixin:
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"),
py/api/restore_plans.py
+146
View File
@@ -0,0 +1,146 @@
"""Reviewed restores: staged uploads held under single-use tokens (#249 S8).
A restore preview stages the upload (``database/restore_stage.py``) and
keeps it here under a ``secrets.token_urlsafe(32)`` token with the live
library's fingerprint and the database identity it was reviewed against.
``take`` hands the stage to the restore job exactly once; a second apply of
the token names that job (``PlanUsed``). An expired, discarded or unknown
token raises ``PlanExpired``; its stage folder is removed. A restart forgets
every token (the startup sweep removes the folders).
At most ``max_stages`` stages are held. Bytes are bounded by free disk space
(the upload is refused when the disk has no room, and held stages use that
disk), not by a fixed budget. A stage being applied belongs to its job and is
never evicted or counted.
The store is only read and changed on the event loop (it has no lock). Its
entries own stage folders, whose removal is blocking: every method that
drops an entry returns it (or keeps it in ``take_doomed``), and the caller
removes the folder in the executor (``HeldRestore.staged.discard``).
"""
import secrets
import time
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional
from .maintenance_plans import PlanExpired, PlanUsed
STAGE_TTL_SECONDS = 900
MAX_STAGES = 2
_USED_KEPT = 32 # applied tokens remembered for plan-used answers
class TooManyStages(Exception):
"""Every stage slot is held by a reviewed restore."""
@dataclass
class HeldRestore:
staged: Any # database.restore_stage.StagedRestore
fingerprint: str
db_identity: Any
live: Dict[str, int]
expires_at: float
class RestorePlanStore:
def __init__(
self,
clock: Optional[Callable[[], float]] = None,
ttl: float = STAGE_TTL_SECONDS,
max_stages: int = MAX_STAGES,
):
self._ttl = ttl
self._max = max_stages
self._clock = clock or time.time
self._held: Dict[str, HeldRestore] = {}
self._used: Dict[str, str] = {}
self._doomed: List[HeldRestore] = []
@property
def ttl(self) -> float:
return self._ttl
def now(self) -> float:
return self._clock()
def __len__(self):
return len(self._held)
def full(self) -> bool:
return len(self._held) >= self._max
def issue(self, staged, fingerprint, db_identity, live) -> str:
"""Hold ``staged`` and return its token; TooManyStages when full."""
if self.full():
raise TooManyStages()
token = secrets.token_urlsafe(32)
self._held[token] = HeldRestore(
staged=staged,
fingerprint=fingerprint,
db_identity=db_identity,
live=dict(live),
expires_at=self._clock() + self._ttl,
)
return token
def peek(self, token) -> HeldRestore:
if token in self._used:
raise PlanUsed(self._used[token])
held = self._held.get(token)
if held is None:
raise PlanExpired()
if held.expires_at <= self._clock():
self._doomed.append(self._held.pop(token))
raise PlanExpired()
return held
def take(self, token, job_id) -> HeldRestore:
"""Hand the held stage to the job applying it (single use)."""
held = self.peek(token)
del self._held[token]
self._used[token] = job_id
while len(self._used) > _USED_KEPT:
self._used.pop(next(iter(self._used)))
return held
def give_back(self, token, held) -> bool:
"""Hold a taken stage again under its token (its job did not apply it).
False, holding nothing, when the review expired meanwhile or its slot
was reused (the store is full): the caller then discards the stage.
"""
if held.expires_at <= self._clock() or self.full():
return False
self._used.pop(token, None)
self._held[token] = held
return True
def entry(self, token) -> Optional[HeldRestore]:
"""The held entry, expired or not (for the response that issued it)."""
return self._held.get(token)
def pop(self, token) -> Optional[HeldRestore]:
"""Drop the token; its entry (to discard) or None when it was not held."""
return self._held.pop(token, None)
def pop_expired(self) -> List[HeldRestore]:
"""Drop every expired entry; them, with any ``peek`` found expired."""
now = self._clock()
expired = [t for t, held in self._held.items() if held.expires_at <= now]
return [self._held.pop(token) for token in expired] + self.take_doomed()
def pop_all(self) -> List[HeldRestore]:
"""Drop every entry (the library was replaced); them, to discard."""
held = list(self._held.values()) + self.take_doomed()
self._held.clear()
return held
def take_doomed(self) -> List[HeldRestore]:
doomed, self._doomed = self._doomed, []
return doomed
def next_expiry(self) -> Optional[float]:
"""When the soonest held entry expires (clock time), or None."""
return min((held.expires_at for held in self._held.values()), default=None)
py/api/review.py
+19 -1
View File
@@ -31,6 +31,8 @@ from aiohttp import web
from .annotations import _HEX64, _TOKEN, _UUID, _capture_error, _error
from .images import media_thumbnail_url
from .inspection_service import VIDEO_EXTENSIONS
from .job_routes import busy_response
from .jobs import JobBusy
from .prompts import bad_request
try:
@@ -239,6 +241,9 @@ def _side_label(titles: dict, side, empty: str) -> str:
return display_title(titles.get(stored, ()), text) or empty
REVIEW_DECISION_LEASE = "Needs review decision"
class ReviewRoutesMixin:
"""The Needs review queue over E1's schema and E2's recorded items."""
@@ -446,7 +451,20 @@ class ReviewRoutesMixin:
# -- decision --------------------------------------------------------------
async def decide_review_item(self, request: web.Request) -> web.Response:
"""POST one explicit decision on one pending item."""
"""POST one explicit decision on one pending item.
The decision runs under a registry lease (#249 S8): it is refused
(409 ``busy``) while a restore or another ``exclusive`` job runs, and
a restore waits for decisions already running before it switches the
database. Decisions never refuse one another or a scan.
"""
try:
async with self._jobs.lease(REVIEW_DECISION_LEASE):
return await self._decide_review_item(request)
except JobBusy as busy:
return busy_response(busy)
async def _decide_review_item(self, request: web.Request) -> web.Response:
data, error = await self._read_json_body(request, max_bytes=MAX_BODY_BYTES)
if error is not None:
return error