Keep direct activity calls outside payload transport

Author: andystaples

azure-functions-durable/CHANGELOG.md

@@ -23,6 +23,8 @@ client bindings, and invocation logging context is preserved.
 Worker payload-size and storage errors retain their original error details.
 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.
 
 ## v2.0.0rc1
 

azure-functions-durable/azure/durable_functions/internal/compat/activity.py

@@ -40,7 +40,6 @@
     deexternalize_payload_async,
     externalize_activity_output,
     externalize_activity_output_async,
-    get_payload_store,
 )
 from .orchestration_context import accepts_two_positional_args
 
@@ -188,17 +187,19 @@ def decode(value: str) -> Any:
             return df_loads(value)
         except json.JSONDecodeError:
             return value
+        except Exception as error:
+            raise ValueError('activity trigger input must be a string or a '
+                             f'valid json serializable ({value})') from error
 
     wrapper: Callable[..., Any]
     if inspect.iscoroutinefunction(fn):
         @wraps(fn)
         async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
-            if get_payload_store() is None:
-                return await fn(*args, **kwargs)
             bound = signature.bind(*args, **kwargs)
             value = bound.arguments.get(input_name)
-            if isinstance(value, ActivityPayload):
-                bound.arguments[input_name] = decode(await deexternalize_payload_async(value.value))
+            if not isinstance(value, ActivityPayload):
+                return await fn(*args, **kwargs)
+            bound.arguments[input_name] = decode(await deexternalize_payload_async(value.value))
             result = await fn(*bound.args, **bound.kwargs)
             if result is None:
                 return None
@@ -209,12 +210,11 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
     else:
         @wraps(fn)
         def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
-            if get_payload_store() is None:
-                return fn(*args, **kwargs)
             bound = signature.bind(*args, **kwargs)
             value = bound.arguments.get(input_name)
-            if isinstance(value, ActivityPayload):
-                bound.arguments[input_name] = decode(deexternalize_payload(value.value))
+            if not isinstance(value, ActivityPayload):
+                return fn(*args, **kwargs)
+            bound.arguments[input_name] = decode(deexternalize_payload(value.value))
             result = fn(*bound.args, **bound.kwargs)
             if result is None:
                 return None

azure-functions-durable/azure/durable_functions/internal/converters.py

@@ -139,7 +139,7 @@ def decode(cls, data: meta.Datum, *,
         if data_type in ['string', 'json']:
             value = data.value
             store = get_transport_payload_store()
-            if store is not None and store.is_known_token(value):
+            if store is not None:
                 return ActivityPayload(value)
             try:
                 result = df_loads(value)

tests/azure-functions-durable/test_activity_adapter_compat.py

@@ -205,6 +205,29 @@ async def native_activity(context, payload):
     store.upload_async.assert_awaited_once()
 
 
+@pytest.mark.parametrize("user_async", [False, True])
+@pytest.mark.parametrize("value", [None, "small", {"small": True}])
+async def test_host_activity_externalizes_large_output_from_inline_input(
+        monkeypatch, payload_store_factory, user_async, value):
+    store = payload_store_factory()
+    monkeypatch.setattr(payloads, "_payload_store", store)
+    output = {"large": "x" * 200}
+
+    def activity(payload):
+        assert payload == value
+        return output
+
+    async def async_activity(payload):
+        return activity(payload)
+
+    wrapper = wrap_activity_payloads(async_activity if user_async else activity, "payload")
+    decoded = ActivityTriggerConverter.decode(
+        meta.Datum(type="json", value=json.dumps(value)), trigger_metadata=None)
+    result = await wrapper(decoded) if user_async else wrapper(decoded)
+    encoded = ActivityTriggerConverter.encode(result, expected_type=None)
+    assert json.loads(store.download(json.loads(encoded.value))) == output
+
+
 def test_sync_activity_uses_sync_storage_on_the_calling_thread(monkeypatch, payload_store_factory):
     store = payload_store_factory()
     monkeypatch.setattr(payloads, "_payload_store", store)

tests/azure-functions-durable/test_converters.py

@@ -36,7 +36,9 @@
 
 
 def _encode_activity(value):
-    result = wrap_activity_payloads(lambda payload: payload, "payload")(value)
+    host_input = ActivityTriggerConverter.decode(
+        meta.Datum(type="json", value="null"), trigger_metadata=None)
+    result = wrap_activity_payloads(lambda payload: value, "payload")(host_input)
     return ActivityTriggerConverter.encode(result, expected_type=None)
 
 

tests/azure-functions-durable/test_decorator_compat.py

@@ -9,7 +9,6 @@
 import pytest
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 from azure.durable_functions.internal import payloads
-from azure.durable_functions.internal.converters import ActivityTriggerConverter
 from azure.durable_functions.constants import (
     ACTIVITY_TRIGGER,
     DURABLE_CLIENT,
@@ -95,8 +94,8 @@ def my_activity(ctx, payload):
 
 @pytest.mark.parametrize("configured", [False, True])
 @pytest.mark.parametrize("user_async", [False, True])
-async def test_activity_trigger_preserves_direct_calls_and_skips_inline_storage(
-        monkeypatch, payload_store_factory, configured, user_async):
+@pytest.mark.parametrize("value", [None, "hello", {"large": "x" * 200}])
+async def test_direct_activity_calls_skip_storage(monkeypatch, payload_store_factory, configured, user_async, value):
     monkeypatch.setattr(payloads, "_payload_store", None)
     app = df.DFApp()
 
@@ -117,12 +116,9 @@ async def async_activity(payload):
     function = registered.build().get_user_function()
     assert inspect.iscoroutinefunction(function) == user_async
     assert list(inspect.signature(function).parameters) == ["payload"]
-    result = await function("hello") if user_async else function("hello")
+    result = await registered(value) if user_async else registered(value)
     assert not inspect.isawaitable(result)
-    if configured:
-        assert ActivityTriggerConverter.encode(result, expected_type=None).value == '"hello"'
-    else:
-        assert result == "hello"
+    assert result is value
     store.download.assert_not_called()
     store.upload.assert_not_called()
     store.download_async.assert_not_called()