From 8d61eda691f628259029044cc648ec7410e48c55 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 24 Oct 2026 06:13:06 -0700 Subject: [PATCH] feat(maintenance): one kept pre-migration backup and nothing-to-do refusal Re-review fixes R2, R3, R4, R5 and R6 (job side): - R3: Apply refuses a plan with nothing to rewrite (409 nothing-to-do, no job, no backup). Only the first apply keeps a pre-workflow-storage backup; later applies (after a Cancel, a rerun) take a pooled safety backup, forced on whatever the policy, so one copy is never pruned. - R4: a second-connection writer test runs against the live job loop; with its pause removed the writer hits "database is locked" (4 of 5 links). - R5: the digest tool loads only database/workflow_documents.py by path (no package start-up, no log handler) and opens a stopped database with immutable=1, so it leaves no -wal/-shm; a live one is read through its WAL. - R6: a backup that fails on a damaged document says which one in the job line. - R2: README's changelog says images recorded by this version use the new form from the first start, and the way back is the earliest pre-workflow-storage backup (or a pre-update backup if the job never ran). --- README.md | 4 +- TESTING.md | 20 +- py/api/maintenance.py | 41 +++- py/api/workflow_storage.py | 66 ++++-- tests/test_workflow_storage_rereview.py | 267 ++++++++++++++++++++++++ tools/workflow_text_digests.py | 43 +++- 6 files changed, 391 insertions(+), 50 deletions(-) create mode 100644 tests/test_workflow_storage_rereview.py diff --git a/README.md b/README.md index a61043c5..b60585ed 100644 --- a/README.md +++ b/README.md @@ -972,8 +972,8 @@ MIT License - see LICENSE file for details. ### Unreleased (Workflows stored once) -- **🗜️ Smaller database**: each distinct workflow and prompt graph is stored once, compressed, instead of again for every image. New images use it at once; the maintenance job "Store each workflow once" rewrites existing records and then reclaims the space. A 20,000-image test library went from 1.4 GB to 60 MB. The job always takes a verified backup first, `prompts.db.pre-workflow-storage_` beside the database, which is kept until you delete it. No image or video file is opened or changed. -- **⚠️ Downgrading after the rewrite**: an older Prompt Manager version still opens and even restores a rewritten database, but shows its images without workflows or prompt graphs. To go back to an older version, stop ComfyUI, delete `prompts.db-wal` and `prompts.db-shm`, and copy `prompts.db.pre-workflow-storage_` over `prompts.db` (records added since that backup are lost), or stay on this version. +- **🗜️ Smaller database**: each distinct workflow and prompt graph is stored once, compressed, instead of again for every image. Images recorded by this version use it from the first start; the maintenance job "Store each workflow once" rewrites the existing records and then reclaims the space. A 20,000-image test library went from 1.4 GB to 60 MB. The job always takes a verified backup first: the first time, `prompts.db.pre-workflow-storage_` beside the database, kept until you delete it; on later runs, a normal pooled safety backup. No image or video file is opened or changed. +- **⚠️ Downgrading**: images recorded by this version (and, after the job, all images) are stored in the new form. An older Prompt Manager version still opens and even restores such a database, but shows those images without workflows or prompt graphs until you update again. To go back to an older version, stop ComfyUI, delete `prompts.db-wal` and `prompts.db-shm`, and copy the **earliest** `prompts.db.pre-workflow-storage_` over `prompts.db`, or, if the job never ran, a backup taken before this update (records added since that backup are lost). Otherwise stay on this version. ### v3.3.1 (Proxy Routing Fix) diff --git a/TESTING.md b/TESTING.md index 766dc45e..213d3ebe 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2743,16 +2743,24 @@ reports `backup.policy: "required"`; `test_the_backup_is_taken_even_with_the_pol range search and read one row at a time, followed by 0.15 s with the write lock free so a writer on another connection gets in (`test_a_writer_on_another_connection_gets_in_between_batches`). Each text is read back through the accessor before its row changes. Each row has its own -savepoint: a row that cannot be rewritten (undecodable text, a text that would not come -back) stays inline, is listed in `errors` and named in the job's last line, and the rest -go on, Reclaim space included (`TestRowErrors`); any other failure rolls the batch back -whole. A reused document is decompressed and checked first and repaired from the text +savepoint: a row that cannot be rewritten (text SQLite cannot decode, a text that would +not come back) stays inline, is listed in `errors` and named in the job's last line, and +the rest go on, Reclaim space included (`TestRowErrors`); any other failure (a full disk, +an I/O error, a malformed database, a bug) rolls the batch back whole and fails the job +(`TestSystemErrorsFailTheBatch`, `TestSystemErrorsInTheJob`). With nothing to rewrite the +plan says "Nothing to rewrite" and Apply answers 409 `nothing-to-do` with no backup; only +the first apply keeps a `pre-workflow-storage` backup, and a later one (after a Cancel, a +rerun) takes a pooled safety backup (`TestOneKeptBackup`). The job loop's pause is tested +with a second-connection writer (`TestLiveLoopLetsWritersIn`). A reused document is decompressed and checked first and repaired from the text when damaged, and `verify_database_file` (restore and backup verification) reads every document and refuses a damaged one (`TestDamagedDocuments`). The legacy table rebuilds refuse a file that already has the document id columns (`TestRebuildGuards`). `tools/workflow_text_digests.py ` writes the sha256 of every record's stored -texts (read-only), so a `diff` of its output before and after the rewrite proves byte -identity. A crash or Cancel leaves +texts, so a `diff` of its output before and after the rewrite proves byte identity. It +loads only `database/workflow_documents.py` (no package start-up, no log handler) and +opens the database `mode=ro`, with `immutable=1` when no `-wal`/`-shm` exists, so it +creates none (`TestDigestToolSideEffects`). A damaged document's verify reason names it +and says what to do, and a backup that fails on it says so in the job line. A crash or Cancel leaves committed batches new form and the rest old form, and the next apply continues (`test_a_crash_mid_migration_leaves_every_row_readable_and_resumes`, `test_cancel_stops_after_the_current_batch_and_a_rerun_continues`). Values that are not diff --git a/py/api/maintenance.py b/py/api/maintenance.py index 0b3b1e1a..dddc1144 100644 --- a/py/api/maintenance.py +++ b/py/api/maintenance.py @@ -58,9 +58,11 @@ from .jobs import MAINTENANCE_JOB_KIND, JobBusy from .maintenance_plans import PlanExpired, PlanUsed, ScopeTooLarge try: + 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.workflow_storage import PRE_MIGRATION_LABEL, pre_migration_backups from database.health import readonly_uri from database.maintenance import MAINTENANCE_OPERATIONS, PlanChanged @@ -69,8 +71,6 @@ PREVIEW_BODY_MAX = 4096 DUPLICATE_PREVIEW_BODY_MAX = 1024 * 1024 DUPLICATE_FILES = "duplicate_files" WORKFLOW_STORAGE = "workflow_storage" # applied by py/api/workflow_storage.py -# Its backup's name, outside the pruned ``.backup_*`` pool. -PRE_MIGRATION_LABEL = "pre-workflow-storage" PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES,) # Room left over beyond what a step writes, so a disk is never filled to 0. DISK_MARGIN_BYTES = 16 * 2**20 @@ -93,12 +93,28 @@ PLAN_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" _NAME_ATTEMPTS = 1000 class SafetyBackupFailed(Exception): - """The safety backup could not be written, verified or put in place.""" + """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): + """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 _now(): @@ -409,9 +425,9 @@ def take_safety_backup(model, label="backup"): if not model.backup_database(temp): raise SafetyBackupFailed("The backup could not be written") _standalone(temp) - ok, _reason = model.verify_database_file(temp) + ok, reason = model.verify_database_file(temp) if not ok: - raise SafetyBackupFailed("The backup did not verify") + raise SafetyBackupFailed("The backup did not verify", reason) _fsync(temp) if os.name != "nt": os.chmod(temp, 0o600) @@ -530,6 +546,7 @@ class MaintenanceRoutesMixin: "targets": draft.targets, "roots": roots, "protected": draft.protected["count"], + "rows": draft.scope["count"], }, ) @@ -551,8 +568,11 @@ class MaintenanceRoutesMixin: policy = self._safety_backup_policy() destination = "backup" if operation == WORKFLOW_STORAGE: - # Always backed up first, whatever the policy (the way back). - policy, destination = "required", PRE_MIGRATION_LABEL + # 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, @@ -611,6 +631,9 @@ class MaintenanceRoutesMixin: 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( @@ -698,9 +721,9 @@ class MaintenanceRoutesMixin: job.log_line("Taking a safety backup of the database") try: backup = await self._run_in_executor(take_safety_backup, db.model) - except SafetyBackupFailed: + except SafetyBackupFailed as error: self.logger.error("Safety backup failed", exc_info=True) - _fail_job(job, "backup-failed", BACKUP_FAILED) + _fail_job(job, "backup-failed", backup_failed_message(error)) return False, None job.log_line(f"Safety backup {backup} verified") return True, backup diff --git a/py/api/workflow_storage.py b/py/api/workflow_storage.py index 472b1e1b..753c197b 100644 --- a/py/api/workflow_storage.py +++ b/py/api/workflow_storage.py @@ -32,10 +32,10 @@ import sqlite3 from .jobs import MAINTENANCE_JOB_KIND, JobBusy from .maintenance import ( - BACKUP_FAILED, MAINTENANCE_GROUPS, PRE_MIGRATION_LABEL, SafetyBackupFailed, + backup_failed_message, _fail_job, _megabytes, disk_plan, @@ -61,8 +61,7 @@ RECLAIM_FAILED = "Reclaim space failed; the database is unchanged." ROW_KEPT = "could not be rewritten; it keeps its inline text and reads as before" -def _backup_part(backup): - return f"Pre-migration backup {backup}" +NOTHING_LEFT = "Every image record already stores its workflow once" def _plural(count, one, many=None): @@ -97,34 +96,57 @@ class WorkflowStorageRoutesMixin: ) async def _run_workflow_storage(self, job, db, _backup_on): - # Always backed up first, whatever the policy: the way back. - backup = await self._pre_migration_backup(job, db) + # Always backed up first, whatever the policy: the way back. The first + # apply keeps a pre-migration backup; later ones (after a Cancel, a + # rerun) take a pooled safety backup, so only one copy is never pruned. + counts = await self._run_in_executor(db.workflow_storage_status) + if not counts["rows"]: + job.log_line(NOTHING_LEFT) + job.update(progress=100, status_line=NOTHING_LEFT) + return + kept = await self._run_in_executor( + workflow_storage.pre_migration_backups, db.model.db_path + ) + label = "backup" if kept else PRE_MIGRATION_LABEL + backup = await self._backup_before_rewrite(job, db, label, kept) if backup is None: return - if await self._rewrite_workflows(job, db, backup): + try: + done = await self._rewrite_workflows(job, db, backup) + finally: + if kept: + await self._prune_backup_pool(job, db) + if done: job.finish("finished") self._start_reclaim_space(job, db) - async def _pre_migration_backup(self, job, db): - """The verified ``.pre-workflow-storage_`` copy, or None. + async def _backup_before_rewrite(self, job, db, label, kept): + """The verified backup's status note, or None (the job failed). - Outside the pruned ``.backup_*`` pool: it stays until the user - removes it. A failure ends the job ``backup-failed``, nothing changed. + ``label`` ``pre-workflow-storage``: kept until the user removes it. + ``backup``: a pooled safety backup, while the earliest kept + pre-migration backup (``kept[0]``) stays. A failure ends the job + ``backup-failed``, nothing changed. """ - job.update(status_line="Taking the pre-migration backup") - job.log_line("Taking the pre-migration backup of the database") + noun = "safety backup" if kept else "pre-migration backup" + job.update(status_line=f"Taking the {noun}") + job.log_line(f"Taking the {noun} of the database") try: - backup = await self._run_in_executor( - take_safety_backup, db.model, PRE_MIGRATION_LABEL - ) - except SafetyBackupFailed: - self.logger.error("Pre-migration backup failed", exc_info=True) - _fail_job(job, "backup-failed", BACKUP_FAILED) + name = await self._run_in_executor(take_safety_backup, db.model, label) + except SafetyBackupFailed as error: + self.logger.error("Backup before the rewrite failed", exc_info=True) + _fail_job(job, "backup-failed", backup_failed_message(error)) return None + if kept: + job.log_line( + f"Safety backup {name} verified; the pre-migration backup " + f"{kept[0]} is kept" + ) + return f"Safety backup {name}" job.log_line( - f"Pre-migration backup {backup} verified; it is kept until you remove it" + f"Pre-migration backup {name} verified; it is kept until you remove it" ) - return backup + return f"Pre-migration backup {name}" async def _rewrite_workflows(self, job, db, backup): """Rewrite every old-form row; True when every row was looked at.""" @@ -185,7 +207,7 @@ class WorkflowStorageRoutesMixin: job.log_line(line) if removed: job.log_line(f"Removed {_plural(removed, 'unused workflow document')}") - job.update(progress=100, status_line=f"{line} · {_backup_part(backup)}") + job.update(progress=100, status_line=f"{line} · {backup}") return True def _stopped(self, job, rows, backup): @@ -194,7 +216,7 @@ class WorkflowStorageRoutesMixin: "run it again to continue" ) job.log_line(line) - job.update(status_line=f"{line} · {_backup_part(backup)}") + job.update(status_line=f"{line} · {backup}") job.finish("cancelled") return False diff --git a/tests/test_workflow_storage_rereview.py b/tests/test_workflow_storage_rereview.py new file mode 100644 index 00000000..be9daf39 --- /dev/null +++ b/tests/test_workflow_storage_rereview.py @@ -0,0 +1,267 @@ +"""PR #315 scoped re-review fixes: R1, R3, R4, R5 and R6 (R2 is wording). + +- R1: only a row's own cause (undecodable text, a text that would not come + back) keeps that row inline; disk-full, I/O and corruption fail the batch. +- R3: nothing to rewrite is refused up front; one kept pre-migration backup + ever, later applies take a pooled safety backup instead. +- R4: the live job loop lets a writer on another connection in. +- R5: the digest tool loads no package, adds no log handler and leaves no + ``-wal``/``-shm`` beside a stopped database. +- R6: a damaged document is named, with what to do, in the verify reason and + in the job's backup failure. +""" + +import asyncio +import os +import sqlite3 +import subprocess # nosec B404 +import sys +import threading +from unittest.mock import patch + +from database import workflow_documents as wd +from database import workflow_storage as ws +from database.models import PromptModel + +try: + from tests.test_api_workflow_storage import OPERATION, WorkflowStorageTestCase + from tests.test_workflow_migration import MigrationTestCase +except ImportError: # discovered with tests/ as the top-level directory + from test_api_workflow_storage import OPERATION, WorkflowStorageTestCase + from test_workflow_migration import MigrationTestCase + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +TOOL = os.path.join(ROOT, "tools", "workflow_text_digests.py") +KEPT = "prompts.db.pre-workflow-storage_" +WAIT = 20 + + +def disk_full(*_args, **_kwargs): + raise sqlite3.OperationalError("database or disk is full") + + +class TestSystemErrorsFailTheBatch(MigrationTestCase): + def test_a_disk_error_fails_the_batch_instead_of_listing_rows(self): + _prompt_id, expected = self.library(count=6) + for error in ( + sqlite3.OperationalError("database or disk is full"), + sqlite3.OperationalError("disk I/O error"), + sqlite3.DatabaseError("database disk image is malformed"), + TypeError("a bug"), + ): + with self.subTest(error=str(error)): + with patch.object(ws, "store_text", side_effect=error): + with self.assertRaises(type(error)): + self.migrate(max_rows=50) + self.assertFalse(self.conn().in_transaction) + self.assertEqual(self.inline_ids(), set(expected)) + self.assertEqual(self.texts(), expected) + + +class TestSystemErrorsInTheJob(WorkflowStorageTestCase): + async def test_a_disk_error_ends_the_job_failed_with_a_truthful_line(self): + self.old_library(count=4) + before = self.texts() + with patch.object(ws, "store_text", side_effect=disk_full): + _plan, _summary, job = await self.start() + await self.finish(job) + self.assertEqual(job.state, "failed") + self.assertIn("The batch in progress was rolled back", job.status_line) + self.assertEqual(job.counters["errors"], 1) # the job's one error row + self.assertNotIn("could not be rewritten", job.status_line) + self.assertIsNone(self.reclaim_job()) + self.assertEqual(self.texts(), before) + + +class TestOneKeptBackup(WorkflowStorageTestCase): + def kept(self): + return sorted(n for n in os.listdir(self.tmpdir) if n.startswith(KEPT)) + + async def test_nothing_to_rewrite_is_said_and_refused_without_a_backup(self): + self.save("no images") + plan = await self.preview(OPERATION) + self.assertEqual(plan["scope"]["count"], 0) + self.assertEqual(plan["scope"]["label"], "Nothing to rewrite") + reply = await self.apply(plan["token"], status=409) + self.assertEqual(reply["code"], "nothing-to-do") + self.assertEqual( + reply["error"], + "Every image record already stores its workflow once; nothing to do.", + ) + self.assertEqual(self.kept(), []) + self.assertEqual(self.backups(), []) + self.assertEqual(self.api._jobs.list(), []) + + async def test_a_cancel_and_rerun_keep_one_kept_backup_and_pool_the_next(self): + self.old_library(count=12) + real = self.api.db.migrate_workflow_batch + holder = {} + + def first_then_cancel(*args, **kwargs): + result = real(*args, **kwargs) + holder["job"].cancelled = True + return result + + with ( + patch.object(ws, "BATCH_ROWS", 5), + patch.object( + self.api.db, "migrate_workflow_batch", side_effect=first_then_cancel + ), + ): + plan = await self.preview(OPERATION) + body = await self.apply(plan["token"]) + holder["job"] = job = self.api._jobs.get(body["job"]["id"]) + await self.finish(job) + self.assertEqual(job.state, "cancelled") + [kept] = self.kept() + plan = await self.preview(OPERATION) + self.assertEqual(plan["backup"]["policy"], "required") + self.assertEqual( + plan["backup"]["destination_label"], "prompts.db.backup_" + ) + self.assertIn( + f"the pre-migration backup {kept} is already kept", + " ".join(row["detail"] for row in plan["details"]), + ) + body = await self.apply(plan["token"]) + again = self.api._jobs.get(body["job"]["id"]) + await self.finish(again) + self.assertEqual(again.state, "finished", again.log) + self.assertEqual(self.kept(), [kept]) # still one never-pruned copy + [pooled] = self.backups() + self.assertIn(f"Safety backup {pooled} verified", "\n".join(again.log)) + self.assertIn(f"pre-migration backup {kept} is kept", "\n".join(again.log)) + + +class TestLiveLoopLetsWritersIn(WorkflowStorageTestCase): + async def test_a_writer_on_another_connection_gets_in_during_the_job(self): + prompt_id = self.save("writer") + conn = self.api.db.model.get_connection() + with conn: + for index in range(200): + conn.execute( + "INSERT INTO generated_images (prompt_id, image_path, filename, " + "file_path, workflow_data) VALUES (?, ?, 'x', ?, ?)", + (prompt_id, f"/x/{index}", f"/x/{index}", f'{{"n": {index}}}'), + ) + started, done = threading.Event(), threading.Event() + errors, written = [], [] + real = ws.store_text + + def slow(conn, text, *args, **kwargs): + started.set() + done.wait(0.01) # a slow disk: each row takes a while + return real(conn, text, *args, **kwargs) + + def writer(): + started.wait(WAIT) + own = self.api.db.model.get_connection() # this thread's connection + own.execute("PRAGMA busy_timeout = 2000") + try: + for index in range(5): + written.append( + self.api.db.link_image_to_prompt( + prompt_id, + os.path.join(self.tmpdir, f"w{index}.png"), + {"workflow": {"w": index}}, + ) + ) + except sqlite3.OperationalError as error: # pragma: no cover - RED + errors.append(str(error)) + finally: + self.api.db.model.close() + + thread = threading.Thread(target=writer) + thread.start() + try: + with patch.object(ws, "store_text", side_effect=slow): + _plan, _summary, job = await self.start() + await asyncio.wait_for(asyncio.shield(job.task), 120) + await self.finish(job) # and Reclaim space after it + finally: + done.set() + await asyncio.get_running_loop().run_in_executor(None, thread.join, WAIT) + self.assertEqual(job.state, "finished", job.log) + self.assertEqual(errors, []) + self.assertEqual(len([image for image in written if image]), 5) + + +class TestDigestToolSideEffects(MigrationTestCase): + def run_tool(self, path, cwd): + out = os.path.join(cwd, "out.sha") + code = ( + "import logging, runpy, sys\n" + "sys.argv = ['tool', sys.argv[1], sys.argv[2]]\n" + "try:\n" + f" runpy.run_path({TOOL!r}, run_name='__main__')\n" + "except SystemExit:\n" + " pass\n" + "loaded = sorted(m for m in sys.modules\n" + " if m.split('.')[0] in ('database', 'utils', 'py'))\n" + "handlers = [h for n in ('', 'prompt_manager')\n" + " for h in logging.getLogger(n).handlers]\n" + "print('LOADED', loaded, 'HANDLERS', len(handlers))\n" + ) + result = subprocess.run( # nosec B603 + [sys.executable, "-c", code, path, out], + capture_output=True, + text=True, + cwd=cwd, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + return result.stdout + + def test_a_stopped_database_gets_no_sidecars_and_no_log(self): + self.library(count=3) + self.migrate() + PromptModel.close_all_instances() + for suffix in ("-wal", "-shm"): + self.assertFalse(os.path.exists(self.db_path + suffix)) + cwd = os.path.join(self.tmpdir, "run") + os.mkdir(cwd) + before = sorted(os.listdir(self.tmpdir)) + printed = self.run_tool(self.db_path, cwd) + self.assertEqual(sorted(os.listdir(self.tmpdir)), before) + self.assertEqual(sorted(os.listdir(cwd)), ["out.sha"]) # no logs/ folder + self.assertIn("LOADED [] HANDLERS 0", printed) + with open(os.path.join(cwd, "out.sha"), encoding="utf-8") as handle: + self.assertEqual(len(handle.read().splitlines()), 3) + + def test_a_live_database_is_read_with_its_wal(self): + prompt_id = self.prompt() + self.old_form_image(prompt_id, "a.png", '{"in": "wal"}', None) + # The connection stays open: the row may still be only in the WAL. + cwd = os.path.join(self.tmpdir, "run") + os.mkdir(cwd) + self.run_tool(self.db_path, cwd) + with open(os.path.join(cwd, "out.sha"), encoding="utf-8") as handle: + self.assertEqual(len(handle.read().splitlines()), 1) + + +class TestDamagedDocumentIsNamed(WorkflowStorageTestCase): + def damage(self): + conn = self.api.db.model.get_connection() + with conn: + document_id = wd.store_text(conn, '{"damaged": true}') + conn.execute( + "UPDATE workflow_documents SET data=x'00' WHERE id=?", (document_id,) + ) + return document_id + + async def test_verify_names_the_document_and_what_to_do(self): + document_id = self.damage() + ok, reason = PromptModel.verify_database_file(self.api.db.model.db_path) + self.assertFalse(ok) + self.assertEqual(reason, wd.damaged_message(document_id)) + self.assertIn(f"Workflow document {document_id} is damaged", reason) + self.assertIn("restore a backup", reason) + + async def test_the_backup_failure_says_which_document(self): + self.old_library(count=2) + document_id = self.damage() + _plan, _summary, job = await self.start() + await self.finish(job) + self.assertEqual(job.state, "failed") + self.assertEqual(job.last_event["code"], "backup-failed") + self.assertIn(f"Workflow document {document_id} is damaged", job.status_line) diff --git a/tools/workflow_text_digests.py b/tools/workflow_text_digests.py index 891ac8ab..366fb244 100644 --- a/tools/workflow_text_digests.py +++ b/tools/workflow_text_digests.py @@ -11,23 +11,44 @@ Run it from this repository on the database before the rewrite (or on its Each line is `` `` (``-`` for none), hashing the exact stored bytes: the inline column before the -rewrite, the document read through the one accessor after it. Read-only: the -database is opened with ``mode=ro`` and nothing is written to it. +rewrite, the document read through the one accessor after it. + +Read-only, with no side effects: the database is opened with ``mode=ro``, +and with ``immutable=1`` when no ``-wal``/``-shm`` file exists beside it (a +stopped database or a backup), so SQLite creates none. A live database +already has them and is read through its WAL. Only +``database/workflow_documents.py`` is loaded (by path; it needs only the +standard library), so no package start-up runs and no log file or handler is +added. Run it as the user that runs ComfyUI. """ import argparse import hashlib +import importlib.util import os import sqlite3 import sys from pathlib import Path -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +_ROOT = Path(__file__).resolve().parents[1] -from database.workflow_documents import ( # noqa: E402 - DocumentReader, - resolve_image_row, -) + +def _load_documents(): + """``database/workflow_documents.py`` alone, without the ``database`` package.""" + path = _ROOT / "database" / "workflow_documents.py" + spec = importlib.util.spec_from_file_location("_pm_workflow_documents", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def database_uri(db_path): + """``mode=ro``, plus ``immutable=1`` when nothing has the database open.""" + path = os.path.abspath(db_path) + uri = Path(path).as_uri() + "?mode=ro" + if not any(os.path.exists(path + suffix) for suffix in ("-wal", "-shm")): + uri += "&immutable=1" + return uri def _digest(value): @@ -39,16 +60,16 @@ def _digest(value): def write_digests(db_path, out): """Write one digest line per image record of ``db_path`` to ``out``.""" - uri = Path(os.path.abspath(db_path)).as_uri() + "?mode=ro" - conn = sqlite3.connect(uri, uri=True) + documents = _load_documents() + conn = sqlite3.connect(database_uri(db_path), uri=True) try: conn.row_factory = sqlite3.Row conn.text_factory = bytes # inline text exactly as stored, even undecodable - reader = DocumentReader(conn) + reader = documents.DocumentReader(conn) count = 0 with open(out, "w", encoding="utf-8") as handle: for row in conn.execute("SELECT * FROM generated_images ORDER BY id"): - texts = resolve_image_row(reader, row) + texts = documents.resolve_image_row(reader, row) handle.write( f"{row['id']} {_digest(texts.get('workflow_data'))} " f"{_digest(texts.get('prompt_metadata'))}\n"