wip(system): library move fix round in progress

This commit is contained in:
Vito Sansevero
2026-10-24 09:28:58 -07:00
parent 573396cac3
commit 55b49fb336
15 changed files with 822 additions and 103 deletions
+3
View File
@@ -211,3 +211,6 @@ coverage/
# Disposable upstream checkout for real-host CI
.ci-comfyui/
# Old library files a Move library renamed (never deleted; #249 S6b)
*.moved-to-*
+39 -1
View File
@@ -2851,7 +2851,9 @@ workflow (README changelog).
database), the logs (`utils/logging_config.py`) and the LoRA image cache
(`py/lora_utils.py`). Order: a valid `PROMPT_MANAGER_DATA_DIR` (absolute, never inside
ComfyUI's output, input or temp folder, a public user folder, an extra gallery root or
a gallery source; an invalid one is ignored and reported as a problem code), used when
an image source listed in any config.json the resolver may read; `config_location` and
`resolve_library` read those sources the same way, so config.json and the database
never split; an invalid one is ignored and reported as a problem code), used when
it holds a library or no node-folder library exists; an explicit
`database.default_path` other than `prompts.db` (relative to the node folder, as
before) and `PROMPT_MANAGER_CONFIG_PATH`, honoured as before; ComfyUI's
@@ -2895,6 +2897,42 @@ path (`tests/test_api_library_location.py`). The real-host `library-location.spe
install creates `user/__promptmanager/prompts.db` and its logs, adds nothing but byte
code to the node folder, and that `/userdata` serves none of it.
The switch runs while the copy is write-locked too (a guard transaction on it), so a
switch that fails at any point (the resolver, part of the models, the observers) is
undone whole: every model points at the old file again, the copy and its marker are
removed, and the move fails "… Nothing was moved." (`TestFailedSwitchIsUndone`). What
follows a complete switch (dropping plans, the health report, the history read fence,
which is closed first) is best effort: logged, never a failed move. A thread that
opens a connection while the model is being repointed keeps it only if the model's
generation did not change meanwhile (`PromptModel.get_connection`).
Writes racing the switch: a background write that was waiting on the old file's lock
when the switch closed its connection fails, and run counting (`count_runs`,
`count_pair_runs`) and the image monitor's link (`ImageGenerationHandler.
link_image_to_prompt`) run it once more through `PromptDatabase.survive_library_switch`
when the model's generation changed during the attempt; the retry opens the new
library (`TestBackgroundWritesSurviveTheSwitch`, including a real UPDATE blocked on the
lock that lands in the new file). Any other failure is not retried. Known limit: that
blocked writer, and the mover with it, waits out the writer's busy timeout (30 s)
before its connection closes: SQLite holds the connection's mutex during the busy
wait, so neither a close nor `Connection.interrupt()` ends it sooner (measured), and
another thread cannot shorten the timeout of a connection in use. Other writes are
unaffected: threads that ask for a connection after the switch get the new file.
Writes from the HTTP routes (a prompt saved in the dashboard at that moment) are not
retried and report their error.
A restart after a crash between the switch and the rename finishes the rename: the
marker names the old `prompts.db` and `config.json` (`from`, `config_from`), and the API
renames them `*.moved-to-<folder>` at startup when they are still there under their
own names, never deleting them and leaving a file still in use alone
(`TestRestartFinishesTheRename`). A second ComfyUI process using the same node folder
(two installs sharing one custom_nodes folder) is not switched: it keeps writing to the
old file, renamed or not, and those writes are not in the moved library; a rename while
it has the file open is refused as `old-in-use`. The Overview flags a library that stays
in the node folder through an explicit `database.default_path` or a node-folder
config.json (`inside-node-folder`: a reinstall deletes it, CivitAI key included); it is
not offered a move.
Windows: every connection to the old file is closed before its rename (the source
snapshot connection, every model's connections, the checkpoint connection); temp files
are closed before `os.replace`; renames stay inside one folder; folder identity compares
+129 -17
View File
@@ -45,7 +45,9 @@ import os
import shutil
import sqlite3
import time
import types
import uuid
from pathlib import Path
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Tuple
@@ -90,8 +92,8 @@ MESSAGES = {
"Nothing was moved; try again when no workflow is running."
),
"switch-failed": (
"The library was copied and verified, but this session could not "
"switch to it. Restart ComfyUI to use it."
"Prompt Manager could not switch to the verified copy, so the copy was "
"removed. Nothing was moved."
),
}
@@ -314,7 +316,7 @@ def _verify_copy(temp: str, counts, documents) -> None:
"verify-failed",
f"{MESSAGES['verify-failed']} {detail}" if detail else None,
)
uri = "file:" + temp.replace(os.sep, "/") + "?mode=ro"
uri = Path(os.path.abspath(temp)).as_uri() + "?mode=ro"
try:
with contextlib.closing(sqlite3.connect(uri, uri=True)) as conn:
copied = _describe(conn)
@@ -367,8 +369,14 @@ def _check_copies(copies) -> None:
raise MoveFailed("copy-failed")
def write_marker(directory: str, state: str, source_db: str) -> str:
"""Atomically write the move marker; return its path."""
def write_marker(
directory: str, state: str, source_db: str, source_config: Optional[str] = None
) -> str:
"""Atomically write the move marker; return its path.
``from`` and ``config_from`` name the old files, so a restart after a
crash between the switch and the rename can finish the rename.
"""
path = os.path.join(directory, ll.MARKER_NAME)
temp = os.path.join(directory, f"{TEMP_PREFIX}{uuid.uuid4().hex}.marker")
payload = json.dumps(
@@ -376,6 +384,7 @@ def write_marker(directory: str, state: str, source_db: str) -> str:
"version": 1,
"state": state,
"from": source_db,
"config_from": source_config,
"updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
},
indent=2,
@@ -425,15 +434,32 @@ def _noop(*_args, **_kwargs):
return None
def switch_process(old_db: str, location) -> None:
"""Point this process at ``location``: resolver, every model, observers."""
def switch_process(old_db: str, location) -> list:
"""Point this process at ``location``: resolver, every model, observers.
Returns the old file's connections, detached and still open: the mover
closes them once the switch is final (``PromptModel.close_detached``).
"""
from .execution_history import rekey_history_restore
from .review_changes import rekey_review_changes
ll.switch_to(location)
PromptModel.repoint_instances(old_db, location.db_path)
detached = PromptModel.repoint_instances(old_db, location.db_path)
rekey_review_changes(old_db, location.db_path)
rekey_history_restore(old_db, location.db_path)
return detached
def unswitch_process(old_db: str, new_db: str) -> list:
"""Undo ``switch_process`` (any part of it); returns detached connections."""
from .execution_history import rekey_history_restore
from .review_changes import rekey_review_changes
ll.clear_override()
detached = PromptModel.repoint_instances(new_db, old_db)
rekey_review_changes(new_db, old_db)
rekey_history_restore(new_db, old_db)
return detached
class _Move:
@@ -508,7 +534,9 @@ class _Move:
raise MoveFailed("copy-failed") from error
_check_copies([(self.plan.config_path, temp, expected)])
copies = copies + [(self.plan.config_path, temp, expected, ll.CONFIG_NAME)]
self.marker = write_marker(target, ll.MARKER_PENDING, self.plan.db_path)
self.marker = write_marker(
target, ll.MARKER_PENDING, self.plan.db_path, self.plan.config_path
)
self.checkpoint("pending")
for _src, temp, _expected, name in copies:
final = os.path.join(target, name)
@@ -521,7 +549,7 @@ class _Move:
files.append(ll.DATABASE_NAME)
_fsync_dir(target)
self.checkpoint("published")
write_marker(target, ll.MARKER_MOVED, self.plan.db_path)
write_marker(target, ll.MARKER_MOVED, self.plan.db_path, self.plan.config_path)
self.moved = True
self.checkpoint("marked")
return files
@@ -581,16 +609,63 @@ def retire_old_files(plan: MovePlan) -> Tuple[List[str], Optional[str]]:
return renamed, None
def finish_interrupted_move(location) -> List[str]:
"""Rename old files a crash left between the switch and the rename.
The marker of the folder in use says ``moved`` and names the old files;
an old ``prompts.db`` (and config.json) still there under its own name is
renamed ``*.moved-to-<folder>`` now, never deleted. Returns the new
names (nothing when there is nothing to finish or the file is in use).
"""
if location.source not in (ll.SOURCE_DEFAULT, ll.SOURCE_ENVIRONMENT):
return []
marker = ll.read_marker(location.data_dir)
old = marker.get("from")
if marker.get("state") != ll.MARKER_MOVED or not isinstance(old, str):
return []
if not os.path.isfile(old) or os.path.basename(old) != ll.DATABASE_NAME:
return []
if ll.same_folder(os.path.dirname(old), location.data_dir):
return []
config = marker.get("config_from")
if not (
isinstance(config, str)
and os.path.basename(config) == ll.CONFIG_NAME
and os.path.isfile(config)
and os.path.normcase(os.path.abspath(config))
!= os.path.normcase(os.path.abspath(location.config_path))
):
config = None
pending = types.SimpleNamespace(
db_path=old, config_path=config, target_dir=location.data_dir
)
renamed, problem = retire_old_files(pending)
if problem:
logger.warning(f"Library move: the old files are still in use ({problem})")
elif renamed:
logger.info(f"Library move finished: renamed {', '.join(renamed)}")
return renamed
def move_library(
plan: MovePlan,
*,
switch: Callable = switch_process,
unswitch: Callable = unswitch_process,
after_switch: Callable = _noop,
config_lock=None,
progress: Callable[[str], None] = _noop,
checkpoint: Callable[[str], None] = _noop,
attempts: int = DEFAULT_ATTEMPTS,
) -> MoveResult:
"""Move the library as planned (blocking; see the module docstring)."""
"""Move the library as planned (blocking; see the module docstring).
``switch(old_db, location)`` points the process at the copy and returns
the old connections it detached; if it raises, ``unswitch(old_db,
new_db)`` undoes it while both files are still locked, the copy is
removed and the move fails (``switch-failed``). ``after_switch()`` runs
once the switch is final; it is best effort (logged, never fails the move).
"""
short = space_problem(plan)
if short is not None:
raise MoveFailed("insufficient-space")
@@ -602,6 +677,7 @@ def move_library(
progress=progress,
checkpoint=checkpoint,
)
move.unswitch = unswitch
location = plan.target_location
try:
move.prepare_target()
@@ -609,6 +685,10 @@ def move_library(
except Exception:
move.cleanup()
raise
try:
after_switch()
except Exception:
logger.exception("Library move: a follow-up after the switch failed")
from .execution_history import notify_history_restore
notify_history_restore(location.db_path)
@@ -618,6 +698,39 @@ def move_library(
return MoveResult(location, counts, files, renamed, problem)
def _switch_guarded(move: _Move, location) -> list:
"""Switch while the copy is locked too; undo it all if any part fails.
The copy's write lock keeps every repointed thread from writing to it
until the switch is whole, so undoing a partial switch loses nothing.
Returns the old file's detached connections (still open).
"""
plan = move.plan
guard = sqlite3.connect(
location.db_path, timeout=BUSY_TIMEOUT_SECONDS, isolation_level=None
)
try:
guard.execute("BEGIN IMMEDIATE")
try:
detached = move.switch(plan.db_path, location)
except Exception as error:
logger.exception("Library move: switching failed; undoing it")
move.moved = False # the copy and its marker are removed
try:
PromptModel.close_detached(
move.unswitch(plan.db_path, location.db_path), logger
)
except Exception:
logger.exception("Library move: undoing the switch failed")
raise MoveFailed("switch-failed") from error
move.checkpoint("switched")
return detached
finally:
if guard.in_transaction:
guard.execute("ROLLBACK")
guard.close()
def _copy_and_switch(move: _Move, location, attempts: int):
plan = move.plan
for attempt in range(1, attempts + 1):
@@ -636,12 +749,11 @@ def _copy_and_switch(move: _Move, location, attempts: int):
move.checkpoint("locked")
with move.config_lock:
files = move.publish(temp_db, copies)
try:
move.switch(plan.db_path, location)
except Exception as error:
logger.exception("Library move: switching failed")
raise MoveFailed("switch-failed") from error
move.checkpoint("switched")
detached = _switch_guarded(move, location)
# Still holding the old file's lock: a write that was waiting
# on it fails when its connection closes (and runs again on
# the new library) instead of landing in the old file.
PromptModel.close_detached(detached, logger)
finally:
if source.in_transaction:
source.execute("ROLLBACK")
+55 -28
View File
@@ -609,19 +609,37 @@ class PromptModel:
if conn is not None and self._local.generation == self._generation:
return conn
# check_same_thread=False only so close_all can clean up other
# threads' connections; each connection is still used by one thread.
conn = sqlite3.connect(
self.db_path, check_same_thread=False, timeout=BUSY_TIMEOUT_SECONDS
)
conn.row_factory = sqlite3.Row
configure_connection(conn)
while True:
# The path and generation are read together, and the new
# connection is kept only if no Move library repointed the model
# meanwhile: otherwise it would write to the old file.
with self._conn_lock:
path, generation = self.db_path, self._generation
# check_same_thread=False only so close_all can clean up other
# threads' connections; each connection is still used by one thread.
conn = sqlite3.connect(
path, check_same_thread=False, timeout=BUSY_TIMEOUT_SECONDS
)
try:
conn.row_factory = sqlite3.Row
configure_connection(conn)
except BaseException:
conn.close()
raise
with self._conn_lock:
if generation == self._generation:
self._connections.add(conn)
break
conn.close()
self._local.conn = conn
self._local.generation = self._generation
with self._conn_lock:
self._connections.add(conn)
self._local.generation = generation
return conn
@property
def generation(self) -> int:
"""Bumped whenever every connection is closed or the model repointed."""
return self._generation
@classmethod
def close_all_instances(cls) -> None:
"""Close every connection of every live model in this process."""
@@ -631,26 +649,36 @@ class PromptModel:
model.close_all()
@classmethod
def repoint_instances(cls, old_path: str, new_path: str) -> int:
def repoint_instances(cls, old_path: str, new_path: str) -> list:
"""Point every model of ``old_path`` at ``new_path`` (Move library).
Each matching model's connections are closed and every thread opens
the new file on its next ``get_connection``. Closing a connection
another thread is using waits for its statement; a write waiting on
the mover's lock then fails instead of landing in the old file.
Returns how many models were repointed.
Every thread opens the new file on its next ``get_connection``. The
old connections are detached, not closed, and returned: the caller
closes them once nothing can block on the new file any more. Closing
a connection another thread is using waits for its statement; a write
waiting on the mover's lock then fails (and, through
``PromptDatabase.survive_library_switch``, runs again on the new file).
"""
old_key = os.path.normcase(os.path.realpath(old_path))
with cls._instances_lock:
models = list(cls._instances)
with cls._init_lock:
cls._initialized_paths.add(os.path.normcase(os.path.realpath(new_path)))
moved = 0
detached = []
for model in models:
if model._schema_key() == old_key:
model._close_connections(new_path)
moved += 1
return moved
detached.extend(model._detach_connections(new_path))
return detached
@staticmethod
def close_detached(connections, logger=None) -> None:
"""Close connections ``repoint_instances`` detached."""
for conn in connections:
try:
conn.close()
except sqlite3.Error as e:
if logger is not None:
logger.warning(f"Error closing database connection: {e}")
def close_all(self) -> None:
"""Close every thread's connection; they reopen lazily on next use.
@@ -658,10 +686,13 @@ class PromptModel:
Needed before deleting the database file (Windows keeps open files
locked) and by tests that ran queries through executor threads.
"""
self._close_connections()
self.close_detached(self._detach_connections(), self.logger)
def _close_connections(self, new_path: Optional[str] = None) -> None:
"""Close every connection (optionally pointing the model elsewhere)."""
def _detach_connections(self, new_path: Optional[str] = None) -> list:
"""Forget every connection (optionally pointing the model elsewhere).
Returns them for the caller to close; threads reopen on next use.
"""
with self._conn_lock:
if new_path is not None:
self.db_path = new_path
@@ -669,11 +700,7 @@ class PromptModel:
self._connections.clear()
self._generation += 1
self._local.conn = None
for conn in connections:
try:
conn.close()
except sqlite3.Error as e:
self.logger.warning(f"Error closing database connection: {e}")
return connections
def close(self) -> None:
"""Close the calling thread's connection, if it has one.
+29 -21
View File
@@ -101,8 +101,8 @@ def escape_like(term: str) -> str:
)
def _configured_library():
"""``(database.default_path, gallery roots)`` from config, or ``(None, [])``.
def _configured_default_path():
"""config.json's ``database.default_path``, or None.
The package-qualified config wins over a standalone ``py.config`` (a
custom-node package uses its own configuration).
@@ -113,28 +113,17 @@ def _configured_library():
except ImportError:
from py.config import PromptManagerConfig
except Exception:
return None, []
roots = []
try:
try:
from ..py.config import GalleryConfig
except ImportError:
from py.config import GalleryConfig
directories = list(GalleryConfig.MONITORING_DIRECTORIES or [])
except Exception:
directories = []
for directory in directories:
try:
roots.append(GalleryConfig.resolve_gallery_root(directory))
except Exception:
continue
return getattr(PromptManagerConfig, "DEFAULT_DB_PATH", None), roots
return None
return getattr(PromptManagerConfig, "DEFAULT_DB_PATH", None)
def resolve_library_location():
"""Where the library lives now (``utils.library_location``, #249 S6b)."""
default_path, roots = _configured_library()
return library_location.resolve_library(default_path, gallery_roots=roots)
"""Where the library lives now (``utils.library_location``, #249 S6b).
The image sources that rule out a ``PROMPT_MANAGER_DATA_DIR`` are read by
the resolver itself, exactly as ``config_location`` reads them.
"""
return library_location.resolve_library(_configured_default_path())
def _resolve_db_path(db_path: Optional[str] = None) -> str:
@@ -394,6 +383,25 @@ class PromptDatabase(
self.model = PromptModel(db_path)
self.logger.debug("Database operations initialized successfully")
def survive_library_switch(self, write):
"""Run ``write()``; once more if it failed while the library moved.
A background write (run counting, the image monitor) that was waiting
on the old file when Move library switched fails when its connection
is closed. When the model's generation changed during the attempt,
the write runs again on a fresh connection, which is the new library;
any other failure is raised as it was. ``write`` must open its own
transaction (every PromptDatabase method does).
"""
generation = self.model.generation
try:
return write()
except Exception:
if self.model.generation == generation:
raise
self.logger.info("Running a write again on the moved library")
return write()
def close(self) -> None:
"""Close the calling thread's database connection (see PromptModel.close)."""
self.model.close()
+4
View File
@@ -379,6 +379,10 @@ class PromptManagerAPI(
except Exception as e:
self.logger.warning(f"Could not sweep interrupted backup copies: {e}")
# A crash between Move library's switch and its rename left the old
# node-folder files under their own names (#249 S6b): rename them now.
self.finish_interrupted_library_move()
# Run cleanup on initialization to remove any existing duplicates
try:
removed = self.db.cleanup_duplicates()
+51 -9
View File
@@ -139,6 +139,9 @@ class LibraryLocationRoutesMixin:
pass
return {
"path": self._public_path(location.db_path),
# Where Move library would put it, and whether config.json goes too.
"target_source": location.target_source if location.movable else None,
"config_moves": location.movable and not location.config_explicit,
"source": location.source,
"movable": location.movable,
"keep": keep,
@@ -146,10 +149,34 @@ class LibraryLocationRoutesMixin:
"target": (
self._public_path(location.target_dir) if location.movable else None
),
"problem": location.problem,
"problem": location.problem or self._node_folder_risk(location),
"size_bytes": size,
}
# The node folder (tests point it at a temp tree).
_extension_root = ll.EXTENSION_ROOT
def _node_folder_risk(self, location):
"""``inside-node-folder`` for a library that stays in the node folder.
A configured database path, or a config.json the resolver still reads
from the node folder, is deleted with the node by a reinstall, and
it is never offered a move.
"""
if location.source == ll.SOURCE_NODE_FOLDER:
return None
paths = [location.db_path] + (
[] if location.config_explicit else [location.config_path]
)
root = self._extension_root
for path in paths:
parent = os.path.dirname(os.path.abspath(path))
if ll.same_folder(parent, root) or os.path.normcase(
os.path.realpath(parent)
).startswith(os.path.normcase(os.path.realpath(root)) + os.sep):
return "inside-node-folder"
return None
def _library_payload(self):
return self._location_payload(self._live_location())
@@ -201,12 +228,26 @@ class LibraryLocationRoutesMixin:
return busy_response(busy)
return web.json_response({"success": True, "job": job.summary()}, status=202)
def _switch_library(self, old_db, location):
"""Point the process and this API at the moved library (blocking)."""
lm.switch_process(old_db, location)
# Plans, health reports and history reads described the old file.
def _after_library_switch(self):
"""Forget what described the old file (best effort, after the switch).
Plans, the health report and the history read fence named the old
file; the fence is closed before it is dropped.
"""
self._database_replaced()
self._history_read_fence = None
fence, self._history_read_fence = self._history_read_fence, None
if fence is not None:
fence.close()
def finish_interrupted_library_move(self):
"""Startup: rename old files a crash left after a move (blocking)."""
try:
location = self._library_resolver()
if _same_file(location.db_path, self.db.model.db_path):
return lm.finish_interrupted_move(location)
except Exception:
self.logger.exception("Could not finish an interrupted library move")
return []
async def _run_library_move(self, job):
from ..config_store import _COMMIT_LOCK
@@ -229,15 +270,16 @@ class LibraryLocationRoutesMixin:
functools.partial(
lm.move_library,
plan,
switch=self._switch_library,
after_switch=self._after_library_switch,
config_lock=_COMMIT_LOCK,
progress=progress,
)
)
except lm.MoveFailed as failed:
self.logger.error(f"Move library failed: {failed.code}")
tail = "" if failed.code == "switch-failed" else f" {STILL_THERE}"
return self._fail_library_move(job, failed.code, failed.message + tail)
return self._fail_library_move(
job, failed.code, f"{failed.message} {STILL_THERE}"
)
except Exception:
self.logger.exception("Move library failed")
return self._fail_library_move(job, None, f"{NOT_MOVED} {STILL_THERE}")
@@ -8,6 +8,7 @@ const locationModule = import('../../web/js/workspace/library-location.mjs');
const NODE_FOLDER = Object.freeze({
path: 'custom_nodes/ComfyUI_PromptManager/prompts.db', source: 'node_folder', movable: true, keep: false, offer: true,
target: 'user/__promptmanager', problem: null, size_bytes: 1_700_000_000,
target_source: 'default', config_moves: true,
});
function scope() {
@@ -46,6 +47,19 @@ test('the row says where the library is and why, the choice and any problem', as
assert.match(locationDescription({ path: 'prompts.db', source: 'config', movable: false }), /config\.json/);
});
test('the notice and the review name the actual target and what moves', async () => {
const { bannerText, reviewGrid, locationDescription } = await locationModule;
assert.match(bannerText(NODE_FOLDER), /Move it to ComfyUI's user folder: it is copied/);
assert.match(bannerText({ ...NODE_FOLDER, target_source: 'environment' }),
/Move it to the folder PROMPT_MANAGER_DATA_DIR names: it is copied/);
assert.match(bannerText(null), /ComfyUI's user folder/);
assert.equal(reviewGrid({ ...NODE_FOLDER, config_moves: false })[3][1],
'The database backups (config.json stays where PROMPT_MANAGER_CONFIG_PATH names it)');
assert.equal(reviewGrid({ ...NODE_FOLDER, config_moves: true })[3][1], 'config.json and the database backups');
assert.match(locationDescription({ path: 'custom_nodes/pm/data/my.db', source: 'config', movable: false,
problem: 'inside-node-folder' }), /inside the node folder, where an update or reinstall deletes it/);
});
test('only a well-formed location is shown', async () => {
const { isLocation } = await locationModule;
assert.equal(isLocation(NODE_FOLDER), true);
+48
View File
@@ -89,6 +89,8 @@ class TestLocationRoute(LibraryAPITestCase):
"target": "user/__promptmanager",
"problem": None,
"size_bytes": os.path.getsize(self.api.db.model.db_path),
"target_source": "default",
"config_moves": True,
},
)
@@ -123,6 +125,52 @@ class TestLocationRoute(LibraryAPITestCase):
self.assertEqual(refused["code"], "not-movable")
class TestLocationDetails(LibraryAPITestCase):
async def test_the_target_and_what_moves_follow_the_environment(self):
data = os.path.join(self.tmpdir, "pm-data")
self.environ = {
ll.DATA_DIR_ENV: data,
ll.CONFIG_PATH_ENV: os.path.join(self.tmpdir, "mine.json"),
}
location = (await self.json("GET", URL))["location"]
self.assertEqual(location["target_source"], "environment")
self.assertFalse(location["config_moves"])
self.assertEqual(location["problem"], "data-dir-empty")
async def test_a_configured_library_inside_the_node_folder_is_flagged(self):
self.api._library_resolver = lambda: ll.LibraryLocation(
ll.SOURCE_CONFIG,
self.api.db.model.db_path,
os.path.join(self.ext, "config.json"),
self.ext,
)
self.api._extension_root = self.ext
location = (await self.json("GET", URL))["location"]
self.assertEqual(location["problem"], "inside-node-folder")
self.assertFalse(location["movable"])
async def test_a_startup_after_a_crash_finishes_the_rename(self):
target_db = os.path.join(self.target, "prompts.db")
os.makedirs(self.target)
with open(target_db, "wb") as handle:
handle.write(b"x")
lm.write_marker(self.target, ll.MARKER_MOVED, self.api.db.model.db_path)
old = self.api.db.model.db_path
self.api.db.model.close_all()
moved = ll.LibraryLocation(
ll.SOURCE_DEFAULT,
target_db,
os.path.join(self.target, "config.json"),
self.target,
self.target,
)
self.api._library_resolver = lambda: moved
with patch.object(self.api.db.model, "db_path", target_db):
renamed = self.api.finish_interrupted_library_move()
self.assertEqual(renamed, ["prompts.db.moved-to-__promptmanager"])
self.assertFalse(os.path.exists(old))
class TestMoveRoute(LibraryAPITestCase):
async def test_move_runs_as_an_exclusive_job_and_switches_the_live_library(self):
body = await self.json("POST", URL + "/move", 202)
+43
View File
@@ -315,6 +315,49 @@ class TestDataDirValidation(_Case):
self.assertEqual(relative.problem, "data-dir-not-absolute")
class TestOneRuleForConfigAndDatabase(_Case):
"""M1: config.json and the database never split over PROMPT_MANAGER_DATA_DIR."""
def test_a_source_listed_only_in_a_config_file_rules_out_the_folder_for_both(self):
source = os.path.join(self.base, "gallery")
os.makedirs(source)
_touch(
os.path.join(self.ext, "config.json"),
json.dumps({"gallery": {"monitoring": {"directories": [source]}}}).encode(),
)
environ = {ll.DATA_DIR_ENV: os.path.join(source, "pm")}
location = self.resolve(environ=environ)
config_path, data_dir = self.config_location(environ=environ)
self.assertEqual(location.problem, "data-dir-served")
self.assertEqual(
(location.config_path, location.data_dir), (config_path, data_dir)
)
self.assertNotEqual(location.source, ll.SOURCE_ENVIRONMENT)
def test_a_relative_source_is_taken_from_comfyui(self):
self.fp.base_path = self.comfy
os.makedirs(os.path.join(self.comfy, "pictures"))
_touch(
os.path.join(self.ext, "config.json"),
json.dumps(
{"gallery": {"monitoring": {"directories": ["pictures"]}}}
).encode(),
)
environ = {ll.DATA_DIR_ENV: os.path.join(self.comfy, "pictures", "pm")}
self.assertEqual(self.resolve(environ=environ).problem, "data-dir-served")
self.assertEqual(self.config_location(environ=environ)[1], self.ext)
def test_unreadable_or_odd_config_files_add_no_source(self):
for payload in (
b"{not json",
b"[]",
json.dumps({"gallery": {"monitoring": {"directories": "x"}}}).encode(),
):
_touch(os.path.join(self.ext, "config.json"), payload)
environ = {ll.DATA_DIR_ENV: os.path.join(self.base, "pm-data")}
self.assertIsNone(self.resolve(environ=environ).problem)
class TestSameFolder(_Case):
def test_same_folder_follows_links_and_case(self):
self.assertTrue(ll.same_folder(self.ext, self.ext + os.sep))
+288
View File
@@ -0,0 +1,288 @@
"""Move library, final review round (#249 S6b PR #318).
- I1: a switch that fails at any point is undone while both files are still
locked: the copy is removed, the old library stays in use and a restart
keeps it. A follow-up after a complete switch only logs.
- I2: a background write (run counting, the image monitor) that fails
because Move library switched runs once more, on the new library; a write
racing the switch on a real connection lands in the new library.
- M2: the copy is verified through a file URI that survives ``#`` and ``%``.
- M6: a restart after a crash between the switch and the rename renames the
old files, never deletes them.
"""
import contextlib
import json
import os
import sqlite3
import sys
import threading
import unittest
from unittest import mock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database import library_move as lm # noqa: E402
from database import models # noqa: E402
from database.operations import PromptDatabase # noqa: E402
from tests.open_handles import assert_closed # noqa: E402
from utils import library_location as ll # noqa: E402
from utils import usage_tracking # noqa: E402
from utils.hashing import generate_prompt_hash # noqa: E402
try:
from tests.test_library_move import KEY, SimulatedCrash, _MoveCase, _rows
except ImportError: # discovered with tests/ as the top-level directory
from test_library_move import KEY, SimulatedCrash, _MoveCase, _rows
WAIT = 30
def _run_count(path, text):
uri = "file:" + path.replace(os.sep, "/") + "?mode=ro"
with contextlib.closing(sqlite3.connect(uri, uri=True)) as conn:
row = conn.execute("SELECT run_count FROM prompts WHERE text=?", (text,))
return row.fetchone()[0]
class TestFailedSwitchIsUndone(_MoveCase):
"""I1: whatever part of the switch fails, nothing was moved."""
def _assert_not_moved(self, caught):
self.assertEqual(caught.exception.code, "switch-failed")
self.assertIn("Nothing was moved", caught.exception.message)
self.assertIsNone(ll.current_override())
self.assertEqual(self.db.model.db_path, self.old_db)
# Writes keep landing in the old library, and a restart keeps it.
self.db.save_prompt("written after the failed switch")
self.assertIn("written after the failed switch", repr(_rows(self.old_db)))
self.assertEqual(self.location().source, ll.SOURCE_NODE_FOLDER)
self.assertNotEqual(ll.read_marker(self.target).get("state"), "moved")
self.assert_target_clean()
self.assertTrue(os.path.isfile(self.config))
assert_closed(self, self.target)
def _failing(self, point):
real_switch_to = ll.switch_to
real_repoint = models.PromptModel.repoint_instances
def switch(old_db, location):
if point == "before":
raise RuntimeError("switch failed before anything")
if point == "resolver":
real_switch_to(location)
raise RuntimeError("switch failed after the resolver")
if point == "repoint":
real_switch_to(location)
real_repoint(old_db, location.db_path)
raise RuntimeError("switch failed after the models")
detached = lm.switch_process(old_db, location)
models.PromptModel.close_detached(detached)
raise RuntimeError("switch failed at its very end")
return switch
def test_a_switch_failing_at_any_point_is_undone(self):
node = PromptDatabase(self.old_db)
node.get_prompt_by_id(1)
for point in ("before", "resolver", "repoint", "complete"):
with self.subTest(point=point):
with self.assertRaises(lm.MoveFailed) as caught:
self.move(switch=self._failing(point))
self._assert_not_moved(caught)
self.assertEqual(node.model.db_path, self.old_db)
self.assertEqual(node.get_prompt_by_id(1)["text"], "a cat number 0")
def test_a_failing_follow_up_after_the_switch_only_logs(self):
def follow_up():
raise RuntimeError("the health report could not be dropped")
with self.assertLogs("prompt_manager.database.library_move", "ERROR"):
result = self.move(after_switch=follow_up)
self.assertIsNone(result.retire_problem)
self.assertEqual(self.db.model.db_path, self.new_db)
self.assertTrue(
os.path.isfile(
os.path.join(self.ext, "prompts.db.moved-to-__promptmanager")
)
)
class _Flaky:
"""A write that fails once as if Move library switched while it waited."""
def __init__(self, db, name):
self.db, self.name, self.calls = db, name, 0
self.real = getattr(db, name)
def __call__(self, *args, **kwargs):
self.calls += 1
if self.calls == 1:
self.db.model.close_all() # what the switch does: a new generation
raise sqlite3.ProgrammingError("Cannot operate on a closed database.")
return self.real(*args, **kwargs)
class TestBackgroundWritesSurviveTheSwitch(_MoveCase):
"""I2: run counting and the image monitor write again on a fresh connection."""
def test_run_counting_retries_once_after_a_switch(self):
db = PromptDatabase(self.old_db)
text = "a counted prompt"
db.save_prompt(text, prompt_hash=generate_prompt_hash(text))
flaky = _Flaky(db, "record_prompt_use")
with mock.patch.object(db, "record_prompt_use", flaky):
pending = usage_tracking.PendingFirstUse()
usage_tracking.count_runs([(generate_prompt_hash(text), text)], db, pending)
self.assertEqual(flaky.calls, 2)
self.assertEqual(_run_count(self.old_db, text), 1)
def test_a_failure_without_a_switch_is_not_retried(self):
db = PromptDatabase(self.old_db)
calls = []
def broken():
calls.append(1)
raise sqlite3.OperationalError("disk I/O error")
with self.assertRaises(sqlite3.OperationalError):
db.survive_library_switch(broken)
self.assertEqual(calls, [1])
def test_the_image_monitor_link_retries_once_after_a_switch(self):
from utils.image_monitor import ImageGenerationHandler
db = PromptDatabase(self.old_db)
handler = ImageGenerationHandler(db, mock.MagicMock())
flaky = _Flaky(db, "link_image_to_prompt")
image = os.path.join(self.comfy, "output", "late.png")
with mock.patch.object(db, "link_image_to_prompt", flaky):
handler.link_image_to_prompt(image, {"id": 1}, {})
self.assertEqual(flaky.calls, 2)
self.assertIn("late.png", repr(_rows(self.old_db)["generated_images"]))
def test_a_write_racing_the_switch_lands_in_the_new_library(self):
text = "a racing prompt"
self.db.save_prompt(text, prompt_hash=generate_prompt_hash(text))
self.db.model.close_all()
# A short busy timeout so the blocked writer gives up in seconds.
with mock.patch.object(models, "BUSY_TIMEOUT_SECONDS", 2):
writer_db = PromptDatabase(self.old_db)
started = threading.Event()
outcome = {}
def write():
conn = writer_db.model.get_connection()
conn.set_trace_callback(
lambda sql: started.set() if sql.startswith("UPDATE") else None
)
try:
usage_tracking.count_runs(
[(generate_prompt_hash(text), text)],
writer_db,
usage_tracking.PendingFirstUse(),
)
outcome["ok"] = True
except Exception as error: # pragma: no cover - reported below
outcome["error"] = error
finally:
writer_db.model.close()
thread = threading.Thread(target=write)
def checkpoint(step):
if step == "locked":
# The old file is write-locked now: start the writer and
# wait until its UPDATE is running (it then waits on us).
thread.start()
self.assertTrue(started.wait(WAIT))
self.move(checkpoint=checkpoint)
thread.join(WAIT)
self.assertFalse(thread.is_alive())
self.assertEqual(outcome, {"ok": True})
self.assertEqual(_run_count(self.new_db, text), 1)
renamed = os.path.join(self.ext, "prompts.db.moved-to-__promptmanager")
self.assertEqual(_run_count(renamed, text), 0)
class TestVerifyUri(unittest.TestCase):
"""M2: a copy whose path holds ``#`` and ``%`` verifies."""
def test_copy_path_with_uri_characters_verifies(self):
import tempfile
with tempfile.TemporaryDirectory() as tmp:
folder = os.path.join(tmp, "lib #1 %20")
os.makedirs(folder)
path = os.path.join(folder, "copy.db")
db = PromptDatabase(path)
db.save_prompt("a prompt")
db.model.close_all()
with contextlib.closing(sqlite3.connect(path)) as conn:
conn.execute("PRAGMA journal_mode=DELETE")
counts, documents = lm._describe(conn)
lm._verify_copy(path, counts, documents)
models.PromptModel.close_all_instances()
class TestRestartFinishesTheRename(_MoveCase):
"""M6: the marker names the old files; a restart renames them."""
def _crash_after_switch(self):
def checkpoint(step):
if step == "switched":
raise SimulatedCrash(step)
with self.assertRaises(SimulatedCrash):
self.move(checkpoint=checkpoint)
models.PromptModel.close_all_instances()
ll.clear_override()
def test_a_restart_renames_the_old_files_never_deletes(self):
self._crash_after_switch()
self.assertTrue(os.path.isfile(self.old_db))
restarted = self.location()
self.assertEqual(restarted.db_path, self.new_db)
marker = ll.read_marker(self.target)
self.assertEqual(
(marker["from"], marker["config_from"]), (self.old_db, self.config)
)
renamed = lm.finish_interrupted_move(restarted)
self.assertEqual(
sorted(renamed),
[
"config.json.moved-to-__promptmanager",
"prompts.db.moved-to-__promptmanager",
],
)
self.assertFalse(os.path.exists(self.old_db))
moved_config = os.path.join(self.ext, "config.json.moved-to-__promptmanager")
with open(moved_config, encoding="utf-8") as handle:
self.assertEqual(
json.load(handle)["integrations"]["lora_manager"]["civitai_api_key"],
KEY,
)
self.assertEqual(lm.finish_interrupted_move(self.location()), [])
def test_nothing_is_renamed_while_the_old_file_is_in_use(self):
self._crash_after_switch()
reader = sqlite3.connect(self.old_db)
try:
reader.execute("BEGIN")
reader.execute("SELECT COUNT(*) FROM prompts").fetchone()
PromptDatabase(self.old_db).save_prompt("in the wal")
self.assertEqual(lm.finish_interrupted_move(self.location()), [])
self.assertTrue(os.path.isfile(self.old_db))
finally:
reader.rollback()
reader.close()
models.PromptModel.close_all_instances()
def test_a_node_folder_library_has_nothing_to_finish(self):
self.assertEqual(lm.finish_interrupted_move(self.location()), [])
self.assertTrue(os.path.isfile(self.old_db))
if __name__ == "__main__":
unittest.main()
+25 -12
View File
@@ -36,6 +36,7 @@ from .metadata_extractor import ComfyUIMetadataExtractor
from .logging_config import get_logger
from .prompt_graph import resolve_text, run_prompt_nodes
from .prompt_pair_resolver import resolve_prompt_pair
from .usage_tracking import survive_switch
try:
from ..database.pair_ingestion import ingest_resolved_pair
@@ -587,19 +588,31 @@ class ImageGenerationHandler(FileSystemEventHandler):
"""
try:
resolution = resolve_prompt_pair(metadata)
if resolution["state"] == "resolved":
positive = resolution["positive"]
expected = self.db_manager.get_prompt_by_hash(
generate_prompt_hash(positive.get("storage_text", positive["text"]))
)
if expected and expected["id"] == prompt_context["id"]:
ingest_resolved_pair(
self.db_manager, image_path, metadata, resolution
def link():
# One write; run again on the moved library when a Move
# library switched while it waited (#249 S6b).
if resolution["state"] == "resolved":
positive = resolution["positive"]
expected = self.db_manager.get_prompt_by_hash(
generate_prompt_hash(
positive.get("storage_text", positive["text"])
)
)
return
image_id = self.db_manager.link_image_to_prompt(
prompt_id=prompt_context["id"], image_path=image_path, metadata=metadata
)
if expected and expected["id"] == prompt_context["id"]:
ingest_resolved_pair(
self.db_manager, image_path, metadata, resolution
)
return None
return self.db_manager.link_image_to_prompt(
prompt_id=prompt_context["id"],
image_path=image_path,
metadata=metadata,
)
image_id = survive_switch(self.db_manager, link)
if image_id is None:
return
fallback_note = " (fallback)" if prompt_context.get("fallback") else ""
self.logger.debug(
f"Successfully linked image {image_id} to prompt "
+45 -1
View File
@@ -295,6 +295,47 @@ class _Decision:
target_source: Optional[str] = None
def _configured_sources(module, environ, extension_root, default, raw_env):
"""The image sources every config.json this resolver might read lists.
``config_location`` runs before any config.json is loaded, so it cannot be
handed the configured sources; both it and ``resolve_library`` read them
here, from each candidate file (the explicit one, the environment
folder's, the node folder's and the default folder's), so the two always
judge ``PROMPT_MANAGER_DATA_DIR`` alike. Relative sources are taken from
ComfyUI's base folder, as the gallery resolves them.
"""
candidates = [environ.get(CONFIG_PATH_ENV, "").strip()]
if raw_env and os.path.isabs(raw_env):
candidates.append(os.path.join(raw_env, CONFIG_NAME))
candidates.append(os.path.join(extension_root, CONFIG_NAME))
if default:
candidates.append(os.path.join(default, CONFIG_NAME))
base = getattr(module, "base_path", None) if module is not None else None
sources = []
for candidate in candidates:
if not candidate or not os.path.isfile(candidate):
continue
try:
with open(candidate, "r", encoding="utf-8") as handle:
data = json.load(handle)
directories = data["gallery"]["monitoring"]["directories"]
except (OSError, ValueError, KeyError, TypeError):
continue
if not isinstance(directories, list):
continue
for directory in directories:
if not isinstance(directory, str) or not directory.strip():
continue
directory = directory.strip()
if not os.path.isabs(directory):
if not base:
continue
directory = os.path.join(str(base), directory)
sources.append(directory)
return sources
def _decide(environ, extension_root, folder_paths, gallery_roots) -> _Decision:
"""Which folder holds the library (before an explicit database path)."""
module = _folder_paths(folder_paths)
@@ -306,11 +347,14 @@ def _decide(environ, extension_root, folder_paths, gallery_roots) -> _Decision:
env_dir = None
raw = environ.get(DATA_DIR_ENV, "")
if raw.strip():
sources = list(gallery_roots or ()) + _configured_sources(
module, environ, extension_root, default, raw.strip()
)
problem = data_dir_problem(
raw.strip(),
folder_paths=module,
environ=environ,
gallery_roots=gallery_roots,
gallery_roots=sources,
)
if problem is None:
env_dir = os.path.abspath(raw.strip())
+29 -8
View File
@@ -187,6 +187,19 @@ def find_pair_id(db: Any, positive: str, negative: str) -> Optional[int]:
return row["id"] if row else None
def survive_switch(db: Any, write: Callable[[], Any]) -> Any:
"""``write()`` through ``db.survive_library_switch`` when the database has it.
A write that was waiting on the old file when Move library switched runs
again on the new library instead of being lost (#249 S6b). Stand-in
databases without the method (tests) just run ``write()``.
"""
survive = getattr(type(db), "survive_library_switch", None)
if survive is None:
return write()
return survive(db, write)
def count_pair_runs(
runs: List[Tuple[str, str, str]],
db: Any,
@@ -194,20 +207,28 @@ def count_pair_runs(
) -> None:
"""Count existing recipes at dequeue and defer new recipes until first save."""
for key, positive, negative in runs:
pair_id = find_pair_id(db, positive, negative)
if pair_id:
db.record_prompt_pair_use(pair_id)
else:
def count(positive=positive, negative=negative):
pair_id = find_pair_id(db, positive, negative)
if pair_id:
db.record_prompt_pair_use(pair_id)
return bool(pair_id)
if not survive_switch(db, count):
pending.add(key)
def count_runs(runs: List[Run], db: Any, pending: PendingFirstUse) -> None:
"""Record one use per prompt in the database; remember the rest for first save."""
for prompt_hash, _text in runs:
existing = db.get_prompt_by_hash(prompt_hash)
if existing:
db.record_prompt_use(existing["id"])
else:
def count(prompt_hash=prompt_hash):
existing = db.get_prompt_by_hash(prompt_hash)
if existing:
db.record_prompt_use(existing["id"])
return bool(existing)
if not survive_switch(db, count):
pending.add(prompt_hash)
+20 -6
View File
@@ -26,8 +26,18 @@ export const KEEP_URL = `${LIBRARY_URL}/keep`;
export const MOVE_KIND = 'library_move';
export const BANNER_TITLE = 'Your library is inside the node folder';
export const BANNER_TEXT = 'Your library is stored inside the node folder, where an update or reinstall can delete it. '
+ 'Move it to ComfyUI\'s user folder: it is copied and verified first, and nothing is deleted.';
const BANNER_LEAD = 'Your library is stored inside the node folder, where an update or reinstall can delete it.';
/** Where Move library puts it, by the server's `target_source`. */
export const TARGET_TEXT = Object.freeze({
default: 'ComfyUI\'s user folder',
environment: 'the folder PROMPT_MANAGER_DATA_DIR names',
});
/** The notice's sentence: it names the actual target. */
export function bannerText(location) {
const target = TARGET_TEXT[location?.target_source] ?? TARGET_TEXT.default;
return `${BANNER_LEAD} Move it to ${target}: it is copied and verified first, and nothing is deleted.`;
}
export const BANNER_TEXT = bannerText({ target_source: 'default' });
export const KEEP_LABEL = 'Keep it here';
export const MOVE_LABEL = 'Move library';
export const ROW_TITLE = 'Library location';
@@ -48,6 +58,8 @@ export const PROBLEM_TEXT = Object.freeze({
'data-dir-served': 'PROMPT_MANAGER_DATA_DIR names a folder whose files are served over the web, so it was ignored.',
'data-dir-empty': 'PROMPT_MANAGER_DATA_DIR has no library yet; Move library puts this one there.',
'move-unfinished': 'An earlier move did not finish; this library is still the one in use.',
'inside-node-folder': 'It is inside the node folder, where an update or reinstall deletes it, config.json and its '
+ 'CivitAI key included. Set database.default_path or PROMPT_MANAGER_DATA_DIR to a folder outside it.',
});
const number = value => Number(value).toLocaleString('en-US');
@@ -81,7 +93,9 @@ export function reviewGrid(location) {
['From', location.path],
['To', `${location.target}/prompts.db`],
['Size', sizeText(location.size_bytes)],
['Moves with it', 'config.json and the database backups'],
['Moves with it', location.config_moves === false
? 'The database backups (config.json stays where PROMPT_MANAGER_CONFIG_PATH names it)'
: 'config.json and the database backups'],
['Old files', 'Renamed *.moved-to-…, never deleted'],
];
}
@@ -175,10 +189,10 @@ export function locationRow(location, { api, activity, scope, notify = showWorks
}
/** D129: the concept's `.notice` for a node-folder library: Keep it here and Move library. */
export function libraryNotice({ scope, onKeep, onMove }) {
export function libraryNotice({ scope, location = null, onKeep, onMove }) {
const notice = el('div', undefined, 'notice');
const text = el('div');
text.append(el('h2', BANNER_TITLE), el('p', BANNER_TEXT));
text.append(el('h2', BANNER_TITLE), el('p', bannerText(location)));
const keep = createButton(KEEP_LABEL);
const move = createButton(MOVE_LABEL, 'ca-primary');
scope.listen(keep, 'click', () => onKeep(keep));
@@ -214,7 +228,7 @@ export function mountLibraryBanner(container, { api, activity, scope, notify = s
container.replaceChildren();
container.hidden = !(isLocation(location) && location.offer === true);
if (container.hidden) return;
container.append(libraryNotice({ scope,
container.append(libraryNotice({ scope, location,
onKeep: async button => {
button.setAttribute('aria-disabled', 'true');
const saved = await keepLibrary({ api, scope, notify });