nodes/MajoorWaldi-ComfyUI-Majoor-AssetsManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor filtersController to manage event listeners with lifecycle support

Browse Source 
- Introduced `addManagedListener` function to handle event listeners with automatic cleanup.
- Updated event listeners for various filter controls to use the new management function.
- Added lifecycle signal support for event listeners to ensure proper disposal.
- Implemented cleanup logic to remove event listeners when no longer needed.

---

Update database schema handling to support async operations

- Changed `db.has_table` to `db.ahas_table` for async compatibility.
- Updated related database checks to use the new async method.

---

Refactor SQLite adapter to prioritize async operations

- Transitioned public API to expose async methods as primary.
- Removed synchronous methods and replaced them with async counterparts.
- Updated internal transaction handling to support async context management.
- Enhanced error handling and logging for async operations.

---

Add async support for FFProbe tool

- Implemented `aread` and `aread_batch` methods for asynchronous metadata reading.
- Utilized asyncio subprocess for non-blocking execution of ffprobe commands.

---

Enhance IndexScanner with I/O throttling

- Introduced `SCAN_IOPS_LIMIT` to control I/O operations during directory scans.
- Implemented pacing logic to manage scan I/O operations effectively.

---

Update MetadataService to utilize async reading methods

- Refactored methods to call async variants of exif and ffprobe reading.
- Improved error handling for async operations.

---

Implement path resolution checks in asset routes

- Added `_is_resolved_path_allowed` function to enforce allowed roots for file paths.
- Updated asset route handlers to validate resolved paths against allowed roots.

---

Refactor database maintenance routes for async handling

- Changed database close operations to use async `aclose` method.
- Updated error handling during database maintenance operations.

---

Update scan routes to utilize async database vacuuming

- Replaced synchronous database vacuum calls with async `avacuum` method.

---

Modify tests to support async database operations

- Updated test examples to use async context management for database interactions.
- Refactored integration tests to ensure proper async cleanup of database connections.
- Adjusted regression tests to count async transaction invocations correctly.
This commit is contained in:
Your Name
2026-02-16 01:05:48 +01:00
parent 0debbd53a2
commit 7183c87400
14 changed files with 417 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+29 -33
View File
@@ -138,7 +138,9 @@ python ComfyUI-Majoor-AssetsManager/scripts/install-requirements.py
```
### Tkinter Note (Embedded Python)
The native folder picker requires built-in `tkinter`. On embedded Python builds (for example `python_embeded`), `tkinter` is often missing, so the UI falls back to manual path entry.
The native folder picker (`/mjr/sys/browse-folder`) requires built-in `tkinter`.
On embedded Python builds (for example `python_embeded` / `python_embedded`), `tkinter` is often missing.
In that case (or on headless Linux without `DISPLAY`), the backend returns a picker-unavailable error and the UI should use manual path entry.
- The extension remains fully functional with manual path input.
- `pip install tk` is not a replacement for built-in `tkinter`.
@@ -257,10 +259,11 @@ Hotkeys are ignored while typing inside inputs.
Settings are stored under the `mjrSettings` key in `localStorage` (browser-side).
Backend persistence is currently limited to `probeBackend.mode` (stored in the SQLite `metadata` table). It uses a version bump so changes propagate quickly even with multiple backend instances. Other settings are intentionally browser-local and won't sync across machines/browsers.
Backend persistence includes `probeBackend.mode`, metadata fallback toggles, output-directory override, and security preferences (stored in SQLite `metadata`). UI layout/visual preferences remain browser-local (`localStorage`) and are not synced across machines/browsers.
Remote access & API Token (new):
- The UI now exposes **API token** settings allowing administrators to configure a token and enable remote access to the Assets Manager. Access is disabled by default and the token is stored in the backend settings; ensure you secure your network if enabling remote access. See the user guide for configuration details.
Remote access & API Token:
- The UI exposes an **API token** field for client-side header injection. The authoritative backend token still comes from environment variables (`MAJOOR_API_TOKEN` / `MJR_API_TOKEN`) at startup.
- Remote write behavior is controlled by backend security settings/env vars (`MAJOOR_REQUIRE_AUTH`, `MAJOOR_ALLOW_REMOTE_WRITE`, and security prefs).
- Page size (assets per request)
@@ -311,35 +314,28 @@ Canonical namespace is now `MJR_AM_...` to avoid collisions with ComfyUI/global
Compatibility aliases are still accepted (`MAJOOR_...` and some `MJR_...` names), but new deployments should prefer `MJR_AM_...`.
- `MAJOOR_OUTPUT_DIRECTORY` - override output directory
- `MAJOOR_EXIFTOOL_PATH` / `MAJOOR_EXIFTOOL_BIN` - ExifTool path
- `MAJOOR_FFPROBE_PATH` / `MAJOOR_FFPROBE_BIN` - FFprobe path
- `MAJOOR_MEDIA_PROBE_BACKEND` - `auto|exiftool|ffprobe|both`
- `MAJOOR_EXIFTOOL_MIN_VERSION` - minimum ExifTool version required (optional, e.g. `12.40`)
- `MAJOOR_FFPROBE_MIN_VERSION` - minimum FFprobe version required (optional, e.g. `6.0.0`)
- `MAJOOR_ENABLE_FILE_WATCHER` - auto reindexing (default `false`)
- `MAJOOR_WATCHER_INTERVAL` / `MAJOOR_WATCHER_JOIN_TIMEOUT` / `MAJOOR_WATCHER_PATHS`
- `MJR_WATCHER_DEBOUNCE_MS` - debounce delay (ms) for watcher batches (default `500`)
- `MJR_WATCHER_DEDUPE_TTL_MS` - dup suppression window (ms) before a file can be reprocessed (default `3000`)
- `MJR_WATCHER_PENDING_MAX` - max outstanding watcher events queued before new ones are dropped (default `5000`)
- `MJR_WATCHER_MAX_FLUSH_CONCURRENCY` - max concurrent watcher flush jobs (default `2`)
- `MJR_DB_CONSISTENCY_SAMPLE` - how many random DB entries to poke during periodic consistency sweeps (default `32`)
- `MJR_DB_CONSISTENCY_COOLDOWN_SECONDS` - minimum seconds between consistency sweeps (default `3600`)
- `MAJOOR_DB_TIMEOUT` / `MAJOOR_DB_MAX_CONNECTIONS` / `MAJOOR_DB_QUERY_TIMEOUT`
- `MAJOOR_TO_THREAD_TIMEOUT` - timeout (seconds) for `asyncio.to_thread(...)` work in HTTP handlers (default `30`)
- `MAJOOR_MAX_METADATA_JSON_BYTES` - max metadata JSON size stored in DB/cache (default `2097152`)
- `MJR_AM_OUTPUT_DIRECTORY` (alias: `MAJOOR_OUTPUT_DIRECTORY`) - override output directory
- `MJR_AM_EXIFTOOL_PATH` / `MJR_AM_FFPROBE_PATH` (aliases: `MAJOOR_*`) - tool executable paths
- `MJR_AM_MEDIA_PROBE_BACKEND` (alias: `MAJOOR_MEDIA_PROBE_BACKEND`) - `auto|exiftool|ffprobe|both`
- `MJR_AM_EXIFTOOL_MIN_VERSION` / `MJR_AM_FFPROBE_MIN_VERSION` (aliases: `MAJOOR_*`)
- `MJR_AM_ENABLE_WATCHER` (alias: `MJR_ENABLE_WATCHER`) - enable file watcher
- `MJR_AM_WATCHER_DEBOUNCE_MS` / `MJR_AM_WATCHER_DEDUPE_TTL_MS` (aliases: `MJR_WATCHER_*`)
- `MJR_AM_WATCHER_PENDING_MAX` / `MJR_AM_WATCHER_MAX_FLUSH_CONCURRENCY` (aliases: `MJR_WATCHER_*`)
- `MJR_AM_WATCHER_MIN_FILE_SIZE_BYTES` / `MJR_AM_WATCHER_MAX_FILE_SIZE_BYTES` (aliases include legacy `MAJOOR_WATCHER_*`)
- `MAJOOR_SCAN_IOPS_LIMIT` - optional scan I/O pacing (ops/sec) for directory walk on slow NAS/SMB (`0` disables)
- `MJR_DB_CONSISTENCY_SAMPLE` / `MJR_DB_CONSISTENCY_COOLDOWN_SECONDS` - periodic consistency sweep tuning
- `MJR_AM_DB_TIMEOUT` / `MJR_AM_DB_MAX_CONNECTIONS` / `MJR_AM_DB_QUERY_TIMEOUT` (aliases: `MAJOOR_*`)
- `MJR_AM_TO_THREAD_TIMEOUT` (alias: `MAJOOR_TO_THREAD_TIMEOUT`) - timeout for `asyncio.to_thread(...)` work in HTTP handlers
- `MJR_AM_MAX_METADATA_JSON_BYTES` (alias: `MAJOOR_MAX_METADATA_JSON_BYTES`)
- `MJR_AM_CLEANUP_RESERVED_NAMES` - set to `1` to enable Windows reserved-name cleanup in `custom_nodes` (disabled by default)
- `MJR_COLLECTION_MAX_ITEMS` - max items per collection JSON (default `50000`)
- `MJR_ALLOW_SYMLINKS` - allow symlink/junction custom roots (default `off`)
- `MAJOOR_TRUSTED_PROXIES` - comma-separated IPs/CIDRs allowed to supply `X-Forwarded-For`/`X-Real-IP` (default `127.0.0.1,::1`)
- `MAJOOR_API_TOKEN` (or `MJR_API_TOKEN`) - optional token to authorize write operations (send via `X-MJR-Token` or `Authorization: Bearer <token>`)
- `MAJOOR_REQUIRE_AUTH` - set to `1` to require `MAJOOR_API_TOKEN` even for localhost
- `MAJOOR_ALLOW_REMOTE_WRITE` - set to `1` to allow write operations from non-local clients without a token (**unsafe**)
- `MAJOOR_SAFE_MODE` - set to `0` to disable Safe Mode (default enabled)
- `MAJOOR_ALLOW_WRITE` - set to `1` to allow rating/tags writes while Safe Mode is enabled
- `MAJOOR_ALLOW_DELETE` - set to `1` to enable asset deletion (disabled by default)
- `MAJOOR_ALLOW_RENAME` - set to `1` to enable asset renaming (disabled by default)
- `MAJOOR_ALLOW_OPEN_IN_FOLDER` - set to `1` to enable `/mjr/am/open-in-folder` (disabled by default)
- `MAJOOR_REQUIRE_AUTH` - require token auth even for localhost
- `MAJOOR_ALLOW_REMOTE_WRITE` - allow non-local write operations without token (**unsafe**)
- `MAJOOR_SAFE_MODE` - safe-mode gate for write operations
- `MAJOOR_ALLOW_WRITE` / `MAJOOR_ALLOW_DELETE` / `MAJOOR_ALLOW_RENAME` / `MAJOOR_ALLOW_OPEN_IN_FOLDER` / `MAJOOR_ALLOW_RESET_INDEX` - operation gates (also configurable via persisted security preferences)
### API Token Setup
@@ -375,10 +371,10 @@ Once set, include the token in your API requests via one of these headers:
### Security Summary
- Without token: write operations are localhost-only.
- Without token: write operations are localhost-only (unless `MAJOOR_ALLOW_REMOTE_WRITE=1`).
- With token: write operations require authentication.
- Safe Mode is enabled by default.
- Destructive operations (delete/rename/open-in-folder) are disabled by default and must be explicitly enabled.
- Operation gates for delete/rename/open-in-folder/reset-index are enforced via security prefs and corresponding `MAJOOR_ALLOW_*` env vars.
- Safe Mode behavior is controlled by `MAJOOR_SAFE_MODE` and persisted security preferences.
You can persist the same secret in ComfyUI's Settings modal under **Security -> Majoor: API Token**. The Assets Manager stores it in browser storage and automatically sends both `X-MJR-Token` and `Authorization: Bearer ...` on behalf of the UI, so remote clients only need to match the shared secret between the backend environment and the settings field.
The backend still requires the environment variable (`MAJOOR_API_TOKEN`/`MJR_API_TOKEN`) when ComfyUI starts; the UI control just remembers it for you so the headers stay in sync.
@@ -387,8 +383,8 @@ The backend still requires the environment variable (`MAJOOR_API_TOKEN`/`MJR_API
- **CSRF**: state-changing endpoints require `X-Requested-With: XMLHttpRequest` (or `X-CSRF-Token`) and validate `Origin` vs `Host` when present.
- **Rate limiting**: in-memory per-client limits exist on expensive endpoints (search/scan/metadata/batch-zip). Client identity is based on IP; `X-Forwarded-For` is only honored when the connection comes from `MAJOOR_TRUSTED_PROXIES`.
- **Path safety**: file operations validate root containment for output/input/custom roots and reject paths outside allowed roots (symlink/junction handling is opt-in).
- **Write access**: destructive actions are loopback-only by default unless a token is configured.
- **Safe Mode**: write operations are disabled by default unless explicitly enabled via environment variables.
- **Write access**: loopback-only by default when no token is configured (can be relaxed by `MAJOOR_ALLOW_REMOTE_WRITE=1`).
- **Operation gating**: delete/rename/open-in-folder/reset-index and write behavior are enforced by security prefs + `MAJOOR_ALLOW_*` / `MAJOOR_SAFE_MODE`.
- **Batch ZIP**: ZIP building streams from an open file handle (avoids TOCTOU rename/replace races).
## UI Preview
js/features/panel/AssetsManagerPanel.js
+25 -12
View File
@@ -234,7 +234,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
updateSummaryBar?.({ state, gridContainer });
} catch {}
};
gridContainer.addEventListener("mjr:grid-stats", handler);
gridContainer.addEventListener("mjr:grid-stats", handler, { signal: panelLifecycleAC?.signal });
registerSummaryDispose(() => {
try {
gridContainer.removeEventListener("mjr:grid-stats", handler);
@@ -628,7 +628,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
gridController.reloadGrid();
} catch {}
};
gridContainer.addEventListener("mjr:reload-grid", _reloadGridHandler);
gridContainer.addEventListener("mjr:reload-grid", _reloadGridHandler, { signal: panelLifecycleAC?.signal });
const onCustomSubfolderChanged = async (e) => {
const next = String(e?.detail?.subfolder || "").trim().replaceAll("\\", "/");
const prev = currentFolderPath();
@@ -642,7 +642,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
} catch {}
}
};
gridContainer.addEventListener("mjr:custom-subfolder-changed", onCustomSubfolderChanged);
gridContainer.addEventListener("mjr:custom-subfolder-changed", onCustomSubfolderChanged, { signal: panelLifecycleAC?.signal });
try {
gridContainer._mjrHasCustomSubfolderHandler = true;
} catch {}
@@ -670,7 +670,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
} catch {}
} catch {}
};
gridContainer.addEventListener("mjr:badge-duplicates-focus", onDuplicateBadgeFocus);
gridContainer.addEventListener("mjr:badge-duplicates-focus", onDuplicateBadgeFocus, { signal: panelLifecycleAC?.signal });
registerSummaryDispose(() => {
try {
if (_reloadGridHandler) gridContainer.removeEventListener("mjr:reload-grid", _reloadGridHandler);
@@ -868,11 +868,11 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
btn.addEventListener("mouseenter", () => {
btn.style.borderColor = "rgba(145,205,255,0.4)";
btn.style.background = "linear-gradient(135deg, rgba(80,140,255,0.18), rgba(32,100,200,0.14))";
});
}, { signal: panelLifecycleAC?.signal });
btn.addEventListener("mouseleave", () => {
btn.style.borderColor = "rgba(120,190,255,0.18)";
btn.style.background = "linear-gradient(135deg, rgba(255,255,255,0.03), rgba(255,255,255,0.01))";
});
}, { signal: panelLifecycleAC?.signal });
return btn;
};
@@ -930,7 +930,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
} catch {}
popovers.close(pinnedFoldersPopover);
await gridController.reloadGrid();
});
}, { signal: panelLifecycleAC?.signal });
const unpinBtn = document.createElement("button");
unpinBtn.type = "button";
@@ -963,7 +963,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
} catch {}
await renderPinnedFoldersMenu();
await gridController.reloadGrid();
});
}, { signal: panelLifecycleAC?.signal });
row.appendChild(openBtn);
row.appendChild(unpinBtn);
@@ -1012,7 +1012,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
});
} catch {}
bindFilters({
const disposeFilters = bindFilters({
state,
kindSelect,
wfCheckbox,
@@ -1026,6 +1026,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
dateRangeSelect,
dateExactInput,
reloadGrid: gridController.reloadGrid,
lifecycleSignal: panelLifecycleAC?.signal || null,
onFiltersChanged: () => {
try {
agendaCalendar?.refresh?.();
@@ -1118,9 +1119,9 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
} catch {}
notifyContextChanged();
};
gridContainer.addEventListener("mjr:grid-stats", onStats);
gridContainer.addEventListener("mjr:selection-changed", onSelectionChanged);
document.addEventListener?.("mjr-settings-changed", onStats);
gridContainer.addEventListener("mjr:grid-stats", onStats, { signal: panelLifecycleAC?.signal });
gridContainer.addEventListener("mjr:selection-changed", onSelectionChanged, { signal: panelLifecycleAC?.signal });
document.addEventListener?.("mjr-settings-changed", onStats, { signal: panelLifecycleAC?.signal });
// Observe DOM changes for card count / selection class changes.
let raf = null;
@@ -1227,6 +1228,12 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
container._mjrVersionUpdateCleanup?.();
} catch {}
container._mjrVersionUpdateCleanup = null;
try {
container._mjrPanelState?._mjrDispose?.();
} catch {}
try {
container._mjrPanelState = null;
} catch {}
try {
if (_scrollTimer) clearTimeout(_scrollTimer);
_scrollTimer = null;
@@ -1245,6 +1252,9 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
try {
agendaCalendar?.dispose?.();
} catch {}
try {
disposeFilters?.();
} catch {}
try {
if (searchTimeout) clearTimeout(searchTimeout);
searchTimeout = null;
@@ -1266,6 +1276,9 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true }
try {
gridContainer?._mjrSummaryBarDispose?.();
} catch {}
try {
gridContainer?._mjrSummaryBarObserver?.disconnect?.();
} catch {}
try {
gridContainer?._mjrGridContextMenuUnbind?.();
} catch {}
js/features/panel/controllers/filtersController.js
+71 -20
View File
@@ -14,8 +14,33 @@ export function bindFilters({
dateRangeSelect,
dateExactInput,
reloadGrid,
onFiltersChanged = null
onFiltersChanged = null,
lifecycleSignal = null
}) {
const disposers = [];
const addManagedListener = (target, event, handler, options) => {
if (!target?.addEventListener || typeof handler !== "function") return;
try {
target.addEventListener(event, handler, options);
} catch {
try {
target.addEventListener(event, handler);
} catch {
return;
}
}
if (lifecycleSignal) return;
disposers.push(() => {
try {
target.removeEventListener(event, handler, options);
} catch {
try {
target.removeEventListener(event, handler);
} catch {}
}
});
};
// Cleanup previous window listener to prevent accumulation
if (_lastAgendaHandler) {
try {
@@ -24,27 +49,27 @@ export function bindFilters({
_lastAgendaHandler = null;
}
kindSelect.addEventListener("change", async () => {
addManagedListener(kindSelect, "change", async () => {
state.kindFilter = kindSelect.value || "";
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
wfCheckbox.addEventListener("change", async () => {
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
addManagedListener(wfCheckbox, "change", async () => {
state.workflowOnly = Boolean(wfCheckbox.checked);
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
ratingSelect.addEventListener("change", async () => {
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
addManagedListener(ratingSelect, "change", async () => {
state.minRating = Number(ratingSelect.value || 0) || 0;
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
if (workflowTypeSelect) {
workflowTypeSelect.addEventListener("change", async () => {
addManagedListener(workflowTypeSelect, "change", async () => {
state.workflowType = String(workflowTypeSelect.value || "").trim().toUpperCase();
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
}
const applySizeFilter = async () => {
@@ -59,8 +84,8 @@ export function bindFilters({
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
};
minSizeInput?.addEventListener("change", applySizeFilter);
maxSizeInput?.addEventListener("change", applySizeFilter);
addManagedListener(minSizeInput, "change", applySizeFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
addManagedListener(maxSizeInput, "change", applySizeFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
const applyResolutionFilter = async () => {
const minW = Number(minWidthInput?.value || 0);
@@ -73,11 +98,11 @@ export function bindFilters({
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
};
minWidthInput?.addEventListener("change", applyResolutionFilter);
minHeightInput?.addEventListener("change", applyResolutionFilter);
addManagedListener(minWidthInput, "change", applyResolutionFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
addManagedListener(minHeightInput, "change", applyResolutionFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
if (resolutionPresetSelect) {
resolutionPresetSelect.addEventListener("change", async () => {
addManagedListener(resolutionPresetSelect, "change", async () => {
const preset = String(resolutionPresetSelect.value || "");
const map = {
hd: [1280, 720],
@@ -92,10 +117,10 @@ export function bindFilters({
if (minHeightInput) minHeightInput.value = state.minHeight > 0 ? String(state.minHeight) : "";
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
}
if (dateRangeSelect) {
dateRangeSelect.addEventListener("change", async () => {
addManagedListener(dateRangeSelect, "change", async () => {
state.dateRangeFilter = dateRangeSelect.value || "";
if (state.dateRangeFilter && state.dateExactFilter) {
state.dateExactFilter = "";
@@ -105,14 +130,14 @@ export function bindFilters({
}
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
}
if (dateExactInput) {
const applyAgendaStyle = (status) => {
dateExactInput.classList.toggle("mjr-agenda-filled", status === "filled");
dateExactInput.classList.toggle("mjr-agenda-empty", status === "empty");
};
dateExactInput.addEventListener("change", async () => {
addManagedListener(dateExactInput, "change", async () => {
state.dateExactFilter = dateExactInput.value ? String(dateExactInput.value) : "";
if (state.dateExactFilter && state.dateRangeFilter) {
state.dateRangeFilter = "";
@@ -125,7 +150,7 @@ export function bindFilters({
}
try { onFiltersChanged?.(); } catch {}
await reloadGrid();
});
}, lifecycleSignal ? { signal: lifecycleSignal } : undefined);
const handleAgendaStatus = (event) => {
if (!event?.detail) return;
const { date, hasResults } = event.detail;
@@ -137,9 +162,35 @@ export function bindFilters({
applyAgendaStyle(hasResults ? "filled" : "empty");
};
try {
window?.addEventListener?.("MJR:AgendaStatus", handleAgendaStatus, { passive: true });
const opts = lifecycleSignal ? { passive: true, signal: lifecycleSignal } : { passive: true };
window?.addEventListener?.("MJR:AgendaStatus", handleAgendaStatus, opts);
_lastAgendaHandler = handleAgendaStatus;
if (!lifecycleSignal) {
disposers.push(() => {
try {
window.removeEventListener("MJR:AgendaStatus", handleAgendaStatus);
} catch {}
});
} else {
lifecycleSignal.addEventListener("abort", () => {
if (_lastAgendaHandler === handleAgendaStatus) _lastAgendaHandler = null;
}, { once: true });
}
} catch {}
}
return () => {
for (const dispose of disposers.splice(0)) {
try {
dispose();
} catch {}
}
if (_lastAgendaHandler) {
try {
window.removeEventListener("MJR:AgendaStatus", _lastAgendaHandler);
} catch {}
_lastAgendaHandler = null;
}
};
}
mjr_am_backend/adapters/db/schema.py
+1 -1
View File
@@ -404,7 +404,7 @@ def _schema_fingerprint() -> str:
async def _ensure_schema_fingerprint(db) -> Result[bool]:
if not db.has_table("metadata"):
if not await db.ahas_table("metadata"):
return Result.Ok(True)
try:
mjr_am_backend/adapters/db/sqlite.py
+65 -166
View File
@@ -1,9 +1,9 @@
"""
SQLite database connection manager.
Implementation note (Option A migration):
Implementation note (async-first migration):
- This adapter uses `aiosqlite` internally.
- The public API remains synchronous for compatibility with the existing codebase.
- The public API exposes async methods as primary surface.
Why this shape:
- Most of the backend logic (scanner/searcher/updater/schema) is synchronous today.
@@ -27,7 +27,8 @@ import time
import uuid
import os
import json
from contextlib import contextmanager, asynccontextmanager
import ctypes
from contextlib import asynccontextmanager
from pathlib import Path
from queue import Empty, Queue
from typing import Any, Dict, List, Optional, Tuple
@@ -381,8 +382,6 @@ class Sqlite:
self._write_lock: Optional[asyncio.Lock] = None
self._tx_write_lock_tokens: set[str] = set()
self._tx_state_lock = threading.Lock()
# Thread-local storage for sync transactions (caller thread).
self._tx_local = threading.local()
self._asset_locks: Dict[str, Dict[str, Any]] = {}
self._asset_locks_lock = threading.Lock()
self._malformed_recovery_lock = threading.Lock()
@@ -817,11 +816,6 @@ class Sqlite:
raise
def _tx_token(self) -> Optional[str]:
# Thread-local token for sync transaction() remains primary for back-compat.
# Fallback to ContextVar to support sync helper calls accidentally invoked from async flow.
tok = getattr(self._tx_local, "token", None) if hasattr(self, "_tx_local") else None
if tok:
return str(tok)
try:
ctx_tok = _TX_TOKEN.get()
except Exception:
@@ -1008,20 +1002,6 @@ class Sqlite:
except Exception:
raise
def execute(self, query: str, params: Optional[tuple] = None, fetch: bool = False) -> Result[Any]:
"""Execute SQL on the DB loop thread (sync)."""
try:
# Sync callers may be inside `transaction()` (token lives on the caller thread).
# Pass it explicitly because DB work runs on the loop thread.
token = self._tx_token()
return self._loop_thread.run(self._execute_async(query, params, fetch, tx_token=token))
except Exception as exc:
return Result.Err(ErrorCode.DB_ERROR, str(exc))
def query(self, sql: str, params: Optional[tuple] = None) -> Result[List[Dict[str, Any]]]:
"""Execute a SELECT query and return rows (sync)."""
return self.execute(sql, params, fetch=True)
async def aexecute(self, query: str, params: Optional[tuple] = None, fetch: bool = False) -> Result[Any]:
"""Execute SQL on the DB loop thread (async)."""
token = _TX_TOKEN.get()
@@ -1032,44 +1012,6 @@ class Sqlite:
"""Execute a SELECT query and return rows (async)."""
return await self.aexecute(sql, params, fetch=True)
def query_in(
self,
base_query: str,
column: str,
values: List[Any],
additional_params: Optional[tuple] = None,
) -> Result[List[Dict[str, Any]]]:
"""
Execute a query with an IN clause safely.
`base_query` must contain the `{IN_CLAUSE}` placeholder, which will be replaced
with `<column> IN (?, ?, ...)` using parameter binding for values.
"""
if not values:
return Result.Ok([])
if not isinstance(values, (list, tuple)):
return Result.Err(ErrorCode.INVALID_INPUT, "values must be a list or tuple")
ok_col, safe_col = _validate_and_repair_column_name(column)
if not ok_col or not safe_col:
return Result.Err(ErrorCode.INVALID_INPUT, f"Invalid column name: {column}")
ok_tpl, why = _validate_in_base_query(base_query)
if not ok_tpl:
return Result.Err(ErrorCode.INVALID_INPUT, why or "Invalid base_query template")
ok_q, query_or_err, _ = _build_in_query(str(base_query), safe_col, len(values))
if not ok_q:
return Result.Err(ErrorCode.INVALID_INPUT, str(query_or_err))
query = query_or_err
params = tuple(values)
if additional_params:
params = params + tuple(additional_params)
return self.query(query, params)
async def aquery_in(
self,
base_query: str,
@@ -1185,14 +1127,6 @@ class Sqlite:
except Exception:
raise
def executemany(self, query: str, params_list: List[Tuple]) -> Result[int]:
"""Execute a parameterized statement over multiple param tuples (sync)."""
try:
token = self._tx_token()
return self._loop_thread.run(self._executemany_async(query, params_list, tx_token=token))
except Exception as exc:
return Result.Err(ErrorCode.DB_ERROR, str(exc))
async def aexecutemany(self, query: str, params_list: List[Tuple]) -> Result[int]:
"""Execute a parameterized statement over multiple param tuples (async)."""
token = _TX_TOKEN.get()
@@ -1243,14 +1177,6 @@ class Sqlite:
return Result.Err(ErrorCode.DB_ERROR, str(exc))
return await self._with_query_timeout(_execute_inner())
def executescript(self, script: str) -> Result[bool]:
"""Execute a multi-statement SQL script (sync)."""
try:
token = self._tx_token()
return self._loop_thread.run(self._executescript_async(script, tx_token=token))
except Exception as exc:
return Result.Err(ErrorCode.DB_ERROR, str(exc))
async def aexecutescript(self, script: str) -> Result[bool]:
"""Execute a multi-statement SQL script (async)."""
token = _TX_TOKEN.get()
@@ -1271,27 +1197,11 @@ class Sqlite:
finally:
await self._release_connection_async(conn)
def vacuum(self) -> Result[bool]:
"""Run SQLite VACUUM (sync)."""
try:
return self._loop_thread.run(self._vacuum_async())
except Exception as exc:
return Result.Err(ErrorCode.DB_ERROR, str(exc))
async def avacuum(self) -> Result[bool]:
"""Run SQLite VACUUM (async)."""
fut = self._loop_thread.submit(self._vacuum_async())
return await asyncio.wrap_future(fut)
def has_table(self, table_name: str) -> bool:
"""Return True if `table_name` exists in sqlite_master."""
result = self.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name = ?",
(table_name,),
fetch=True,
)
return bool(result.ok and result.data and len(result.data) > 0)
async def ahas_table(self, table_name: str) -> bool:
"""Return True if `table_name` exists in sqlite_master (async)."""
result = await self.aexecute(
@@ -1301,23 +1211,6 @@ class Sqlite:
)
return bool(result.ok and result.data and len(result.data) > 0)
def get_schema_version(self) -> int:
"""Get the schema version from the `metadata` table (0 if missing)."""
if not self.has_table("metadata"):
return 0
result = self.execute(
"SELECT value FROM metadata WHERE key = 'schema_version'",
fetch=True,
)
if result.ok and result.data and len(result.data) > 0:
try:
return int(result.data[0]["value"])
except (ValueError, KeyError, TypeError):
logger.warning("Invalid schema_version value in database")
return 0
return 0
async def aget_schema_version(self) -> int:
"""Get the schema version from the `metadata` table (0 if missing) (async)."""
if not await self.ahas_table("metadata"):
@@ -1335,13 +1228,6 @@ class Sqlite:
return 0
return 0
def set_schema_version(self, version: int) -> Result[bool]:
"""Set the schema version in the `metadata` table."""
return self.execute(
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('schema_version', ?)",
(str(version),),
)
async def aset_schema_version(self, version: int) -> Result[bool]:
"""Set the schema version in the `metadata` table (async)."""
return await self.aexecute(
@@ -1470,9 +1356,6 @@ class Sqlite:
except Exception:
pass
# NOTE: `transaction()` is defined later (sync back-compat). Keeping a single
# implementation avoids subtle bugs (e.g. missing `_tx_local`).
async def _close_all_async(self):
# 1. Close transaction connections first.
tokens = list(self._tx_conns.keys())
@@ -1509,15 +1392,6 @@ class Sqlite:
# Reset counters/semaphores as everything is closed
self._async_sem = None
def close(self):
"""Close connections and stop the DB loop thread (sync)."""
try:
self._loop_thread.run(self._close_all_async())
except Exception as exc:
logger.debug("DB close failed: %s", exc)
finally:
self._loop_thread.stop()
async def aclose(self):
"""Close connections and stop the DB loop thread (async)."""
fut = self._loop_thread.submit(self._close_all_async())
@@ -1530,6 +1404,13 @@ class Sqlite:
"""Prepare for reset: lock new connections, wait for active ones, close all."""
self._resetting = True
# Best effort: force WAL checkpoint while connections are still alive.
# This reduces the chance of carrying stale WAL state into a reset on Windows.
try:
await self._checkpoint_wal_before_reset_async()
except Exception as exc:
logger.debug("DB Reset: WAL checkpoint pre-drain failed: %s", exc)
# Wait deterministically for in-flight checkouts to drain (max 5s), then force close.
drained = await asyncio.to_thread(self._active_conns_idle.wait, 5.0)
if not drained and self._active_conns:
@@ -1537,6 +1418,47 @@ class Sqlite:
await self._close_all_async()
async def _checkpoint_wal_before_reset_async(self) -> None:
"""
Best-effort checkpoint prior to hard reset.
Must be called on DB loop thread while connections are still valid.
"""
await self._ensure_initialized_async()
conn = await self._acquire_connection_async()
try:
for attempt in range(self._lock_retry_attempts + 1):
try:
cur = await conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
try:
await cur.fetchall()
finally:
try:
await cur.close()
except Exception:
pass
break
except sqlite3.OperationalError as exc:
if self._is_locked_error(exc) and attempt < self._lock_retry_attempts:
await self._sleep_backoff(attempt)
continue
raise
finally:
await self._release_connection_async(conn)
@staticmethod
def _schedule_delete_on_reboot(path: Path) -> bool:
"""
Windows fallback: request deletion on next reboot for locked files.
"""
if os.name != "nt":
return False
try:
MOVEFILE_DELAY_UNTIL_REBOOT = 0x4
ok = ctypes.windll.kernel32.MoveFileExW(str(path), None, MOVEFILE_DELAY_UNTIL_REBOOT)
return bool(ok)
except Exception:
return False
async def areset(self) -> Result[bool]:
"""
Aggressive reset:
@@ -1558,7 +1480,7 @@ class Sqlite:
# 3. Delete files with RETRY logic
base = str(self.db_path)
for ext in ["", "-wal", "-shm"]:
for ext in ["", "-wal", "-shm", "-journal"]:
p = Path(base + ext)
if not p.exists():
continue
@@ -1587,8 +1509,18 @@ class Sqlite:
logger.info(f"Renamed locked DB file to {trash}")
deleted = True
except Exception as rename_err:
# If main file fails, it's fatal. WAL/SHM failures are also fatal here
# because the user explicitly requested a hard reset.
# Final fallback (Windows): schedule delete at reboot.
# We still fail the reset to avoid reinitializing over potentially
# stale sidecar files in the current process/session.
scheduled = self._schedule_delete_on_reboot(p)
with self._lock:
self._resetting = False
if scheduled:
return Result.Err(
"DELETE_PENDING_REBOOT",
f"Could not delete {p} now; scheduled for deletion at reboot. "
f"Delete error: {last_error} | Rename failed: {rename_err}",
)
return Result.Err(
"DELETE_FAILED",
f"Could not delete {p}: {last_error} | Rename failed: {rename_err}",
@@ -1630,39 +1562,6 @@ class Sqlite:
fut = self._loop_thread.submit(self._rollback_tx_async(token))
return await asyncio.wrap_future(fut)
@contextmanager
def transaction(self, mode: str = "immediate"):
"""Context manager for a DB transaction (sync)."""
# Back-compat sync transaction: routes token via thread-local.
tx_state = Result.Ok(True)
begin_res = self._loop_thread.run(self._begin_tx_async(mode))
if not begin_res.ok or not begin_res.data:
tx_state = Result.Err(ErrorCode.DB_ERROR, str(begin_res.error or "Failed to begin transaction"))
yield tx_state
return
token = str(begin_res.data)
self._tx_local.token = token
try:
yield tx_state
commit_res = self._loop_thread.run(self._commit_tx_async(token))
if not commit_res.ok:
tx_state.ok = False
tx_state.code = str(getattr(commit_res, "code", None) or ErrorCode.DB_ERROR)
tx_state.error = str(commit_res.error or "Commit failed")
except Exception:
try:
self._loop_thread.run(self._rollback_tx_async(token))
except Exception:
pass
raise
finally:
try:
if hasattr(self._tx_local, "token"):
del self._tx_local.token
except Exception:
pass
@asynccontextmanager
async def atransaction(self, mode: str = "immediate"):
"""Async context manager for a DB transaction."""
mjr_am_backend/adapters/tools/ffprobe.py
+118
View File
@@ -5,6 +5,7 @@ import subprocess
import json
import shutil
import os
import asyncio
from pathlib import Path
from typing import Optional, List, Dict
@@ -184,6 +185,98 @@ class FFProbe:
quality="degraded"
)
async def aread(self, path: str) -> Result[dict]:
"""
Async variant of read() using native asyncio subprocess execution.
"""
if not self._available:
return Result.Err(
ErrorCode.TOOL_MISSING,
"ffprobe not found in PATH",
quality="none"
)
try:
cmd = [
self._resolved_bin or self.bin,
"-v", "error",
"-print_format", "json",
"-show_format",
"-show_streams",
path,
]
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
close_fds=os.name != "nt",
)
try:
stdout_b, stderr_b = await asyncio.wait_for(process.communicate(), timeout=self.timeout)
except asyncio.TimeoutError:
try:
process.kill()
except Exception:
pass
logger.error(f"ffprobe timeout for {path}")
return Result.Err(
ErrorCode.TIMEOUT,
f"ffprobe timeout after {self.timeout}s",
quality="degraded"
)
stdout = (stdout_b or b"").decode("utf-8", errors="replace")
stderr = (stderr_b or b"").decode("utf-8", errors="replace")
if process.returncode != 0:
stderr_msg = stderr.strip()
logger.warning(f"ffprobe error for {path}: {stderr_msg}")
return Result.Err(
ErrorCode.FFPROBE_ERROR,
stderr_msg or "ffprobe command failed",
quality="degraded"
)
if not stdout.strip():
logger.warning(f"ffprobe returned empty output for {path}")
return Result.Err(
ErrorCode.FFPROBE_ERROR,
"No ffprobe output",
quality="degraded"
)
data = json.loads(stdout)
if not isinstance(data, dict):
return Result.Err(
ErrorCode.PARSE_ERROR,
"Invalid ffprobe output format",
quality="degraded"
)
result = {
"format": data.get("format", {}),
"streams": data.get("streams", []),
"video_stream": self._find_video_stream(data.get("streams", [])),
"audio_stream": self._find_audio_stream(data.get("streams", [])),
}
return Result.Ok(result, quality="full")
except json.JSONDecodeError as e:
logger.error(f"ffprobe JSON parse error: {e}")
return Result.Err(
ErrorCode.PARSE_ERROR,
f"Failed to parse ffprobe output: {e}",
quality="degraded"
)
except Exception as e:
logger.error(f"ffprobe unexpected error: {e}")
return Result.Err(
ErrorCode.FFPROBE_ERROR,
str(e),
quality="degraded"
)
def read_batch(self, paths: List[str]) -> Dict[str, Result[dict]]:
"""
Read metadata from multiple video files.
@@ -223,6 +316,31 @@ class FFProbe:
return results
async def aread_batch(self, paths: List[str]) -> Dict[str, Result[dict]]:
"""
Async batch variant using bounded asyncio concurrency.
"""
if not self._available:
err: Result[dict] = Result.Err(ErrorCode.TOOL_MISSING, "ffprobe not found in PATH", quality="none")
return {str(p): err for p in paths}
if not paths:
return {}
max_workers = min(self._max_workers, len(paths))
sem = asyncio.Semaphore(max_workers)
results: Dict[str, Result[dict]] = {}
async def _one(path: str):
async with sem:
try:
results[str(path)] = await self.aread(path)
except Exception as e:
logger.error(f"FFProbe async batch error for {path}: {e}")
results[str(path)] = Result.Err(ErrorCode.FFPROBE_ERROR, str(e), quality="degraded")
await asyncio.gather(*[_one(p) for p in paths])
return results
def _find_video_stream(self, streams: list) -> dict:
"""Find first video stream in streams list."""
for stream in streams:
mjr_am_backend/features/index/scanner.py
+32 -1
View File
@@ -44,6 +44,11 @@ MAX_TRANSACTION_BATCH_SIZE = 500
MAX_SCAN_JOURNAL_LOOKUP = 5000
STAT_RETRY_COUNT = 3
STAT_RETRY_BASE_DELAY_S = 0.15
try:
# 0 disables scan I/O throttling (default: current behavior).
SCAN_IOPS_LIMIT = float(os.getenv("MAJOOR_SCAN_IOPS_LIMIT", "0") or 0.0)
except Exception:
SCAN_IOPS_LIMIT = 0.0
# Extensions explicitly excluded from indexing
_EXCLUDED_EXTENSIONS: set = {".psd", ".json", ".txt", ".csv", ".db", ".sqlite", ".log"}
@@ -97,6 +102,25 @@ class IndexScanner:
self._current_scan_id: Optional[str] = None
self._batch_fallback_count = 0
self._batch_fallback_lock = threading.Lock()
# Throttle only directory walk I/O operations (scandir entry processing).
self._scan_iops_limit = max(0.0, float(SCAN_IOPS_LIMIT))
self._scan_iops_next_ts = 0.0
def _scan_iops_wait(self) -> None:
"""
Best-effort I/O pacing for directory scans.
Runs in the walk producer thread to avoid blocking the event loop.
"""
limit = self._scan_iops_limit
if limit <= 0.0:
return
now = time.perf_counter()
next_ts = self._scan_iops_next_ts
if next_ts > now:
time.sleep(next_ts - now)
now = time.perf_counter()
step = 1.0 / limit if limit > 0.0 else 0.0
self._scan_iops_next_ts = max(next_ts, now) + step
@staticmethod
def _drain_walk_queue(q: "Queue[Optional[Path]]", max_items: int) -> list[Optional[Path]]:
@@ -122,6 +146,8 @@ class IndexScanner:
def _walk_and_enqueue(self, dir_path: Path, recursive: bool, stop_event: threading.Event, q: "Queue[Optional[Path]]") -> None:
"""Producer running on executor: walks filesystem and pushes paths into queue."""
# Reset pacing window for each full walk.
self._scan_iops_next_ts = 0.0
try:
for fp in self._iter_files(dir_path, recursive):
if stop_event.is_set():
@@ -1270,6 +1296,7 @@ class IndexScanner:
try:
with os.scandir(current) as it:
for entry in it:
self._scan_iops_wait()
try:
if entry.is_dir(follow_symlinks=False):
stack.append(Path(entry.path))
@@ -1294,6 +1321,7 @@ class IndexScanner:
continue
else:
for item in directory.iterdir():
self._scan_iops_wait()
if item.is_file() and _is_supported(item):
yield item
@@ -1312,7 +1340,10 @@ class IndexScanner:
fallback_count = int(self._batch_fallback_count)
except Exception:
fallback_count = 0
return {"batch_fallbacks_total": fallback_count}
return {
"batch_fallbacks_total": fallback_count,
"scan_iops_limit": float(self._scan_iops_limit),
}
def _chunk_file_batches(self, files: List[Path]) -> Iterable[List[Path]]:
"""Yield batches of files for bounded transactions."""
mjr_am_backend/features/metadata/service.py
+25 -7
View File
@@ -206,6 +206,24 @@ class MetadataService:
max_concurrency = 1
self._extract_sem = asyncio.Semaphore(max_concurrency)
async def _exif_read(self, file_path: str) -> Result[Dict[str, Any]]:
aread = getattr(self.exiftool, "aread", None)
if callable(aread):
return await aread(file_path)
return await asyncio.to_thread(self.exiftool.read, file_path)
async def _ffprobe_read(self, file_path: str) -> Result[Dict[str, Any]]:
aread = getattr(self.ffprobe, "aread", None)
if callable(aread):
return await aread(file_path)
return await asyncio.to_thread(self.ffprobe.read, file_path)
async def _ffprobe_read_batch(self, paths: list[str]) -> Dict[str, Result[Dict[str, Any]]]:
aread_batch = getattr(self.ffprobe, "aread_batch", None)
if callable(aread_batch):
return await aread_batch(paths)
return await asyncio.to_thread(self.ffprobe.read_batch, paths)
async def _enrich_with_geninfo_async(self, combined: Dict[str, Any]) -> None:
"""Helper to parse geninfo from prompt/workflow in combined metadata (Worker Thread)."""
prompt_graph = combined.get("prompt")
@@ -355,7 +373,7 @@ class MetadataService:
ext = os.path.splitext(file_path)[1].lower()
exif_start = time.perf_counter()
exif_result = await asyncio.to_thread(self.exiftool.read, file_path)
exif_result = await self._exif_read(file_path)
exif_duration = time.perf_counter() - exif_start
exif_data = exif_result.data if exif_result.ok else None
image_fallback_enabled, _ = await self._resolve_fallback_prefs()
@@ -411,7 +429,7 @@ class MetadataService:
image_fallback_enabled, _ = await self._resolve_fallback_prefs()
if allow_exif:
exif_start = time.perf_counter()
exif_result = await asyncio.to_thread(self.exiftool.read, file_path)
exif_result = await self._exif_read(file_path)
exif_duration = time.perf_counter() - exif_start
exif_data = exif_result.data if exif_result.ok else None
if not exif_result.ok:
@@ -479,7 +497,7 @@ class MetadataService:
_, media_fallback_enabled = await self._resolve_fallback_prefs()
if allow_exif:
exif_start = time.perf_counter()
exif_result = await asyncio.to_thread(self.exiftool.read, file_path)
exif_result = await self._exif_read(file_path)
exif_duration = time.perf_counter() - exif_start
exif_data = exif_result.data if exif_result.ok else None
if not exif_result.ok:
@@ -500,7 +518,7 @@ class MetadataService:
ffprobe_duration = 0.0
if allow_ffprobe:
ffprobe_start = time.perf_counter()
ffprobe_result = await asyncio.to_thread(self.ffprobe.read, file_path)
ffprobe_result = await self._ffprobe_read(file_path)
ffprobe_duration = time.perf_counter() - ffprobe_start
if not ffprobe_result.ok:
self._log_metadata_issue(
@@ -593,7 +611,7 @@ class MetadataService:
if allow_exif:
exif_start = time.perf_counter()
exif_result = await asyncio.to_thread(self.exiftool.read, file_path)
exif_result = await self._exif_read(file_path)
exif_duration = time.perf_counter() - exif_start
exif_data = exif_result.data if exif_result.ok else None
if not exif_result.ok:
@@ -609,7 +627,7 @@ class MetadataService:
if allow_ffprobe:
ffprobe_start = time.perf_counter()
ffprobe_result = await asyncio.to_thread(self.ffprobe.read, file_path)
ffprobe_result = await self._ffprobe_read(file_path)
ffprobe_duration = time.perf_counter() - ffprobe_start
ffprobe_data = ffprobe_result.data if ffprobe_result.ok else None
if not ffprobe_result.ok:
@@ -736,7 +754,7 @@ class MetadataService:
await self.exiftool.aread_batch(exif_targets) if exif_targets else {}
)
ffprobe_results: Dict[str, Result[Dict[str, Any]]] = (
await asyncio.to_thread(self.ffprobe.read_batch, ffprobe_targets) if ffprobe_targets else {}
await self._ffprobe_read_batch(ffprobe_targets) if ffprobe_targets else {}
)
def _exif_for(path: str) -> Optional[Dict[str, Any]]:
mjr_am_backend/routes/handlers/assets.py
+36 -10
View File
@@ -87,6 +87,15 @@ def _validate_filename(name: str) -> tuple[bool, str]:
return True, ""
def _is_resolved_path_allowed(path: Path) -> bool:
"""
Enforce allowed roots check on canonical paths.
"""
try:
return bool(_is_path_allowed(path) or _is_path_allowed_custom(path))
except Exception:
return False
def _delete_file_best_effort(path: Path) -> Result[bool]:
@@ -564,6 +573,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
resolved = candidate.resolve(strict=True)
except Exception:
return _json_response(Result.Err("NOT_FOUND", "File does not exist"))
if not _is_resolved_path_allowed(resolved):
return _json_response(Result.Err("FORBIDDEN", "Path is not within allowed roots"))
def _execute_command(command: list[str]) -> None:
if not shutil.which(command[0]):
@@ -745,6 +756,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
resolved = candidate.resolve(strict=True)
except Exception:
return _json_response(Result.Err("NOT_FOUND", "File does not exist"))
if not _is_resolved_path_allowed(resolved):
return _json_response(Result.Err("FORBIDDEN", "Path is not within allowed roots"))
if not (resolved.exists() and resolved.is_file()):
return _json_response(Result.Err("NOT_FOUND", "File does not exist"))
@@ -908,6 +921,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
current_resolved = current_path.resolve(strict=True)
except Exception:
return _json_response(Result.Err("NOT_FOUND", "Current file does not exist"))
if not _is_resolved_path_allowed(current_resolved):
return _json_response(Result.Err("FORBIDDEN", "Path is not within allowed roots"))
if not current_resolved.exists() or not current_resolved.is_file():
return _json_response(Result.Err("NOT_FOUND", "Current file does not exist"))
@@ -930,13 +945,22 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
)
)
async def _rollback_physical_rename() -> None:
try:
if new_path.exists() and not current_resolved.exists():
new_path.rename(current_resolved)
except Exception as rollback_exc:
logger.error("Failed to rollback rename for asset %s: %s", asset_id, rollback_exc)
# Update database record
try:
try:
mtime = int(new_path.stat().st_mtime)
except FileNotFoundError:
await _rollback_physical_rename()
return _json_response(Result.Err("NOT_FOUND", "Renamed file does not exist"))
except Exception as exc:
await _rollback_physical_rename()
return _json_response(
Result.Err(
"FS_ERROR",
@@ -947,7 +971,7 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
async with svc["db"].atransaction(mode="immediate"):
defer_fk = await svc["db"].aexecute("PRAGMA defer_foreign_keys = ON")
if not defer_fk.ok:
return _json_response(Result.Err("DB_ERROR", defer_fk.error or "Failed to defer foreign key checks"))
raise RuntimeError(defer_fk.error or "Failed to defer foreign key checks")
if asset_id is not None:
update_res = await svc["db"].aexecute(
@@ -955,12 +979,14 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
(new_name, str(new_path), mtime, asset_id)
)
if not update_res.ok:
return _json_response(Result.Err("DB_ERROR", update_res.error or "Failed to update assets filepath"))
raise RuntimeError(update_res.error or "Failed to update assets filepath")
else:
await svc["db"].aexecute(
up2 = await svc["db"].aexecute(
"UPDATE assets SET filename = ?, filepath = ?, mtime = ? WHERE filepath = ?",
(new_name, str(new_path), mtime, str(current_resolved)),
)
if not up2.ok:
raise RuntimeError(up2.error or "Failed to update assets filepath")
# Keep FK-linked tables in sync with renamed filepath.
sj_res = await svc["db"].aexecute(
@@ -968,20 +994,16 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
(str(new_path), str(current_resolved)),
)
if not sj_res.ok:
return _json_response(Result.Err("DB_ERROR", sj_res.error or "Failed to update scan_journal filepath"))
raise RuntimeError(sj_res.error or "Failed to update scan_journal filepath")
mc_res = await svc["db"].aexecute(
"UPDATE metadata_cache SET filepath = ? WHERE filepath = ?",
(str(new_path), str(current_resolved)),
)
if not mc_res.ok:
return _json_response(Result.Err("DB_ERROR", mc_res.error or "Failed to update metadata_cache filepath"))
raise RuntimeError(mc_res.error or "Failed to update metadata_cache filepath")
except Exception as exc:
try:
if new_path.exists() and not current_resolved.exists():
new_path.rename(current_resolved)
except Exception as rollback_exc:
logger.error("Failed to rollback rename for asset %s: %s", asset_id, rollback_exc)
await _rollback_physical_rename()
return _json_response(
Result.Err(
"DB_ERROR",
@@ -1091,6 +1113,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None:
resolved = candidate.resolve(strict=True)
except Exception:
resolved = None
if resolved is not None and not _is_resolved_path_allowed(resolved):
return _json_response(Result.Err("FORBIDDEN", f"Path for asset ID {asset_id} is not within allowed roots"))
validated_assets.append({"id": int(asset_id), "filepath": str(raw_path), "resolved": resolved})
@@ -1202,6 +1226,8 @@ async def download_asset(request: web.Request) -> web.StreamResponse:
resolved = candidate.resolve(strict=True)
except (OSError, RuntimeError, ValueError):
return web.Response(status=404, text="File not found")
if not _is_resolved_path_allowed(resolved):
return web.Response(status=403, text="Path is not within allowed roots")
if not resolved.is_file():
return web.Response(status=404, text="File not found")
mjr_am_backend/routes/handlers/db_maintenance.py
+1 -1
View File
@@ -313,7 +313,7 @@ def register_db_maintenance_routes(routes: web.RouteTableDef) -> None:
logger.warning("DB adapter areset() failed (%s), falling back to manual file delete", exc)
# Close whatever we can before manual delete
try:
db.close()
await db.aclose()
except Exception:
pass
mjr_am_backend/routes/handlers/scan.py
+1 -1
View File
@@ -1327,7 +1327,7 @@ def register_scan_routes(routes: web.RouteTableDef) -> None:
if scope == "all" and (clear_scan_journal or clear_metadata_cache):
# 1. Vacuum DB to reclaim space and rebuild file
try:
await asyncio.to_thread(db.execute, "VACUUM")
await db.avacuum()
logger.info("Database VACUUM completed during index reset")
except Exception as exc:
logger.warning(f"Failed to VACUUM database: {exc}")
tests/README.md
+4 -3
View File
@@ -170,11 +170,12 @@ pytest tests/ --cov=backend --cov-report=html
Each test should be independent using in-memory databases:
```python
def test_example():
@pytest.mark.asyncio
async def test_example():
db = Sqlite(":memory:")
migrate_schema(db)
await migrate_schema(db)
# ... test logic
db.close()
await db.aclose()
```
### Result pattern
tests/core/test_integration.py
+1 -1
View File
@@ -137,7 +137,7 @@ def main():
logger.info("=" * 60)
# Cleanup
services["db"].close()
asyncio.run(services["db"].aclose())
test_db = Path("test_integration.db")
if test_db.exists():
test_db.unlink()
tests/regressions/test_scan_batching.py
+8 -8
View File
@@ -8,7 +8,7 @@ async def test_scan_uses_batched_transactions(tmp_path: Path):
"""
Large scans should not BEGIN/COMMIT per file.
We count `db.transaction()` invocations during a scan and ensure it's bounded
We count `db.atransaction()` invocations during a scan and ensure it's bounded
by the batch size logic (<< number of files).
"""
from mjr_am_backend.deps import build_services
@@ -30,15 +30,15 @@ async def test_scan_uses_batched_transactions(tmp_path: Path):
files.append(p)
txn_calls = {"count": 0}
original_txn = db.transaction
original_txn = db.atransaction
@contextlib.contextmanager
def counted_transaction(*args, **kwargs):
@contextlib.asynccontextmanager
async def counted_transaction(*args, **kwargs):
txn_calls["count"] += 1
with original_txn(*args, **kwargs):
yield
async with original_txn(*args, **kwargs) as tx:
yield tx
db.transaction = counted_transaction # type: ignore[assignment]
db.atransaction = counted_transaction # type: ignore[assignment]
try:
res = await index.scan_directory(
str(scan_dir),
@@ -51,7 +51,7 @@ async def test_scan_uses_batched_transactions(tmp_path: Path):
)
assert res.ok, res.error
finally:
db.transaction = original_txn # type: ignore[assignment]
db.atransaction = original_txn # type: ignore[assignment]
# With default batching for ~120 files we expect ~3 transactions (50-sized batches),
# but allow a small buffer if batching logic changes slightly.