Add an option to stop polling on persistent errors

Author: krassowski

projects/jupyter-server-ydoc/jupyter_server_ydoc/app.py

@@ -50,6 +50,14 @@ class YDocExtension(ExtensionApp):
         saving changes from the front-end.""",
     )
 
+    file_stop_poll_on_errors_after = Float(
+        7 * 24 * 60 * 60,
+        allow_none=True,
+        config=True,
+        help="""The duration in seconds to stop polling a file after consecutive errors.
+        Defaults to 7 days, if None then polling will not stop on errors.""",
+    )
+
     document_cleanup_delay = Float(
         60,
         allow_none=True,
@@ -121,7 +129,10 @@ def initialize_handlers(self):
         # the global app settings in which the file id manager will register
         # itself maybe at a later time.
         self.file_loaders = FileLoaderMapping(
-            self.serverapp.web_app.settings, self.log, self.file_poll_interval
+            self.serverapp.web_app.settings,
+            self.log,
+            self.file_poll_interval,
+            file_stop_poll_on_errors_after=self.file_stop_poll_on_errors_after,
         )
 
         self.handlers.extend(

projects/jupyter-server-ydoc/jupyter_server_ydoc/loaders.py

@@ -5,7 +5,10 @@
 
 import asyncio
 from logging import Logger, getLogger
+from time import time
 from typing import Any, Callable, Coroutine
+from tornado.web import HTTPError
+from http import HTTPStatus
 
 from jupyter_server.services.contents.manager import (
     AsyncContentsManager,
@@ -29,12 +32,16 @@ def __init__(
         contents_manager: AsyncContentsManager | ContentsManager,
         log: Logger | None = None,
         poll_interval: float | None = None,
+        max_consecutive_logs: int = 3,
+        stop_poll_on_errors_after: float | None = None,
     ) -> None:
         self._file_id: str = file_id
 
         self._lock = asyncio.Lock()
         self._poll_interval = poll_interval
+        self._stop_poll_on_errors_after = stop_poll_on_errors_after
         self._file_id_manager = file_id_manager
+        self._max_consecutive_logs = max_consecutive_logs
         self._contents_manager = contents_manager
 
         self._log = log or getLogger(__name__)
@@ -204,8 +211,8 @@ async def _watch_file(self) -> None:
             return
 
         consecutive_error_logs = 0
-        max_consecutive_logs = 3
         suppression_logged = False
+        consecutive_errors_started = None
 
         while True:
             try:
@@ -214,13 +221,37 @@ async def _watch_file(self) -> None:
                     await self.maybe_notify()
                     consecutive_error_logs = 0
                     suppression_logged = False
+                    consecutive_errors_started = None
                 except Exception as e:
-                    if consecutive_error_logs < max_consecutive_logs:
-                        self._log.error(f"Error watching file: {self.path}\n{e!r}", exc_info=e)
+                    # We do not want to terminate the watcher if the content manager request
+                    # fails due to timeout, server error or similar temporary issue; we only
+                    # terminate if the file is not found or we get unauthorized error for
+                    # an extended period of time.
+                    if isinstance(e, HTTPError) and e.status_code in {
+                        HTTPStatus.NOT_FOUND,
+                        HTTPStatus.UNAUTHORIZED,
+                    }:
+                        if (
+                            consecutive_errors_started
+                            and self._stop_poll_on_errors_after is not None
+                        ):
+                            errors_duration = time() - consecutive_errors_started
+                            if errors_duration > self._stop_poll_on_errors_after:
+                                self._log.warning(
+                                    "Stopping watching file due to consecutive errors over %s seconds: %s",
+                                    self._stop_poll_on_errors_after,
+                                    self.path,
+                                )
+                                break
+                        else:
+                            consecutive_errors_started = time()
+                    # Otherwise we just log the error
+                    if consecutive_error_logs < self._max_consecutive_logs:
+                        self._log.error("Error watching file %s: %s", self.path, e, exc_info=e)
                         consecutive_error_logs += 1
                     elif not suppression_logged:
                         self._log.warning(
-                            "Too many errors while watching %s — suppressing further logs.",
+                            "Too many errors while watching %s - suppressing further logs.",
                             self.path,
                         )
                         suppression_logged = True
@@ -268,6 +299,7 @@ def __init__(
         settings: dict,
         log: Logger | None = None,
         file_poll_interval: float | None = None,
+        file_stop_poll_on_errors_after: float | None = None,
     ) -> None:
         """
         Args:
@@ -279,6 +311,7 @@ def __init__(
         self.__dict: dict[str, FileLoader] = {}
         self.log = log or getLogger(__name__)
         self.file_poll_interval = file_poll_interval
+        self._stop_poll_on_errors_after = file_stop_poll_on_errors_after
 
     @property
     def contents_manager(self) -> AsyncContentsManager | ContentsManager:
@@ -309,6 +342,7 @@ def __getitem__(self, file_id: str) -> FileLoader:
                 self.contents_manager,
                 self.log,
                 self.file_poll_interval,
+                stop_poll_on_errors_after=self._stop_poll_on_errors_after,
             )
             self.__dict[file_id] = file