From a6637d9e78bd1c4ffed1556077c563d7a92fcfac Mon Sep 17 00:00:00 2001 From: Injae Ryou Date: Mon, 29 Jun 2026 21:27:52 +0900 Subject: [PATCH 1/3] fix(client): non-blocking fingerprint + immediate stop_event cancel (async) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address gemini-code-assist review on #1: - Run the SHA-256 fingerprint via asyncio.to_thread so a large file's hashing no longer blocks the event loop (only computed when store_url is set). - Check stop_event at the top of AsyncUploader.upload_chunk and in the upload loop, so cancellation works even with retries disabled (max_retries=0) or while chunks keep succeeding — previously it was only honored during the retry backoff wait. Client-side only; no wire/header/offset behavior changes. Co-Authored-By: Claude Opus 4.8 (1M context) --- resumable_upload/client/aio/client.py | 5 ++++- resumable_upload/client/aio/uploader.py | 6 ++++++ tests/test_async_client.py | 26 +++++++++++++++++++++++++ 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/resumable_upload/client/aio/client.py b/resumable_upload/client/aio/client.py index e0aea99..ba7a55c 100644 --- a/resumable_upload/client/aio/client.py +++ b/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 ) diff --git a/resumable_upload/client/aio/uploader.py b/resumable_upload/client/aio/uploader.py index 07e2f36..009393a 100644 --- a/resumable_upload/client/aio/uploader.py +++ b/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) diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 08debee..56a8893 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -70,6 +70,32 @@ 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_checksum_roundtrip(asgi_base): import io From 17f0fba149ee8a9a74c83369838016b0a1341b47 Mon Sep 17 00:00:00 2001 From: Injae Ryou Date: Mon, 29 Jun 2026 21:33:40 +0900 Subject: [PATCH 2/3] test(client): cover async upload_chunk stop_event + store_url fingerprint Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_async_client.py | 55 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 56a8893..4ca3551 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -96,6 +96,61 @@ async def test_async_uploader_stop_event_cancels_without_retries(asgi_base): 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 From c52227197afb9acb806ad55521ed70dd49b15fc6 Mon Sep 17 00:00:00 2001 From: Injae Ryou Date: Mon, 29 Jun 2026 21:53:01 +0900 Subject: [PATCH 3/3] fix(server): non-blocking async cleanup guard (no lock across await) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The async dispatch held a threading.Lock across `await cleanup_expired_uploads_async()`. With `cleanup_interval <= 0` the double-checked guard always passes, so a second concurrent request would call `acquire()` and block the event loop thread — a hard deadlock, since the awaiting coroutine can never resume to release it. Replace the lock with a non-blocking `_cleanup_running` flag on the async path only; the sync path keeps the lock (no await, lock is correct there). Single event loop => a plain flag is enough to prevent overlapping runs. Regression test runs the loop in a worker thread and joins with a timeout, since a frozen loop can't honor an in-loop asyncio.wait_for. Addresses gemini-code-assist critical comment on #1. Co-Authored-By: Claude Opus 4.8 (1M context) --- resumable_upload/server/core.py | 23 +++++++++------- tests/test_server_async_dispatch.py | 42 +++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 10 deletions(-) diff --git a/resumable_upload/server/core.py b/resumable_upload/server/core.py index e0b7e0d..e070f4b 100644 --- a/resumable_upload/server/core.py +++ b/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)}) diff --git a/tests/test_server_async_dispatch.py b/tests/test_server_async_dispatch.py index 09f54d1..756fbc2 100644 --- a/tests/test_server_async_dispatch.py +++ b/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)