fix(restore,threads): a stopped restore waits for its worker; exit-window thread refusals are quiet (#360, #375) (#404)

* fix(restore): a restore cut off by shutdown waits for its worker thread (#360)

A cancelled restore handler or job no longer removes the stage or the
upload, or ends its exclusive registry admission, while the executor
thread still stages or publishes. The cancel waits for the worker; a
publish that finished is adopted before the cancel goes on, and a stage
finished after the cancel is removed.

* fix(threads): a thread or pool refused at interpreter exit is quiet and leaves nothing pending (#375)

During interpreter exit, threading._shutdown shuts every thread pool
before the package teardown runs while our daemon threads keep working.
map_parallel now runs the items a refused pool did not start on the
calling thread, schedule() drops the path and returns False when its
worker cannot start, and the preview queue skips the file. Each
refusal is logged once at DEBUG without a traceback; other start
failures behave as before.

* fix(restore): keep the cancel and the adoption when a stopped restore's publish ends (#360)

A publish that fails after the cancel now ends in CancelledError (after
its prune) instead of a 500, and the restored library is adopted to its
end even when a second cancel arrives. The blocking calls go through
_run_in_executor again, as the executor contract test expects.

Exit-window refusals: the comments state what was measured (a pool is
refused during threading._shutdown; Thread.start only once the
interpreter is finalizing), and the tests assert the one DEBUG line.

py/api/backup_restore.py

@@ -483,13 +483,59 @@ class BackupRestoreRoutesMixin:
             self.logger.info(f"Removed {len(removed)} old safety backups")
 
     async def _publish(self, model, staged):
-        """Publish ``staged``; on a page-copy failure prune and re-raise."""
+        """Publish ``staged``; on a page-copy failure prune and re-raise.
+
+        A cancel (server stop) waits for the publishing thread (#360): the
+        caller's cleanup and its registry admission must outlive it. When the
+        copy then succeeded the library was replaced, so it is adopted
+        (``_after_publish``, itself run to its end) before the cancel goes
+        on; a copy that failed is handled as usual, then the cancel goes on.
+        """
+        future = self._in_executor_task(model.publish_staged, staged)
+        cancelled = await _outlive_cancel(future)
         try:
-            return await self._run_in_executor(model.publish_staged, staged)
+            backup_path = future.result()
         except sqlite3.OperationalError as exc:
             if getattr(exc, "_prompt_manager_restore_backup_created", False) is True:
                 await self._prune_after(model.db_path, failed=True)
+            if cancelled:
+                raise asyncio.CancelledError from exc
             raise
+        except Exception as exc:
+            if cancelled:
+                raise asyncio.CancelledError from exc
+            raise
+        if cancelled:
+            self.logger.info("The restore finished while the server was stopping")
+            # A second cancel (the loop's own shutdown) must not cut it short.
+            adopted = asyncio.ensure_future(self._after_publish(model.db_path))
+            await _outlive_cancel(adopted)
+            if not adopted.cancelled() and adopted.exception() is not None:
+                self.logger.error(
+                    "Could not adopt the restored library",
+                    exc_info=adopted.exception(),
+                )
+            raise asyncio.CancelledError
+        return backup_path
+
+    async def _stage_to_end(self, model, temp_path):
+        """Stage ``temp_path``; a cancel waits for the worker (#360).
+
+        The upload stays until the staging thread is done with it, and a
+        stage it finished after the cancel is removed before the cancel goes
+        on.
+        """
+        future = self._in_executor_task(model.stage_restore, temp_path)
+        if await _outlive_cancel(future):
+            if not future.cancelled() and future.exception() is None:
+                await self._discard_held([_Orphan(future.result())])
+            raise asyncio.CancelledError
+        return future.result()
+
+    def _in_executor_task(self, func, *args):
+        """``_run_in_executor(func, *args)`` as a task ``_outlive_cancel`` can
+        wait for (shielded, the executor call is never cancelled)."""
+        return asyncio.ensure_future(self._run_in_executor(func, *args))
 
     async def _after_publish(self, db_path):
         """Everything after the copy replaced the library: best effort, logged.
@@ -609,7 +655,7 @@ class BackupRestoreRoutesMixin:
                     {"success": False, "error": REQUESTS_BUSY}, status=409
                 )
             model = self.db.model
-            staged = await self._run_in_executor(model.stage_restore, temp_path)
+            staged = await self._stage_to_end(model, temp_path)
             try:
                 backup_path = await self._publish(model, staged)
             finally:
@@ -927,6 +973,30 @@ class BackupRestoreRoutesMixin:
         job.update(progress=100, status_line=line)
 
 
+async def _outlive_cancel(future):
+    """Await ``future`` (an executor call's task) to its end, even through
+    cancels (#360).
+
+    A cancelled task cannot stop the worker thread, so the files it uses and
+    the registry admission it runs under must outlive it: the cancel is held
+    until the worker returned. Returns whether the task was cancelled
+    meanwhile; the caller reads the outcome from ``future`` and re-raises
+    ``CancelledError``. A future cancelled itself (an executor shut down) is
+    not a cancel of the task: ``future.result()`` raises it. Never catches
+    ``GeneratorExit``: a closed loop still closes the coroutine.
+    """
+    cancelled = False
+    while not future.done():
+        try:
+            await asyncio.shield(future)
+        except asyncio.CancelledError:
+            if not future.cancelled():
+                cancelled = True
+        except Exception:
+            pass  # the caller reads it from the future
+    return cancelled
+
+
 class _Orphan:
     """A stage with no store entry, in the shape ``_discard_held`` takes."""
 

py/api/thumbnail_auto.py

@@ -88,9 +88,9 @@ from .thumbnail_encode import shutdown_encoders, writer_settings
 from .thumbnail_jobs import ThumbnailOptions
 
 try:
-    from ...utils.parallel import map_parallel
+    from ...utils.parallel import map_parallel, note_refusal, refused_at_shutdown
 except ImportError:  # tests import the package top-level
-    from utils.parallel import map_parallel
+    from utils.parallel import map_parallel, note_refusal, refused_at_shutdown
 
 # The placeholder frame a video without a preview is answered with (JPEG
 # bytes, never written to disk); written previews follow the settings.
@@ -410,7 +410,8 @@ class AutoThumbnailer:
 
     def enqueue(self, src, dst, is_video):
         """Queue *dst*'s preview for the worker; False when already queued,
-        full or closed. Never blocks on a write."""
+        full, closed or its worker is refused at interpreter exit. Never
+        blocks on a write."""
         key = _dst_key(dst)
         with self._changed:
             if self._closed or key in self._queued:
@@ -420,30 +421,41 @@ class AutoThumbnailer:
             self._queued.add(key)
             self._queue.append((key, Path(src), Path(dst), bool(is_video)))
             self._busy = True
+            started = True
             if self._start_worker and not self._worker_running:
-                self._start_worker_locked()
+                started = self._start_worker_locked()
+            if not started:
+                # Refused at interpreter exit: no worker will take it (#375).
+                self._queue.pop()
+                self._queued.discard(key)
             self._changed.notify_all()
-        return True
+        return started
 
     def _start_worker_locked(self):
-        """Start the worker (lock held). A start that fails leaves the file
-        queued and the next enqueue tries again."""
+        """Start the worker (lock held); False only when the interpreter is
+        exiting (#375: logged once at DEBUG, the caller drops the file).
+        Another start failure leaves the file queued and the next enqueue
+        tries again."""
         worker = threading.Thread(target=self._run, name=WORKER_NAME, daemon=True)
         worker.pm_preview_token = _MODULE_TOKEN
         worker.pm_preview_close = self.close
         self._worker_running = True
         try:
             worker.start()
-        except RuntimeError:
+        except RuntimeError as error:
             self._worker_running = False
             self._busy = False
+            if refused_at_shutdown(error):
+                note_refusal(self._logger, "The automatic preview worker")
+                return False
             self._logger.warning(
                 "The automatic preview worker could not start; the next output "
                 "tries again",
                 exc_info=True,
             )
-            return
+            return True
         self._worker = worker
+        return True
 
     def _run(self):
         try:
@@ -790,8 +802,8 @@ class AutoThumbnailMixin:
 
         Returns ``queued``, ``fresh`` (its preview is current) or ``skipped``
         (not a media file under a current root, a link, a preview itself,
-        already queued, or after the app's cleanup). Never writes on the
-        caller's thread.
+        already queued, after the app's cleanup, or when its worker is refused
+        at interpreter exit). Never writes on the caller's thread.
         """
         if self._auto_thumbnails_shut:
             return "skipped"