LeoVasanko/mediahive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 11 Wiki Activity

Faster startup by loading existing index without any checks, then incremental scanning to update.

Browse Source
This commit is contained in:
LeoVasanko
2026-05-29 22:25:13 +00:00
parent 25e9bab57b
commit 550f67531a
8 changed files with 153 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/API.md
-1
View File
@@ -11,7 +11,6 @@ All media paths are scoped to a **root**, identified by a stable `root_id`.
| `GET` | `/api/config` | Returns the current root configuration. |
| `GET` | `/api/roots` | List all active roots with status. |
| `PUT` | `/api/roots` | Atomically replace the full root set. |
| `GET` | `/api/roots/{root_id}/index` | Returns the media index for one root. |
| `GET` | `/api/roots/{root_id}/status` | Returns scanner and library status for one root. |
| `POST` | `/api/roots/{root_id}/scan` | Triggers a new scan for one root. |
| `POST` | `/api/roots/{root_id}/play` | Opens a media file with the system player. |
docs/multi-index-plan.md
-1
View File
@@ -45,7 +45,6 @@ Every `Movie.id` and `Series.id` is namespaced with its `root_id`:
|----------|-------------|
| `GET /api/roots` | List all roots (name, path, root_id, status) |
| `PUT /api/roots` | Atomically replace full root map `{name: path}` |
| `GET /api/roots/{root_id}/index` | Full index for one root |
| `GET /api/roots/{root_id}/status` | Per-root scanning/loading/error state |
| `POST /api/roots/{root_id}/scan` | Trigger scan for one root |
| `WS /api/roots/{root_id}/ws` | Per-root WebSocket (init/upsert/remove/task) |
frontend/src/App.vue
+1 -1
View File
@@ -273,7 +273,7 @@ async function refreshRoots() {
path: r.path,
status: r.status,
})
if (r.status === "ready" || r.status === "scanning" || r.status === "loading") {
if (r.status === "ready" || r.status === "scanning") {
activeIds.push(r.root_id)
}
}
frontend/src/composables/useMediaWebSocket.ts
+32 -6
View File
@@ -16,6 +16,8 @@ interface RootState {
movieMap: Map<string, Movie>
seriesMap: Map<string, Series>
connected: boolean
initialized: boolean
pendingMessages: WsMessage[]
reconnectTimer: ReturnType<typeof setTimeout> | null
}
@@ -252,30 +254,48 @@ export function useMediaWebSocket() {
function updateMergedState() {
mediaIndex.value = buildIndex()
// Loading is done when at least one root has connected and sent init
let anyConnected = false
// Consider a root "connected" only after init is received.
let anyInitialized = false
for (const state of roots.value.values()) {
if (state.connected) {
anyConnected = true
if (state.connected && state.initialized) {
anyInitialized = true
break
}
}
if (anyConnected) {
if (anyInitialized) {
loading.value = false
error.value = null
}
connected.value = anyConnected
connected.value = anyInitialized
}
function processJson(state: RootState, text: string) {
const msg = JSON.parse(text) as WsMessage
// Prevent out-of-order corruption: buffer delta messages until we receive
// the initial full-state payload.
if (msg.type !== "init" && !state.initialized) {
state.pendingMessages.push(msg)
return
}
switch (msg.type) {
case "init": {
state.movieMap.clear()
state.seriesMap.clear()
for (const m of msg.data.movies) state.movieMap.set(m.id, m)
for (const s of msg.data.series) state.seriesMap.set(s.id, s)
state.initialized = true
// Replay any deltas that arrived before init completed.
if (state.pendingMessages.length > 0) {
const queued = state.pendingMessages
state.pendingMessages = []
for (const queuedMsg of queued) {
processJson(state, JSON.stringify(queuedMsg))
}
}
updateMergedState()
console.log(
`[WS ${state.rootId}] init: ${state.movieMap.size} movies, ${state.seriesMap.size} series`,
@@ -347,6 +367,8 @@ export function useMediaWebSocket() {
movieMap: new Map(),
seriesMap: new Map(),
connected: false,
initialized: false,
pendingMessages: [],
reconnectTimer: null,
}
roots.value.set(rootId, state)
@@ -359,6 +381,8 @@ export function useMediaWebSocket() {
ws.onopen = () => {
state.connected = true
state.initialized = false
state.pendingMessages = []
updateMergedState()
console.log(`[WS ${rootId}] Connected`)
}
@@ -367,6 +391,8 @@ export function useMediaWebSocket() {
ws.onclose = (ev) => {
state.connected = false
state.initialized = false
state.pendingMessages = []
state.ws = null
updateMergedState()
console.log(`[WS ${rootId}] Closed (code=${ev.code})`)
mediahive/hivescan/__main__.py
-1
View File
@@ -37,7 +37,6 @@ The server exposes per-root endpoints:
WS /api/roots/{root_id}/ws Live index updates & task progress
POST /api/roots/{root_id}/scan Trigger a new scan
GET /api/roots/{root_id}/status Current root status
GET /api/roots/{root_id}/index Full index as JSON (HTTP fallback)
""",
)
parser.add_argument(
mediahive/index_store.py
+87 -71
View File
@@ -33,6 +33,8 @@ logger = logging.getLogger("mediahive.index_store")
# Debounce interval for writing snapshots to disk (seconds)
SNAPSHOT_DEBOUNCE = 5.0
# Debounce interval for rebuilding the in-memory API snapshot (seconds)
SNAPSHOT_CACHE_DEBOUNCE = 0.25
class IndexStore:
@@ -67,6 +69,17 @@ class IndexStore:
self._snapshot_dirty = False
self._snapshot_task: asyncio.Task | None = None
# In-memory API snapshot cache (served by get_full_index)
self._snapshot_cache_dirty = True
self._snapshot_cache_task: asyncio.Task | None = None
self._cached_snapshot = IndexSnapshot(
generated_at=datetime.now().isoformat(),
media_root=self.media_root,
stats=MediaStats(),
movies=[],
series=[],
)
# ------------------------------------------------------------------
# Persistence
# ------------------------------------------------------------------
@@ -82,90 +95,55 @@ class IndexStore:
ap = AsyncPath(self.snapshot_path)
if not await ap.exists():
logger.info("No snapshot found at %s, starting fresh", self.snapshot_path)
self._schedule_snapshot_cache_refresh()
return
try:
raw = await ap.read_bytes()
await asyncio.to_thread(self._load_snapshot_sync, raw)
loaded_movies, loaded_series = await asyncio.to_thread(
self._load_snapshot_sync,
raw,
)
self._merge_loaded_snapshot(
loaded_movies,
loaded_series,
)
self._schedule_snapshot_cache_refresh()
except Exception:
logger.exception("Failed to load snapshot from %s", self.snapshot_path)
def _load_snapshot_sync(self, raw: bytes) -> None:
def _load_snapshot_sync(self, raw: bytes) -> tuple[list[Movie], list[Series]]:
"""Parse snapshot bytes in a thread-pool context."""
data = msgspec.json.decode(raw, type=IndexSnapshot)
loaded_movies: list[Movie] = []
loaded_series: list[Series] = []
for m in data.movies:
if m.showreel_source_sets:
filtered_source_sets = []
for source_set in m.showreel_source_sets:
filtered_sources = [
p
for p in source_set
if (
Path(self.media_root, p).exists()
if self.media_root
else Path(p).exists()
)
]
if filtered_sources:
filtered_source_sets.append(filtered_sources)
m.showreel_source_sets = filtered_source_sets or None
m.showreel_images = (
[source_set[0] for source_set in filtered_source_sets]
if filtered_source_sets
else None
)
elif m.showreel_images:
filtered_images = [
p
for p in m.showreel_images
if (
Path(self.media_root, p).exists()
if self.media_root
else Path(p).exists()
)
]
m.showreel_images = filtered_images or None
m.showreel_source_sets = (
[[p] for p in filtered_images] if filtered_images else None
)
m.id = self._maybe_migrate_id(m.id)
m.root_id = self.root_id
self.movies[m.id] = m
loaded_movies.append(m)
for s in data.series:
for season in s.seasons:
for ep in season.episodes:
if ep.reel_sources:
filtered_sources = [
p
for p in ep.reel_sources
if (
Path(self.media_root, p).exists()
if self.media_root
else Path(p).exists()
)
]
ep.reel_sources = filtered_sources or None
ep.reel_image = (
filtered_sources[0] if filtered_sources else None
)
elif ep.reel_image:
full = (
Path(self.media_root, ep.reel_image)
if self.media_root
else Path(ep.reel_image)
)
if full.exists():
ep.reel_sources = [ep.reel_image]
else:
ep.reel_image = None
ep.reel_sources = None
s.id = self._maybe_migrate_id(s.id)
s.root_id = self.root_id
self.series[s.id] = s
loaded_series.append(s)
logger.info(
"Loaded snapshot: %d movies, %d series",
len(self.movies),
len(self.series),
len(loaded_movies),
len(loaded_series),
)
return loaded_movies, loaded_series
def _merge_loaded_snapshot(
self,
movies: list[Movie],
series: list[Series],
) -> None:
"""Merge loaded snapshot items without overriding newer in-memory updates."""
for movie in movies:
if movie.id not in self.movies:
self.movies[movie.id] = movie
for show in series:
if show.id not in self.series:
self.series[show.id] = show
async def _write_snapshot(self) -> None:
"""Write current index to disk (called from debounce task)."""
@@ -190,6 +168,35 @@ class IndexStore:
self._snapshot_dirty = True
if self._snapshot_task is None or self._snapshot_task.done():
self._snapshot_task = asyncio.create_task(self._snapshot_writer())
self._schedule_snapshot_cache_refresh()
def _schedule_snapshot_cache_refresh(self) -> None:
"""Schedule a debounced rebuild of the in-memory API snapshot cache."""
self._snapshot_cache_dirty = True
if self._snapshot_cache_task is None or self._snapshot_cache_task.done():
self._snapshot_cache_task = asyncio.create_task(
self._snapshot_cache_writer()
)
async def _refresh_snapshot_cache_once(self) -> None:
"""Rebuild cached snapshot once using copied store values."""
movies = list(self.movies.values())
series = list(self.series.values())
self._cached_snapshot = await asyncio.to_thread(
self._build_snapshot_from_lists,
movies,
series,
)
async def _snapshot_cache_writer(self) -> None:
"""Refresh the in-memory snapshot cache while mutations are pending."""
while True:
await asyncio.sleep(SNAPSHOT_CACHE_DEBOUNCE)
if self._snapshot_cache_dirty:
self._snapshot_cache_dirty = False
await self._refresh_snapshot_cache_once()
else:
break
async def _snapshot_writer(self) -> None:
"""Flush to disk every SNAPSHOT_DEBOUNCE seconds while dirty."""
@@ -203,6 +210,12 @@ class IndexStore:
async def flush_snapshot(self) -> None:
"""Force-write a snapshot immediately (e.g. on shutdown)."""
if self._snapshot_cache_task and not self._snapshot_cache_task.done():
self._snapshot_cache_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._snapshot_cache_task
await self._refresh_snapshot_cache_once()
if self._snapshot_task and not self._snapshot_task.done():
self._snapshot_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
@@ -310,8 +323,11 @@ class IndexStore:
series: list[Series],
) -> IndexSnapshot:
"""Build a sorted IndexSnapshot with computed stats from list copies."""
movies_list = sorted(movies, key=lambda x: (x.title.lower(), x.year or 0))
series_list = sorted(series, key=lambda x: x.title.lower())
movies_list = sorted(
movies,
key=lambda x: ((x.title or "").lower(), x.year or 0),
)
series_list = sorted(series, key=lambda x: (x.title or "").lower())
total_movie_versions = sum(len(m.torrents) for m in movies_list)
total_series_episodes = sum(
@@ -339,5 +355,5 @@ class IndexStore:
)
def get_full_index(self) -> IndexSnapshot:
"""Return the full index as an IndexSnapshot."""
return self._build_snapshot()
"""Return the latest in-memory IndexSnapshot cache."""
return self._cached_snapshot
mediahive/root_registry.py
+20 -3
View File
@@ -105,9 +105,21 @@ class RootContext:
# Event queue and consumer
self._events: asyncio.Queue[ScanEvent] = asyncio.Queue()
self._consumer_task: asyncio.Task | None = None
self._startup_task: asyncio.Task | None = None
async def start(self) -> None:
"""Load snapshot and start event consumer."""
"""Start consumer immediately and load snapshot in the background."""
self.status = "loading"
self.error = None
if self._consumer_task is None or self._consumer_task.done():
self._consumer_task = asyncio.create_task(self._consume_events())
if self._startup_task is None or self._startup_task.done():
self._startup_task = asyncio.create_task(self._load_snapshot_background())
async def _load_snapshot_background(self) -> None:
"""Load snapshot without blocking root activation paths."""
try:
await self.store.load_snapshot()
self.status = "ready"
@@ -117,15 +129,20 @@ class RootContext:
len(self.store.movies),
len(self.store.series),
)
except asyncio.CancelledError:
raise
except Exception as exc:
self.status = "error"
self.error = str(exc)
logger.exception("Root %s failed to load snapshot", self.root_id)
self._consumer_task = asyncio.create_task(self._consume_events())
async def stop(self) -> None:
"""Stop consumer, flush snapshot, stop scanner."""
if self._startup_task and not self._startup_task.done():
self._startup_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._startup_task
if self.scanner is not None:
try:
await self.scanner.stop()
mediahive/server.py
+13 -9
View File
@@ -34,7 +34,6 @@ from mediahive.hivescan.images import close_image_client
from mediahive.hivescan.scanner import RootScanner
from mediahive.hivescan.tmdb_client import close_http_client
from mediahive.models.protocol import (
MsgspecResponse,
OpenFolderRequest,
PlayMediaRequest,
RootsRequest,
@@ -285,7 +284,7 @@ async def _attach_scanners() -> None:
"""Ensure every active root context has a running scanner."""
async with _attach_scanners_lock:
for ctx in supervisor.all_contexts().values():
if ctx.scanner is None and ctx.status in {"ready", "loading"}:
if ctx.scanner is None and ctx.status == "ready":
try:
scanner = RootScanner(ctx.root_id, ctx.root_path, ctx.send_event)
await scanner.start()
@@ -296,6 +295,13 @@ async def _attach_scanners() -> None:
)
async def _attach_scanners_loop() -> None:
"""Periodically attach scanners as roots transition to ready."""
while True:
await _attach_scanners()
await asyncio.sleep(1)
async def _activate_all_roots() -> None:
"""Background task: validate and activate all configured roots.
@@ -359,6 +365,7 @@ async def lifespan(_app: FastAPI):
# Defer root activation to a background task so the server starts
# immediately and macOS permission dialogs do not block startup.
activation_task = asyncio.create_task(_activate_all_roots())
scanner_attach_task = asyncio.create_task(_attach_scanners_loop())
logger.info("Server ready; waiting for root activation")
@@ -369,6 +376,10 @@ async def lifespan(_app: FastAPI):
with suppress(asyncio.CancelledError):
await activation_task
scanner_attach_task.cancel()
with suppress(asyncio.CancelledError):
await scanner_attach_task
await supervisor.shutdown()
with suppress(Exception):
@@ -434,13 +445,6 @@ async def put_roots(request: Request):
}
@app.get("/api/roots/{root_id}/index")
async def get_root_index(root_id: str):
"""Return the full index for a single root."""
ctx = _get_context(root_id)
return MsgspecResponse(ctx.store.get_full_index())
@app.get("/api/roots/{root_id}/status")
async def get_root_status(root_id: str):
"""Return status for a single root."""