diff --git a/cista-front/src/repositories/WS.ts b/cista-front/src/repositories/WS.ts index c8d9db3..b760e17 100644 --- a/cista-front/src/repositories/WS.ts +++ b/cista-front/src/repositories/WS.ts @@ -110,8 +110,9 @@ function handleUpdateMessage(updateData: { update: UpdateEntry[] }) { delete node.dir[elem.name] break // Deleted elements can't have further children } - if (elem.name !== undefined) { + if (elem.name) { // @ts-ignore + console.log(node, elem.name) node = node.dir[elem.name] ||= {} } if (elem.key !== undefined) node.key = elem.key diff --git a/cista/api.py b/cista/api.py index f8d7fee..d7d17b8 100644 --- a/cista/api.py +++ b/cista/api.py @@ -1,5 +1,6 @@ import asyncio import typing +from secrets import token_bytes import msgspec from sanic import Blueprint @@ -100,9 +101,10 @@ async def watch(req, ws): } ).decode() ) + uuid = token_bytes(16) try: with watching.tree_lock: - q = watching.pubsub[ws] = asyncio.Queue() + q = watching.pubsub[uuid] = asyncio.Queue() # Init with disk usage and full tree await ws.send(watching.format_du()) await ws.send(watching.format_tree()) @@ -110,4 +112,4 @@ async def watch(req, ws): while True: await ws.send(await q.get()) finally: - del watching.pubsub[ws] + del watching.pubsub[uuid] diff --git a/cista/protocol.py b/cista/protocol.py index 02a35d9..131b0e0 100644 --- a/cista/protocol.py +++ b/cista/protocol.py @@ -45,7 +45,7 @@ class Rm(ControlBase): sel = [root / filename.sanitize(p) for p in self.sel] for p in sel: if p.is_dir(): - shutil.rmtree(p, ignore_errors=True) + shutil.rmtree(p) else: p.unlink() @@ -147,9 +147,9 @@ DirList = dict[str, FileEntry | DirEntry] class UpdateEntry(msgspec.Struct, omit_defaults=True): """Updates the named entry in the tree. Fields that are set replace old values. A list of entries recurses directories.""" - name: str = "" + name: str + key: str deleted: bool = False - key: str | None = None size: int | None = None mtime: int | None = None dir: DirList | None = None diff --git a/cista/watching.py b/cista/watching.py index df98116..f268b5f 100644 --- a/cista/watching.py +++ b/cista/watching.py @@ -6,6 +6,7 @@ from pathlib import Path, PurePosixPath import inotify.adapters import msgspec +from sanic.log import logging from cista import config from cista.fileio import fuid @@ -29,7 +30,7 @@ disk_usage = None def watcher_thread(loop): - global disk_usage + global disk_usage, rootpath while True: rootpath = config.config.path @@ -38,6 +39,7 @@ def watcher_thread(loop): with tree_lock: # Initialize the tree from filesystem tree[""] = walk(rootpath) + print(" ".join(tree[""].dir.keys())) msg = format_tree() if msg != old: asyncio.run_coroutine_threadsafe(broadcast(msg), loop) @@ -67,8 +69,8 @@ def watcher_thread(loop): try: update(path.relative_to(rootpath), loop) except Exception as e: - print("Watching error", e) - break + print("Watching error", e, path, rootpath) + raise i = None # Free the inotify object @@ -120,6 +122,8 @@ def walk(path: Path) -> DirEntry | FileEntry | None: def update(relpath: PurePosixPath, loop): """Called by inotify updates, check the filesystem and broadcast any changes.""" + if rootpath is None or relpath is None: + print("ERROR", rootpath, relpath) new = walk(rootpath / relpath) with tree_lock: update = update_internal(relpath, new) @@ -160,7 +164,7 @@ def update_internal( # Update parents update = [] for name, entry in elems[:-1]: - u = UpdateEntry(name) + u = UpdateEntry(name, entry.key) if szdiff: entry.size += szdiff u.size = entry.size @@ -170,14 +174,14 @@ def update_internal( # The last element is the one that changed name, entry = elems[-1] parent = elems[-2][1] if len(elems) > 1 else tree - u = UpdateEntry(name) + u = UpdateEntry(name, new.key if new else entry.key) if new: parent[name] = new if u.size != new.size: u.size = new.size if u.mtime != new.mtime: u.mtime = new.mtime - if isinstance(new, DirEntry) and u.dir == new.dir: + if isinstance(new, DirEntry) and u.dir != new.dir: u.dir = new.dir else: del parent[name] @@ -187,8 +191,12 @@ def update_internal( async def broadcast(msg): - for queue in pubsub.values(): - await queue.put_nowait(msg) + try: + for queue in pubsub.values(): + queue.put_nowait(msg) + except Exception: + # Log because asyncio would silently eat the error + logging.exception("Broadcast error") async def start(app, loop):