Improve/dav buffer

README.md

@@ -61,6 +61,7 @@ This file handles the DAV server logic and RD polling.
 | `poll_interval_secs` | `10` | How often Buzz polls Real-Debrid for changes. |
 | `server.bind` | `0.0.0.0` | IP address the DAV server binds to. |
 | `server.port` | `9999` | Port for the DAV server. |
+| `server.stream_buffer_size` | `0` | Read-ahead buffer size in bytes for streaming media (e.g., 50MB: `52428800`). Set to `0` to disable. |
 | `state_dir` | `/app/data` | Path to store the SQLite DB and snapshots inside the container. |
 | `hooks.on_library_change` | `sh /app/scripts/media_update.sh` | Shell command executed when a change in the library is detected. |
 | `hooks.curator_url` | `http://buzz-curator:8400/rebuild` | Internal URL to trigger the Curator rebuild. |

buzz.dist.yml

@@ -4,6 +4,10 @@ poll_interval_secs: 10
 server:
   bind: 0.0.0.0
   port: 9999
+  # Read-ahead buffer size in bytes for streaming media (e.g., 52428800 for 50MB).
+  # Set to 0 to disable. When enabled, a background thread pre-fetches data from
+  # Real-Debrid into a bounded queue to smooth out network variations.
+  stream_buffer_size: 0
 state_dir: /app/data
 hooks:
   on_library_change: "bash /app/scripts/media_update.sh"

buzz/core/curator.py

@@ -16,6 +16,7 @@
     YEAR_RE,
 )
 from .events import record_event
+from .state import is_internal_category
 from .media import (
     is_sidecar_file,
     is_video_file,
@@ -506,8 +507,8 @@ def trigger_jellyfin_selective_refresh(
         return
 
     categories = {root.split("/")[0] for root in changed_roots if "/" in root}
-    # Filter out internal/virtual categories like __unplayable__ that shouldn't trigger scans
-    categories = {cat for cat in categories if cat != "__unplayable__"}
+    # Filter out internal/virtual categories like __unplayable__ that shouldn't trigger scans.
+    categories = {cat for cat in categories if not is_internal_category(cat)}
 
     if not categories:
         return

buzz/core/state.py

@@ -62,6 +62,10 @@ def canonical_snapshot(snapshot: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def is_internal_category(name: str) -> bool:
+    return name.startswith("__")
+
+
 class LibraryBuilder:
     def __init__(self, config: DavConfig):
         self.config = config
@@ -319,9 +323,9 @@ def _root_for_snapshot_path(self, path: str) -> str | None:
         parts = tuple(part for part in normalized.split("/") if part)
         if len(parts) < 2:
             return None
-        if parts[0] == "__all__":
+        if is_internal_category(parts[0]):
             return None
-        if parts[0] not in {"movies", "shows", "anime", "__unplayable__"}:
+        if parts[0] not in {"movies", "shows", "anime"}:
             return None
         return "/".join(parts[:2])
 
@@ -417,7 +421,9 @@ def sync(self, *, trigger_hook: bool = True) -> dict[str, Any]:
                     self.snapshot_digest = digest
                     self._write_json(self.snapshot_path, self.snapshot)
                     self.snapshot_loaded = True
-                    if trigger_hook and self.config.hook_command:
+                    if trigger_hook and (
+                        self.config.hook_command or self.config.curator_url
+                    ):
                         hook_paths = changed_paths
 
                 self.last_sync_at = report["timestamp"]
@@ -598,17 +604,45 @@ def _trigger_curator(self, changed_roots: list[str]) -> None:
     def _run_hook(self, changed_roots: list[str]) -> None:
         if not self.config.hook_command:
             return
-        # Filter out internal/virtual categories like __unplayable__
-        filtered_roots = [r for r in changed_roots if not r.startswith("__unplayable__")]
+        # Filter out internal/virtual categories like __unplayable__ and __all__.
+        filtered_roots = [
+            r for r in changed_roots if not is_internal_category(r.split("/", 1)[0])
+        ]
         if not filtered_roots:
             return
 
         self.verbose_log(f"Running library update hook: {self.config.hook_command}...")
         try:
             cmd = shlex.split(self.config.hook_command)
             cmd.extend(filtered_roots)
-            subprocess.run(cmd, check=True, timeout=60)
+            subprocess.run(
+                cmd,
+                check=True,
+                timeout=60,
+                capture_output=True,
+                text=True,
+            )
             self.verbose_log("Library update hook completed successfully")
+        except subprocess.TimeoutExpired as exc:
+            details = [f"Library update hook timed out after {exc.timeout}s: {exc.cmd}"]
+            stdout = (exc.stdout or "").strip()
+            stderr = (exc.stderr or "").strip()
+            if stdout:
+                details.append(f"stdout:\n{stdout}")
+            if stderr:
+                details.append(f"stderr:\n{stderr}")
+            record_event("\n".join(details), level="error")
+        except subprocess.CalledProcessError as exc:
+            details = [
+                f"Library update hook failed with exit code {exc.returncode}: {exc.cmd}"
+            ]
+            stdout = (exc.stdout or "").strip()
+            stderr = (exc.stderr or "").strip()
+            if stdout:
+                details.append(f"stdout:\n{stdout}")
+            if stderr:
+                details.append(f"stderr:\n{stderr}")
+            record_event("\n".join(details), level="error")
         except Exception as exc:
             record_event(f"Library update hook failed: {exc}", level="error")
 
@@ -821,9 +855,18 @@ def resolve_download_url(self, source_url: str, force_refresh: bool = False) ->
                 if download_url:
                     return download_url
 
-        download_url = self.client.unrestrict.link(source_url).json().get("download")
+        try:
+            res = self.client.unrestrict.link(source_url)
+            data = res.json()
+        except Exception as exc:
+            raise ValueError(f"Failed to unrestrict {source_url}: {exc}") from exc
+
+        download_url = data.get("download")
         if not download_url:
-            raise ValueError(f"Failed to resolve download link for {source_url}")
+            error_msg = data.get("error") or "no download link in response"
+            raise ValueError(
+                f"Failed to resolve download link for {source_url}: {error_msg}"
+            )
 
         with self.lock:
             self.resolved_urls[source_url] = {"download_url": download_url}
@@ -845,6 +888,25 @@ def __init__(self, state: BuzzState):
         self.state = state
         self._stop_event = threading.Event()
 
+    def _format_change_message(
+        self,
+        added: list[str],
+        removed: list[str],
+        updated: list[str],
+        synced: int,
+    ) -> str:
+        lines = [f"Real-Debrid library changed ({synced} torrents):"]
+        if added:
+            lines.append(f"  +{len(added)} added")
+            lines.extend(f"    {path}" for path in added)
+        if removed:
+            lines.append(f"  -{len(removed)} removed")
+            lines.extend(f"    {path}" for path in removed)
+        if updated:
+            lines.append(f"  ~{len(updated)} updated")
+            lines.extend(f"    {path}" for path in updated)
+        return "\n".join(lines)
+
     def run(self) -> None:
         while not self._stop_event.wait(self.state.config.poll_interval_secs):
             try:
@@ -854,15 +916,10 @@ def run(self) -> None:
                     removed = report.get("removed_paths", [])
                     updated = report.get("updated_paths", [])
                     synced = report.get("synced_torrents", 0)
-                    parts = []
-                    if added:
-                        parts.append(f"+{len(added)} added: {', '.join(added)}")
-                    if removed:
-                        parts.append(f"-{len(removed)} removed: {', '.join(removed)}")
-                    if updated:
-                        parts.append(f"~{len(updated)} updated: {', '.join(updated)}")
+                    if not any((added, removed, updated)):
+                        continue
                     record_event(
-                        f"Real-Debrid library changed: {'; '.join(parts)} ({synced} torrents)",
+                        self._format_change_message(added, removed, updated, synced),
                         event="realdebrid_update",
                     )
             except Exception as exc:  # noqa: BLE001

buzz/dav_app.py

@@ -1,6 +1,8 @@
 import hashlib
 import json
 import os
+import queue
+import threading
 from contextlib import asynccontextmanager
 from http import HTTPStatus
 from typing import Any
@@ -338,15 +340,77 @@ def stream_generator():
                     response, first_chunk = open_remote_media(
                         self.state, node, range_header
                     )
+
+                    chunk_size = 64 * 1024
+                    buffer_size = self.config.stream_buffer_size
+
+                    if buffer_size < chunk_size:
+                        try:
+                            if first_chunk:
+                                yield first_chunk
+                            while True:
+                                chunk = response.read(chunk_size)
+                                if not chunk:
+                                    break
+                                yield chunk
+                        finally:
+                            response.close()
+                        return
+
+                    # Buffered path: background thread reads ahead into a bounded queue.
+                    q = queue.Queue(maxsize=max(1, buffer_size // chunk_size))
+                    stop_event = threading.Event()
+
+                    def buffer_reader():
+                        try:
+                            while not stop_event.is_set():
+                                chunk = response.read(chunk_size)
+                                if not chunk:
+                                    break
+                                while not stop_event.is_set():
+                                    try:
+                                        q.put(chunk, timeout=1)
+                                        break
+                                    except queue.Full:
+                                        continue
+                        except Exception as exc:
+                            print(
+                                json.dumps(
+                                    {"event": "buffer_reader_error", "error": str(exc)},
+                                    sort_keys=True,
+                                ),
+                                flush=True,
+                            )
+                        finally:
+                            # Signal end-of-stream; use timeout to avoid hanging
+                            # if the queue is full and the consumer is gone.
+                            while not stop_event.is_set():
+                                try:
+                                    q.put(None, timeout=1)
+                                    break
+                                except queue.Full:
+                                    continue
+
+                    t = threading.Thread(target=buffer_reader, daemon=True)
+                    t.start()
+
                     try:
                         if first_chunk:
                             yield first_chunk
+
                         while True:
-                            chunk = response.read(64 * 1024)
-                            if not chunk:
+                            try:
+                                item = q.get(timeout=1)
+                            except queue.Empty:
+                                if not t.is_alive():
+                                    break
+                                continue
+                            if item is None:
                                 break
-                            yield chunk
+                            yield item
                     finally:
+                        stop_event.set()
+                        t.join(timeout=5)
                         response.close()
 
                 return StreamingResponse(

buzz/dav_protocol.py

@@ -54,10 +54,20 @@ def open_remote_media(
     if not source_url:
         raise ValueError("missing Real-Debrid source URL")
     last_error = "unable to resolve upstream media"
+    state.verbose_log(f"Opening remote media from {source_url!r}")
     for attempt in range(2):
-        download_url = state.resolve_download_url(
-            source_url, force_refresh=attempt == 1
-        )
+        try:
+            download_url = state.resolve_download_url(
+                source_url, force_refresh=attempt == 1
+            )
+        except Exception as exc:
+            last_error = str(exc)
+            state.verbose_log(f"Failed to resolve download URL: {exc}")
+            if attempt == 0:
+                continue
+            raise
+
+        state.verbose_log(f"Resolved to {download_url!r} (attempt {attempt + 1}/2)")
         req = request.Request(download_url, method="GET")
         if range_header:
             start, end = range_header
@@ -66,17 +76,29 @@ def open_remote_media(
             response = request.urlopen(req, timeout=60)
         except error.HTTPError as exc:
             state.invalidate_download_url(source_url)
-            last_error = f"upstream returned HTTP {exc.code}"
+            last_error = f"upstream returned HTTP {exc.code} for {download_url}"
+            state.verbose_log(
+                f"HTTP Error {exc.code} on attempt {attempt + 1}: {exc.reason}"
+            )
             if attempt == 0:
                 continue
             raise ValueError(last_error) from exc
+        except Exception as exc:
+            state.invalidate_download_url(source_url)
+            last_error = f"failed to connect to upstream: {exc}"
+            state.verbose_log(f"Connection error on attempt {attempt + 1}: {exc}")
+            if attempt == 0:
+                continue
+            raise ValueError(last_error) from exc
+
         try:
             first_chunk = validate_remote_media_response(response, range_header)
             return response, first_chunk
         except ValueError as exc:
             response.close()
             state.invalidate_download_url(source_url)
             last_error = str(exc)
+            state.verbose_log(f"Validation failed on attempt {attempt + 1}: {exc}")
             if attempt == 0:
                 continue
             raise

buzz/models.py

@@ -15,6 +15,7 @@ class DavConfig(BaseModel):
     poll_interval_secs: int = 10
     bind: str = "0.0.0.0"
     port: int = 9999
+    stream_buffer_size: int = 0
     state_dir: str = "/app/data"
     hook_command: str = ""
     anime_patterns: tuple[str, ...] = (DEFAULT_ANIME_PATTERN,)
@@ -52,6 +53,7 @@ def load(cls, path: str = DEFAULT_DAV_CONFIG_PATH) -> "DavConfig":
             poll_interval_secs=int(raw.get("poll_interval_secs", 10)),
             bind=str(server.get("bind", "0.0.0.0")),
             port=int(server.get("port", 9999)),
+            stream_buffer_size=int(server.get("stream_buffer_size", 0)),
             state_dir=str(raw.get("state_dir", "/app/data")),
             hook_command=str(hooks.get("on_library_change", "")).strip(),
             curator_url=str(

scripts/media_update.sh

@@ -6,9 +6,6 @@ case "$media_server" in
 plex)
     exec bash /app/scripts/plex_update.sh "$@"
     ;;
-jellyfin)
-    exec bash /app/scripts/jellyfin_update.sh "$@"
-    ;;
 *)
     echo "Unsupported MEDIA_SERVER: $media_server" >&2
     exit 1