Offload gzip work in asynchronous Blob payload transfers

Author: andystaples

CHANGELOG.md

@@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+FIXED
+
+- Asynchronous Azure Blob payload uploads and downloads no longer run gzip
+compression or decompression on the calling event loop, keeping concurrent
+async operations responsive during large transfers.
+
 ## v1.10.1
 
 FIXED

azure-functions-durable/CHANGELOG.md

@@ -26,6 +26,8 @@ Activity wrappers preserve the application's source directory for
 `context.function_directory`, including when payload storage is not configured.
 Direct activity calls retain their Python return values and do not access payload
 storage; externalization applies only to host binding invocations.
+With the corresponding core SDK update, asynchronous Blob payload transfers
+also keep compression and decompression off the Functions invocation event loop.
 
 ## v2.0.0rc1
 

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+FIXED
+
+- With the corresponding core SDK update, asynchronous Azure Blob payload
+transfers no longer block the event loop during compression or decompression.
+
 ## v1.10.1
 
 CHANGED

durabletask/extensions/azure_blob_payloads/blob_payload_store.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import gzip
 import logging
 import uuid
@@ -173,7 +174,7 @@ async def upload_async(self, data: bytes, *, instance_id: str | None = None) ->
         await self._ensure_container_async()
 
         if self._options.enable_compression:
-            data = gzip.compress(data)
+            data = await asyncio.to_thread(gzip.compress, data)
 
         blob_name = self._make_blob_name(instance_id)
         container_client: AsyncContainerClient = self._async_blob_service_client.get_container_client(
@@ -194,7 +195,7 @@ async def download_async(self, token: str) -> bytes:
         blob_data = await stream.readall()
 
         if self._options.enable_compression:
-            blob_data = gzip.decompress(blob_data)
+            blob_data = await asyncio.to_thread(gzip.decompress, blob_data)
 
         logger.debug("Downloaded %d bytes <- %s", len(blob_data), token)
         return blob_data

tests/durabletask/test_large_payload.py

@@ -3,7 +3,10 @@
 
 """Tests for large-payload externalization and de-externalization."""
 
-from unittest.mock import MagicMock
+import asyncio
+import gzip
+import threading
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from google.protobuf import wrappers_pb2
@@ -433,6 +436,72 @@ def test_is_known_token(self):
 # ------------------------------------------------------------------
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["upload", "download"])
+@pytest.mark.parametrize("enable_compression", [False, True])
+async def test_async_blob_compression_yields(monkeypatch, operation, enable_compression):
+    pytest.importorskip("azure.storage.blob")
+    from durabletask.extensions.azure_blob_payloads import blob_payload_store as module
+    from durabletask.extensions.azure_blob_payloads import BlobPayloadStoreOptions
+
+    data = b"payload" * 1000
+    stored_data = gzip.compress(data) if enable_compression else data
+    loop = asyncio.get_running_loop()
+    loop_thread = threading.get_ident()
+    entered = asyncio.Event()
+    release = threading.Event()
+    gzip_method = "compress" if operation == "upload" else "decompress"
+    original = getattr(gzip, gzip_method)
+
+    def gated_gzip(value):
+        assert enable_compression
+        assert threading.get_ident() != loop_thread
+        loop.call_soon_threadsafe(entered.set)
+        assert release.wait(5), "Event loop did not release compression"
+        return original(value)
+
+    async def upload_blob(**kwargs):
+        assert threading.get_ident() == loop_thread
+
+    async def readall():
+        assert threading.get_ident() == loop_thread
+        return stored_data
+
+    stream = MagicMock()
+    stream.readall = AsyncMock(side_effect=readall)
+    container = MagicMock()
+    container.create_container = AsyncMock()
+    container.upload_blob = AsyncMock(side_effect=upload_blob)
+    container.download_blob = AsyncMock(return_value=stream)
+    service = MagicMock()
+    service.get_container_client.return_value = container
+    service.close = AsyncMock()
+    monkeypatch.setattr(module, "BlobServiceClient", MagicMock())
+    monkeypatch.setattr(module, "AsyncBlobServiceClient", MagicMock(return_value=service))
+    monkeypatch.setattr(gzip, gzip_method, gated_gzip)
+    store = module.BlobPayloadStore(BlobPayloadStoreOptions(
+        account_url="https://example.blob.core.windows.net", enable_compression=enable_compression))
+    transfer = asyncio.create_task(store.upload_async(data) if operation == "upload"
+                                   else store.download_async("blob:v1:durabletask-payloads:example"))
+    try:
+        if enable_compression:
+            await asyncio.wait_for(entered.wait(), timeout=2)
+            assert not transfer.done()
+    finally:
+        release.set()
+        try:
+            result = await transfer
+        finally:
+            store.close()
+            await store.close_async()
+    if operation == "upload":
+        uploaded = container.upload_blob.call_args.kwargs["data"]
+        assert (gzip.decompress(uploaded) if enable_compression else uploaded) == data
+        assert store.is_known_token(result)
+    else:
+        assert result == data
+
+
 class TestBlobPayloadStoreDefaults:
     def test_default_options(self):
         """Constructing with connection_string should use .NET SDK defaults."""