fix(client): non-blocking fingerprint + immediate stop_event cancel (async)

resumable_upload/client/aio/client.py

@@ -275,7 +275,10 @@ async def upload_file(
             metadata["filename"] = os.path.basename(file_path)
 
         fingerprint = (
-            self.fingerprinter.get_fingerprint(file_path or file_stream)  # type: ignore[arg-type]  # ty: ignore[invalid-argument-type]
+            await asyncio.to_thread(
+                self.fingerprinter.get_fingerprint,
+                file_path or file_stream,  # type: ignore[arg-type]  # ty: ignore[invalid-argument-type]
+            )
             if self.store_url
             else None
         )

resumable_upload/client/aio/uploader.py

@@ -274,6 +274,9 @@ async def upload_chunk(self) -> bool:
         Raises:
             TusUploadFailed: If upload fails.
         """
+        if self._stop_event.is_set():
+            raise TusUploadFailed("Upload cancelled via stop_event")
+
         if self.offset >= self.file_size:
             return False
 
@@ -318,6 +321,9 @@ async def upload(
         max_offset = min(stop_at, self.file_size) if stop_at is not None else self.file_size
 
         while self.offset < max_offset:
+            if self._stop_event.is_set():
+                raise TusUploadFailed("Upload cancelled via stop_event")
+
             chunk_size = min(self.chunk_size, max_offset - self.offset)
             chunk = await asyncio.to_thread(self._read_at, self.offset, chunk_size)
 

resumable_upload/server/core.py

@@ -131,6 +131,10 @@ def __init__(
         self.request_timeout = request_timeout
         self._last_cleanup: Optional[datetime] = None
         self._cleanup_lock = threading.Lock()
+        # Async path uses a non-blocking flag instead of the lock above: holding
+        # a threading.Lock across an await would block the event loop (deadlock
+        # when cleanup_interval <= 0). Single event loop => a plain flag suffices.
+        self._cleanup_running = False
         self._on_incoming_request = on_incoming_request
         self._on_upload_create = on_upload_create
         self._on_upload_complete = on_upload_complete
@@ -471,19 +475,18 @@ async def handle_request_async(
 
         if self.upload_expiry is not None:
             now = datetime.now(timezone.utc)
-            if (
+            if not self._cleanup_running and (
                 self._last_cleanup is None
                 or (now - self._last_cleanup).total_seconds() >= self.cleanup_interval
             ):
-                with self._cleanup_lock:
-                    if (
-                        self._last_cleanup is None
-                        or (now - self._last_cleanup).total_seconds() >= self.cleanup_interval
-                    ):
-                        self._last_cleanup = now
-                        count = await self.storage.cleanup_expired_uploads_async()
-                        if count:
-                            logger.info("Cleaned up %s expired upload(s)", count)
+                self._cleanup_running = True
+                self._last_cleanup = now
+                try:
+                    count = await self.storage.cleanup_expired_uploads_async()
+                    if count:
+                        logger.info("Cleaned up %s expired upload(s)", count)
+                finally:
+                    self._cleanup_running = False
 
         if self._metrics is not None and status >= 400:
             self._metrics.inc("tusd_errors_total", labels={"status": str(status)})

tests/test_async_client.py

@@ -70,6 +70,87 @@ async def test_async_uploader_uploads_in_chunks(asgi_base):
         assert head.headers["Upload-Offset"] == str(len(payload))
 
 
+@pytest.mark.anyio
+async def test_async_uploader_stop_event_cancels_without_retries(asgi_base):
+    """stop_event must cancel even when chunks succeed and retries are off."""
+    import asyncio
+    import io
+    import os
+
+    from resumable_upload.client.aio.uploader import AsyncUploader
+    from resumable_upload.exceptions import TusUploadFailed
+
+    transport, base = asgi_base
+    payload = os.urandom(50_000)
+    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as c:
+        r = await c.request(
+            "POST", base, headers={"Tus-Resumable": "1.0.0", "Upload-Length": str(len(payload))}
+        )
+        url = r.headers["Location"]
+        stop = asyncio.Event()
+        stop.set()
+        up = await AsyncUploader.open(
+            c, url, file_stream=io.BytesIO(payload), chunk_size=16_384, stop_event=stop
+        )
+        with pytest.raises(TusUploadFailed, match="cancelled via stop_event"):
+            await up.upload()
+
+
+@pytest.mark.anyio
+async def test_async_uploader_chunk_stop_event(asgi_base):
+    """upload_chunk() honors stop_event before sending a chunk."""
+    import asyncio
+    import io
+    import os
+
+    from resumable_upload.client.aio.uploader import AsyncUploader
+    from resumable_upload.exceptions import TusUploadFailed
+
+    transport, base = asgi_base
+    payload = os.urandom(20_000)
+    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as c:
+        r = await c.request(
+            "POST", base, headers={"Tus-Resumable": "1.0.0", "Upload-Length": str(len(payload))}
+        )
+        url = r.headers["Location"]
+        stop = asyncio.Event()
+        up = await AsyncUploader.open(
+            c, url, file_stream=io.BytesIO(payload), chunk_size=8192, stop_event=stop
+        )
+        # First chunk succeeds; then cancel and the next upload_chunk must refuse.
+        assert await up.upload_chunk() is True
+        stop.set()
+        with pytest.raises(TusUploadFailed, match="cancelled via stop_event"):
+            await up.upload_chunk()
+        # Offset did not advance past the one successful chunk (append-only preserved).
+        head = await c.request("HEAD", url, headers={"Tus-Resumable": "1.0.0"})
+        assert head.headers["Upload-Offset"] == "8192"
+
+
+@pytest.mark.anyio
+async def test_async_client_store_url_resumes(asgi_base, tmp_path):
+    """store_url path computes the fingerprint and reuses the stored URL."""
+    import os
+
+    from resumable_upload.client.aio.client import AsyncTusClient
+    from resumable_upload.url_storage import FileURLStorage
+
+    transport, base = asgi_base
+    f = tmp_path / "data.bin"
+    f.write_bytes(os.urandom(40_000))
+    async with AsyncTusClient(
+        base,
+        _transport=transport,
+        chunk_size=8192,
+        store_url=True,
+        url_storage=FileURLStorage(str(tmp_path / "urls.json")),
+    ) as client:
+        url1 = await client.upload_file(str(f))
+        # Second call for the same file returns the same stored URL (fingerprint hit).
+        url2 = await client.upload_file(str(f))
+        assert url1 == url2
+
+
 @pytest.mark.anyio
 async def test_async_uploader_checksum_roundtrip(asgi_base):
     import io

tests/test_server_async_dispatch.py

@@ -273,3 +273,45 @@ def test_head_non_partial_omits_upload_concat(server, dispatch):
     status, headers, _ = dispatch(server, "HEAD", location, _h(), b"")
     assert status == 200
     assert "Upload-Concat" not in headers
+
+
+def test_async_cleanup_no_deadlock_with_zero_interval(tmp_path):
+    """Regression: concurrent async requests must not deadlock during cleanup.
+
+    With ``cleanup_interval <= 0`` the double-checked guard always passes, so
+    the old ``with self._cleanup_lock:`` (a threading.Lock held across an
+    ``await``) let a second coroutine block the event loop forever. The
+    non-blocking ``_cleanup_running`` flag must keep this lock-free.
+
+    Run the loop in a worker thread and join with a timeout: a true deadlock
+    freezes the loop thread, so an in-loop ``asyncio.wait_for`` could never
+    fire — only an outside thread can observe the hang.
+    """
+    import threading
+
+    srv = TusServer(
+        storage=SQLiteStorage(db_path=str(tmp_path / "u.db"), upload_dir=str(tmp_path / "f")),
+        base_path="/files",
+        upload_expiry=3600,
+        cleanup_interval=0,
+    )
+
+    results: list[tuple[int, dict[str, str], bytes]] = []
+
+    def run() -> None:
+        async def both() -> None:
+            # Two concurrent OPTIONS; each triggers end-of-dispatch cleanup.
+            results.extend(
+                await asyncio.gather(
+                    srv.handle_request_async("OPTIONS", "/files", _h(), b""),
+                    srv.handle_request_async("OPTIONS", "/files", _h(), b""),
+                )
+            )
+
+        asyncio.run(both())
+
+    t = threading.Thread(target=run, daemon=True)
+    t.start()
+    t.join(timeout=5)
+    assert not t.is_alive(), "async cleanup deadlocked (threading.Lock held across await)"
+    assert all(status == 204 for status, _, _ in results)