Unified watcher that *may* receive events from inotify and other sources. Added change notify messages from control and upload WebSockets. Cleanup debug printouts.

cista/__main__.py

@@ -141,7 +141,7 @@ def _main():
         settings["listen"] = listen
     elif not exists:
         settings["listen"] = ":8000"
-    operation = config.update_config(settings)
+    config.update_config(settings)
     # Prepare to serve
     url, opts = serve.parse_listen(config.config.listen)
     if not config.config.path.is_dir():

cista/api.py

@@ -1,5 +1,6 @@
 import asyncio
 import typing
+from pathlib import PurePosixPath
 from secrets import token_bytes
 
 import msgspec
@@ -53,6 +54,9 @@ async def upload(req, ws):
         if pos != req.end:
             d = f"{len(data)} bytes" if isinstance(data, bytes) else data
             raise ValueError(f"Expected {req.end - pos} more bytes, got {d}")
+        # Signal the watcher about the uploaded file and its parent directories
+        path = PurePosixPath(req.name)
+        watching.notify_change(path, *path.parents)
         # Report success
         res = StatusMsg(status="ack", req=req)
         await asend(ws, res)
@@ -87,6 +91,8 @@ async def control(req, ws):
     while True:
         cmd = msgspec.json.decode(await ws.recv(), type=ControlTypes)
         await asyncio.to_thread(cmd)
+        # Signal the watcher about affected paths
+        watching.notify_change(*cmd.affected_paths())
         await asend(ws, StatusMsg(status="ack", req=cmd))
 
 

cista/protocol.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import shutil
+from pathlib import PurePosixPath
 from typing import Any
 
 import msgspec
@@ -16,6 +17,10 @@ class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower):
     def __call__(self):
         raise NotImplementedError
 
+    def affected_paths(self) -> list[str]:
+        """Return list of paths affected by this operation for change notification."""
+        return []
+
 
 class MkDir(ControlBase):
     path: str
@@ -24,6 +29,9 @@ class MkDir(ControlBase):
         path = config.config.path / filename.sanitize(self.path)
         path.mkdir(parents=True, exist_ok=False)
 
+    def affected_paths(self) -> list[str]:
+        return [filename.sanitize(self.path)]
+
 
 class Rename(ControlBase):
     path: str
@@ -36,6 +44,11 @@ class Rename(ControlBase):
         path = config.config.path / filename.sanitize(self.path)
         path.rename(path.with_name(to))
 
+    def affected_paths(self) -> list[str]:
+        sanitized = filename.sanitize(self.path)
+        new_path = str(PurePosixPath(sanitized).with_name(filename.sanitize(self.to)))
+        return [sanitized, new_path]
+
 
 class Rm(ControlBase):
     sel: list[str]
@@ -49,6 +62,9 @@ class Rm(ControlBase):
             else:
                 p.unlink()
 
+    def affected_paths(self) -> list[str]:
+        return [filename.sanitize(p) for p in self.sel]
+
 
 class Mv(ControlBase):
     sel: list[str]
@@ -63,6 +79,13 @@ class Mv(ControlBase):
         for p in sel:
             shutil.move(p, dst)
 
+    def affected_paths(self) -> list[str]:
+        dst = filename.sanitize(self.dst)
+        paths = [filename.sanitize(p) for p in self.sel]
+        # Include new locations in dst
+        paths.extend(f"{dst}/{PurePosixPath(p).name}" for p in self.sel)
+        return paths
+
 
 class Cp(ControlBase):
     sel: list[str]
@@ -86,6 +109,11 @@ class Cp(ControlBase):
             else:
                 shutil.copy2(p, dst)
 
+    def affected_paths(self) -> list[str]:
+        dst = filename.sanitize(self.dst)
+        # Only destinations are new (sources unchanged)
+        return [f"{dst}/{PurePosixPath(filename.sanitize(p)).name}" for p in self.sel]
+
 
 ControlTypes = MkDir | Rename | Rm | Mv | Cp
 

cista/watching.py

@@ -1,4 +1,5 @@
 import asyncio
+import queue
 import shutil
 import sys
 import threading
@@ -9,13 +10,6 @@ from pathlib import Path, PurePosixPath
 from stat import S_ISDIR, S_ISREG
 
 import msgspec
-
-
-# Debug instrumentation
-def _dbg(msg):
-    print(f"[watch] {time.perf_counter():.3f} {msg}", file=sys.stderr, flush=True)
-
-
 from natsort import humansorted, natsort_keygen, ns
 from sanic.log import logger
 
@@ -53,7 +47,6 @@ def treeiter(rootmod):
 
 
 def treeget(rootmod: list[FileEntry], path: PurePosixPath):
-    t0 = time.perf_counter()
     begin = None
     ret = []
 
@@ -67,24 +60,16 @@ def treeget(rootmod: list[FileEntry], path: PurePosixPath):
             break
         ret.append(entry)
 
-    dur = time.perf_counter() - t0
-    if dur > 0.01:
-        _dbg(
-            f"treeget({path}) scanned {len(rootmod)} entries in {dur * 1000:.1f}ms, found {len(ret)} items"
-        )
     return begin, ret
 
 
 def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
     # Find the first entry greater than the new one
     # precondition: the new entry doesn't exist
-    t0 = time.perf_counter()
     isfile = 0
     level = 0
     i = 0
-    iter_count = 0
     for i, rel, entry in treeiter(rootmod):
-        iter_count += 1
         if entry.level > level:
             # We haven't found item at level, skip subdirectories
             continue
@@ -125,11 +110,6 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
     else:
         i += 1
 
-    dur = time.perf_counter() - t0
-    if dur > 0.01:
-        _dbg(
-            f"treeinspos({relpath}) iterated {iter_count}/{len(rootmod)} entries in {dur * 1000:.1f}ms -> pos {i}"
-        )
     return i
 
 
@@ -137,11 +117,32 @@ state = State()
 rootpath: Path = None  # type: ignore
 quit = threading.Event()
 
+# Thread-safe queue for signaling path updates from websockets
+_update_queue: queue.Queue[PurePosixPath] = queue.Queue()
+
+
+def notify_change(*paths: PurePosixPath | str):
+    """Signal that paths have changed. Called from control/upload websockets."""
+    for path in paths:
+        if isinstance(path, str):
+            path = PurePosixPath(path)
+        # Convert absolute paths to relative (strip leading /)
+        if path.is_absolute():
+            path = (
+                PurePosixPath(*path.parts[1:])
+                if len(path.parts) > 1
+                else PurePosixPath()
+            )
+        # Skip root paths (empty, '.') to avoid full tree walks
+        if not path.parts or path.parts == (".",):
+            continue
+        _update_queue.put(path)
+
+
 ## Filesystem scanning
 
 
 def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]:
-    t0 = time.perf_counter()
     path = rootpath / rel
     ret = []
     try:
@@ -193,62 +194,34 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]
         logger.error(f"Watching {path=}: {e!r}")
     if ret:
         ret[0] = entry
-    dur = time.perf_counter() - t0
-    if dur > 0.05:
-        _dbg(f"walk({rel}) returned {len(ret)} entries in {dur * 1000:.1f}ms")
     return ret
 
 
 def update_root(loop):
     """Full filesystem scan"""
-    t0 = time.perf_counter()
     old = state.root
     new = walk(PurePosixPath())
-    t_walk = time.perf_counter()
     if old != new:
         update = format_update(old, new)
-        t_format = time.perf_counter()
         with state.lock:
             broadcast(update, loop)
             state.root = new
-        t_done = time.perf_counter()
-        _dbg(
-            f"update_root: walk={t_walk - t0:.3f}s format={t_format - t_walk:.3f}s broadcast={t_done - t_format:.3f}s total={t_done - t0:.3f}s ({len(new)} entries)"
-        )
 
 
 def update_path(rootmod: list[FileEntry], relpath: PurePosixPath, loop):
     """Called on FS updates, check the filesystem and broadcast any changes."""
-    t0 = time.perf_counter()
     new = walk(relpath)
-    t_walk = time.perf_counter()
     obegin, old = treeget(rootmod, relpath)
-    t_get = time.perf_counter()
 
     if old == new:
-        dur = time.perf_counter() - t0
-        if dur > 0.01:
-            _dbg(
-                f"update_path({relpath}) no change, walk={t_walk - t0:.3f}s get={t_get - t_walk:.3f}s"
-            )
         return
 
     if obegin is not None:
         del rootmod[obegin : obegin + len(old)]
-    t_del = time.perf_counter()
 
     if new:
         i = treeinspos(rootmod, relpath, new[0].isfile)
-        t_inspos = time.perf_counter()
         rootmod[i:i] = new
-        t_ins = time.perf_counter()
-        _dbg(
-            f"update_path({relpath}) walk={t_walk - t0:.3f}s get={t_get - t_walk:.3f}s del={t_del - t_get:.3f}s inspos={t_inspos - t_del:.3f}s ins={t_ins - t_inspos:.3f}s total={t_ins - t0:.3f}s (old={len(old)} new={len(new)} tree={len(rootmod)})"
-        )
-    else:
-        _dbg(
-            f"update_path({relpath}) DELETED walk={t_walk - t0:.3f}s get={t_get - t_walk:.3f}s del={t_del - t_get:.3f}s (old={len(old)})"
-        )
 
 
 def update_space(loop):
@@ -268,7 +241,6 @@ def update_space(loop):
 
 
 def format_update(old, new):
-    t0 = time.perf_counter()
     # Make keep/del/insert diff until one of the lists ends
     oidx, nidx = 0, 0
     oremain, nremain = set(old), set(new)
@@ -279,7 +251,6 @@ def format_update(old, new):
     # candidates exist in both sequences but are not equal (rename/move cases)
     old_pos = {e: i for i, e in enumerate(old)}
     new_pos = {e: i for i, e in enumerate(new)}
-    t_setup = time.perf_counter()
 
     while oidx < len(old) and nidx < len(new):
         iteration_count += 1
@@ -366,13 +337,7 @@ def format_update(old, new):
     elif nremain:
         update.append(UpdIns(new[nidx:]))
 
-    t_diff = time.perf_counter()
+    return msgspec.json.encode({"update": update}).decode()
-    result = msgspec.json.encode({"update": update}).decode()
-    t_encode = time.perf_counter()
-    _dbg(
-        f"format_update: setup={t_setup - t0:.3f}s diff={t_diff - t_setup:.3f}s ({iteration_count} iters) encode={t_encode - t_diff:.3f}s total={t_encode - t0:.3f}s (old={len(old)} new={len(new)} ops={len(update)} msg={len(result)}B)"
-    )
-    return result
 
 
 def format_space(usage):
@@ -384,20 +349,11 @@ def format_root(root):
 
 
 def broadcast(msg, loop):
-    t0 = time.perf_counter()
     fut = asyncio.run_coroutine_threadsafe(abroadcast(msg), loop)
-    t_scheduled = time.perf_counter()
+    return fut.result()
-    result = fut.result()
-    t_done = time.perf_counter()
-    if t_done - t0 > 0.01:
-        _dbg(
-            f"broadcast: schedule={t_scheduled - t0:.3f}s wait={t_done - t_scheduled:.3f}s total={t_done - t0:.3f}s ({len(msg)}B to {result} clients)"
-        )
-    return result
 
 
 async def abroadcast(msg):
-    t0 = time.perf_counter()
     client_count = 0
     try:
         for queue in pubsub.values():
@@ -406,9 +362,6 @@ async def abroadcast(msg):
     except Exception:
         # Log because asyncio would silently eat the error
         logger.exception("Broadcast error")
-    dur = time.perf_counter() - t0
-    if dur > 0.001:
-        _dbg(f"abroadcast: {client_count} clients in {dur * 1000:.2f}ms")
     return client_count
 
 
@@ -558,6 +511,10 @@ def collapse_paths(paths: set[PurePosixPath]) -> set[PurePosixPath]:
     """Remove child paths if parent is in set."""
     if not paths:
         return paths
+    # Filter out root paths (empty or '.') which would cause full tree walks
+    paths = {p for p in paths if p.parts and p.parts != (".",)}
+    if not paths:
+        return set()
     # Sort by depth (fewest parts first)
     sorted_paths = sorted(paths, key=lambda p: len(p.parts))
     result = set()
@@ -574,28 +531,37 @@ def collapse_paths(paths: set[PurePosixPath]) -> set[PurePosixPath]:
     return result
 
 
-def watcher_inotify(loop):
+# Debounce settings
-    """Inotify watcher thread (Linux only)"""
+DEBOUNCE_DELAY = 0.01  # Wait 10ms after last event
-    import inotify.adapters
+DEBOUNCE_MAX = 0.1  # But no more than 100ms total
 
-    modified_flags = frozenset(
+
-        (
+def watcher(loop):
-            "IN_CREATE",
+    """Unified watcher thread handling inotify, websocket signals, and periodic scans."""
-            "IN_DELETE",
+    use_inotify = sys.platform == "linux"
-            "IN_DELETE_SELF",
+    inotify_tree = None
-            "IN_MODIFY",
+    modified_flags = frozenset()
-            "IN_MOVE_SELF",
+
-            "IN_MOVED_FROM",
+    if use_inotify:
-            "IN_MOVED_TO",
+        import inotify.adapters
+
+        modified_flags = frozenset(
+            (
+                "IN_CREATE",
+                "IN_DELETE",
+                "IN_DELETE_SELF",
+                "IN_MODIFY",
+                "IN_MOVE_SELF",
+                "IN_MOVED_FROM",
+                "IN_MOVED_TO",
+            )
         )
-    )
-
-    # Debounce settings
-    DEBOUNCE_DELAY = 0.1  # Wait 100ms after last event
-    DEBOUNCE_MAX = 0.5  # But no more than 500ms total
 
     while not quit.is_set():
-        inotify_tree = inotify.adapters.InotifyTree(rootpath.as_posix())
+        if use_inotify:
+            import inotify.adapters
+
+            inotify_tree = inotify.adapters.InotifyTree(rootpath.as_posix())
 
         # Initialize the tree from filesystem
         update_root(loop)
@@ -604,11 +570,44 @@ def watcher_inotify(loop):
         trefresh = time.monotonic() + 300.0
         tspace = time.monotonic() + 5.0
 
-        # Pending changes
+        # Pending changes: path -> {"ws": count, "inotify": count}
-        dirty_paths: set[PurePosixPath] = set()
+        dirty_paths: dict[PurePosixPath, dict[str, int]] = {}
         first_event_time: float | None = None
         last_event_time: float | None = None
-        event_count = 0
+
+        def add_dirty(path: PurePosixPath, source: str) -> bool:
+            """Add path to dirty set. Returns True if added, False if redundant."""
+            nonlocal first_event_time, last_event_time
+            # Check if already covered by an existing dirty path
+            for existing in dirty_paths:
+                if path == existing or (
+                    len(path.parts) > len(existing.parts)
+                    and path.parts[: len(existing.parts)] == existing.parts
+                ):
+                    # Count the event even if skipped
+                    dirty_paths[existing][source] = (
+                        dirty_paths[existing].get(source, 0) + 1
+                    )
+                    return False
+            # Remove any paths that would be covered by this new one
+            covered = {
+                p
+                for p in dirty_paths
+                if len(p.parts) > len(path.parts)
+                and p.parts[: len(path.parts)] == path.parts
+            }
+            # Aggregate counts from covered paths
+            counts: dict[str, int] = {source: 1}
+            for p in covered:
+                for s, c in dirty_paths[p].items():
+                    counts[s] = counts.get(s, 0) + c
+                del dirty_paths[p]
+            dirty_paths[path] = counts
+            now = time.monotonic()
+            if first_event_time is None:
+                first_event_time = now
+            last_event_time = now
+            return True
 
         while not quit.is_set():
             now = time.monotonic()
@@ -627,19 +626,20 @@ def watcher_inotify(loop):
             if dirty_paths:
                 time_since_last = now - last_event_time if last_event_time else 0
                 time_since_first = now - first_event_time if first_event_time else 0
-                if time_since_last >= DEBOUNCE_DELAY or time_since_first >= DEBOUNCE_MAX:
+                if (
+                    time_since_last >= DEBOUNCE_DELAY
+                    or time_since_first >= DEBOUNCE_MAX
+                ):
                     should_flush = True
 
             if should_flush:
-                t_start = time.perf_counter()
+                paths_to_process = dirty_paths.copy()
+                dirty_paths.clear()
+                first_event_time = None
+                last_event_time = None
 
                 # Collapse paths (remove children if parent present)
-                collapsed = collapse_paths(dirty_paths)
+                collapsed = collapse_paths(set(paths_to_process.keys()))
-                t_collapse = time.perf_counter()
-
-                _dbg(
-                    f"flush: {event_count} events -> {len(dirty_paths)} paths -> {len(collapsed)} collapsed"
-                )
 
                 # Process each collapsed path
                 new_root = path_index.root
@@ -647,20 +647,13 @@ def watcher_inotify(loop):
                     new_entries = walk(path)
                     new_root = path_index.apply_update(path, new_entries)
 
-                t_update = time.perf_counter()
-
                 # Broadcast if changed
                 if new_root != state.root:
                     try:
                         update_msg = format_update(state.root, new_root)
-                        t_format = time.perf_counter()
                         with state.lock:
                             broadcast(update_msg, loop)
                             state.root = new_root
-                        t_broadcast = time.perf_counter()
-                        _dbg(
-                            f"inotify->broadcast: collapse={t_collapse - t_start:.3f}s update={t_update - t_collapse:.3f}s format={t_format - t_update:.3f}s broadcast={t_broadcast - t_format:.3f}s TOTAL={t_broadcast - t_start:.3f}s"
-                        )
                     except Exception:
                         logger.exception("format_update failed; full rescan")
                         try:
@@ -676,63 +669,53 @@ def watcher_inotify(loop):
                                 broadcast(format_root(fresh), loop)
                                 state.root = fresh
 
-                # Reset pending state
+            # Collect events from websocket signals (non-blocking)
-                dirty_paths.clear()
+            try:
-                first_event_time = None
+                while True:
-                last_event_time = None
+                    path = _update_queue.get_nowait()
-                event_count = 0
+                    add_dirty(path, "ws")
+            except queue.Empty:
+                pass
 
-            # Collect inotify events (short timeout for responsiveness)
+            # Collect inotify events if available (short timeout for responsiveness)
-            for event in inotify_tree.event_gen(yield_nones=False, timeout_s=0.05):
+            if inotify_tree:
-                if quit.is_set():
+                for event in inotify_tree.event_gen(yield_nones=False, timeout_s=0.05):
-                    return
+                    if quit.is_set():
-                if not (modified_flags & set(event[1])):
+                        return
-                    continue
+                    if not (modified_flags & set(event[1])):
+                        continue
 
-                # Extract relative path
+                    # Extract relative path
-                path = PurePosixPath(event[2]) / event[3]
+                    path = PurePosixPath(event[2]) / event[3]
-                try:
+                    try:
-                    rel_path = path.relative_to(rootpath)
+                        rel_path = path.relative_to(rootpath)
-                except ValueError:
+                    except ValueError:
-                    continue
+                        continue
 
-                # Skip dotfiles
+                    # Skip dotfiles
-                if any(part.startswith(".") for part in rel_path.parts):
+                    if any(part.startswith(".") for part in rel_path.parts):
-                    continue
+                        continue
 
-                dirty_paths.add(rel_path)
+                    add_dirty(rel_path, "inotify")
-                event_count += 1
-                now = time.monotonic()
-                if first_event_time is None:
-                    first_event_time = now
-                last_event_time = now
 
-                # Don't block too long collecting events
+                    # Don't block too long collecting events
-                if now - first_event_time >= DEBOUNCE_MAX:
+                    now = time.monotonic()
-                    break
+                    if first_event_time and now - first_event_time >= DEBOUNCE_MAX:
+                        break
+            else:
+                # No inotify, just sleep briefly for responsiveness
+                time.sleep(0.05)
 
-        del inotify_tree
+        if inotify_tree:
-
+            del inotify_tree
-
-def watcher_poll(loop):
-    """Polling version of the watcher thread."""
-    while not quit.is_set():
-        t0 = time.perf_counter()
-        update_root(loop)
-        update_space(loop)
-        dur = time.perf_counter() - t0
-        if dur > 1.0:
-            logger.debug(f"Reading the full file list took {dur:.1f}s")
-        quit.wait(0.1 + 8 * dur)
 
 
 def start(app):
     global rootpath
     config.load_config()
     rootpath = config.config.path
-    use_inotify = sys.platform == "linux"
     app.ctx.watcher = threading.Thread(
-        target=watcher_inotify if use_inotify else watcher_poll,
+        target=watcher,
         args=[app.loop],
         # Descriptive name for system monitoring
         name=f"cista-watcher {rootpath}",