Skip to content

Commit ba15b23

Browse files
andystaplesCopilot
andauthored
Add configurable large payload storage to the Azure Functions provider (#270)
* Add configurable large payload storage to Azure Functions provider * Make Functions payload tests independently collectable * Document and test reserved payload reference strings * Address Functions large-payload review feedback * Await Functions payload storage across durable triggers * Preserve synchronous activities and narrow replay payload hydration Keep orchestration and entity payload I/O asynchronous while restoring native activity calling conventions. Skip unused historical entity input downloads during replay and retain full client history hydration. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> * Preserve activity source directories through Functions wrappers * Keep direct activity calls outside payload transport * Skip unused scheduled activity inputs during Functions replay * Offload gzip work in asynchronous Blob payload transfers * Document payload reference trust and invocation boundaries * Keep Functions changelog focused on user-facing changes --------- Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
1 parent a6c18f2 commit ba15b23

25 files changed

Lines changed: 2018 additions & 38 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
77

88
## Unreleased
99

10+
FIXED
11+
12+
- Asynchronous Azure Blob payload uploads and downloads no longer run gzip
13+
compression or decompression on the calling event loop, keeping concurrent
14+
async operations responsive during large transfers.
15+
1016
## v1.10.1
1117

1218
FIXED

‎azure-functions-durable/CHANGELOG.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
77

88
## Unreleased
99

10+
ADDED
11+
12+
- Added `DFApp.configure_large_payloads(payload_store=...)` to externalize large
13+
durable payloads to Azure Blob Storage or a custom payload store.
14+
- Configured clients automatically hydrate stored payloads, including
15+
orchestration history and entity operation inputs and results.
16+
17+
FIXED
18+
19+
- Preserved the application's source directory in `context.function_directory`
20+
for decorated activities and durable-client functions.
21+
1022
## v2.0.0rc1
1123

1224
CHANGED

‎azure-functions-durable/README.md‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,129 @@ Key capabilities include durable orchestrations and sub-orchestrations, durable
3232
timers, external events, durable entities, retries, versioning, durable HTTP
3333
calls (`context.call_http(...)`), recurring scheduled tasks, and history export.
3434

35+
## Large payloads
36+
37+
Configure a `durabletask.payload.PayloadStore` once at app startup to store large
38+
serialized payloads outside orchestration history. For Azure Blob Storage, install
39+
the optional dependencies:
40+
41+
```bash
42+
pip install azure-functions-durable "durabletask[azure-blob-payloads]" aiohttp
43+
```
44+
45+
In your Function app, configure the root `DFApp` before any invocations:
46+
47+
```python
48+
import os
49+
50+
import azure.durable_functions as df
51+
from durabletask.extensions.azure_blob_payloads import (
52+
BlobPayloadStore,
53+
BlobPayloadStoreOptions,
54+
)
55+
56+
app = df.DFApp()
57+
app.configure_large_payloads(
58+
payload_store=BlobPayloadStore(BlobPayloadStoreOptions(
59+
connection_string=os.environ["PAYLOAD_STORAGE_CONNECTION_STRING"],
60+
container_name="durable-payloads",
61+
threshold_bytes=256 * 1024,
62+
))
63+
)
64+
```
65+
66+
Set `PAYLOAD_STORAGE_CONNECTION_STRING` in your Function app settings (or in
67+
`local.settings.json` for local development). The store automatically uploads
68+
serialized payloads above the threshold and downloads their contents when the
69+
SDK consumes them. Orchestration and activity inputs and outputs, custom status,
70+
external events, and entity inputs, results, and state use the configured store.
71+
Sub-orchestrations and continue-as-new use it as well. The default maximum stored
72+
payload size is 10 MiB; `max_stored_payload_bytes` can configure this limit.
73+
74+
Configuration applies to both synchronous and asynchronous durable clients and
75+
all registered blueprints, including blueprints imported before configuration.
76+
There is one store per Python worker process. Registering the same store object
77+
again is allowed; registering a different object raises `ValueError`. Configure
78+
every scaled-out worker with access to the same backing storage and retain that
79+
access across deployments. Keep the store open for the process lifetime.
80+
81+
> [!WARNING]
82+
> Keep payload blobs for as long as any retained orchestration history or entity
83+
> state references them, including histories needed for replay. Purging an
84+
> orchestration does not delete its payload blobs; manage retention separately.
85+
86+
This is SDK-managed storage, separate from the Azure Storage backend's automatic
87+
large-message handling. Without configuration, the SDK keeps payloads inline.
88+
Use the configured Python clients to retrieve hydrated payloads. Host management
89+
HTTP endpoints and other consumers that do not use this configuration can expose
90+
reference strings instead. Applications exchanging externalized payloads must
91+
agree on the store and reference encoding; Functions references are JSON strings.
92+
93+
> [!WARNING]
94+
> Storage failures can fail durable invocations, including orchestrations.
95+
> Storage transport retries are separate from durable activity retry policies.
96+
> This SDK does not add an activity retry policy or guarantee that the Functions
97+
> host abandons and redelivers a work item after a storage failure. A transient
98+
> storage error can therefore become a terminal orchestration failure.
99+
100+
Registered orchestration and entity handlers await the store's async methods
101+
before and after execution. Orchestrators remain synchronous generators, and
102+
orchestration replay and entity code run on execution threads with their
103+
invocation logging context preserved. Their payload downloads and uploads do
104+
not occupy those threads, and serialization does not access storage during
105+
replay. Custom stores must implement genuinely nonblocking async methods to
106+
benefit from this behavior.
107+
108+
For orchestration and entity execution, the SDK reuses the Functions runtime's
109+
thread pool when the runtime exposes it; otherwise it uses a process-wide SDK
110+
pool. Both honor `PYTHON_THREADPOOL_THREAD_COUNT`.
111+
112+
Activities retain their synchronous or asynchronous calling convention.
113+
Synchronous activities use synchronous storage inside the host-managed execution
114+
thread and remain directly callable without `await`; async activities await
115+
async storage. Direct calls to decorated activities return ordinary Python values
116+
without accessing payload storage; transport processing applies only to host
117+
binding invocations. Binding converters perform no storage I/O. Synchronous functions
118+
still receive the synchronous durable client, and synchronous client APIs use
119+
synchronous storage. Direct `Orchestrator.handle()` and `Orchestrator.create()`
120+
adapters also remain synchronous.
121+
122+
Both client history APIs hydrate entity operation inputs and results, including
123+
values nested in the host's entity protocol envelopes. During orchestration
124+
replay, nested entity results are hydrated, but historical nested request inputs
125+
are not downloaded because replay only needs their correlation metadata.
126+
Historical scheduled activity inputs are also not downloaded during replay;
127+
explicit history retrieval continues to hydrate those inputs.
128+
129+
> [!WARNING]
130+
> With payload storage configured, whole payload strings recognized by the
131+
> store's `is_known_token()` are reserved references, not literal application
132+
> data. For `BlobPayloadStore`, this includes strings of the form
133+
> `blob:v1:<container>:<blobName>`, with nonempty container and blob names.
134+
> Functions recognizes both raw and JSON-quoted references. A matching string
135+
> is treated as already externalized on output and downloaded on input, even
136+
> below the size threshold. Missing or inaccessible references raise errors;
137+
> they do not fall back to literal strings.
138+
139+
To pass a reference as application data for later retrieval, wrap it in an
140+
object, for example `{"reference": "blob:v1:container:blob"}`. Reference detection
141+
does not recursively inspect strings inside application JSON objects. The
142+
wrapper preserves the literal reference whether the object stays inline or is
143+
itself externalized. Keep the wrapper whenever passing that value across a
144+
durable payload boundary; passing its string field alone opts back into reference
145+
interpretation. Custom payload stores define their own reserved token syntax.
146+
147+
> [!WARNING]
148+
> Recognized references are trusted transport inputs, not authorization
149+
> boundaries. `BlobPayloadStore` reads from the container named in the token
150+
> using its configured credentials; `container_name` selects the upload
151+
> container and does not restrict downloads. An account-wide connection string
152+
> can therefore allow reads outside that container. Use least-privilege
153+
> credentials scoped to the intended payload storage, and explicitly decide
154+
> whether external callers may supply references. Reject untrusted references
155+
> or validate their allowed storage locations before passing them into durable
156+
> APIs; token recognition alone does not authorize a read.
157+
35158
## Unit testing entities
36159

37160
Use `execute_entity()` to run one entity operation in-process without a

‎azure-functions-durable/azure/durable_functions/client.py‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,12 @@
88
import threading
99

1010
from datetime import datetime, timedelta
11-
from typing import Any, Mapping, Optional, Union, cast
11+
from typing import Any, Mapping, Optional, Union, cast, override
1212
from warnings import deprecated
1313
import azure.functions as func
1414
from urllib.parse import urlparse, quote
1515

16+
from durabletask import history
1617
from durabletask.client import (
1718
AsyncTaskHubGrpcClient,
1819
OrchestrationQuery,
@@ -26,6 +27,11 @@
2627
AzureFunctionsDefaultClientInterceptorImpl,
2728
)
2829
from .internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER
30+
from .internal.payloads import (
31+
get_transport_payload_store,
32+
hydrate_entity_history,
33+
hydrate_entity_history_async,
34+
)
2935
from .http.http_management_payload import HttpManagementPayload, replace_url_origin
3036
from .internal.compat.durable_orchestration_status import DurableOrchestrationStatus
3137
from .internal.compat.entity_state_response import EntityStateResponse
@@ -178,6 +184,7 @@ def __init__(self, client_as_string: str):
178184
interceptors=interceptors,
179185
channel_options=channel_options,
180186
data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER,
187+
payload_store=get_transport_payload_store(),
181188
emit_trace_spans=False,
182189
logger=_LOGGER)
183190

@@ -193,6 +200,16 @@ def __init__(self, client_as_string: str):
193200
self._creation_loop = None
194201
self._close_scheduled = False
195202

203+
@override
204+
async def get_orchestration_history(
205+
self, instance_id: str, *, execution_id: str | None = None,
206+
for_work_item_processing: bool = False) -> list[history.HistoryEvent]:
207+
events = await super().get_orchestration_history(
208+
instance_id, execution_id=execution_id,
209+
for_work_item_processing=for_work_item_processing)
210+
await hydrate_entity_history_async(events, self._payload_store, instance_id)
211+
return events
212+
196213
def schedule_close(self) -> None:
197214
"""Schedule the underlying gRPC channel to close after the invocation.
198215
@@ -664,9 +681,20 @@ def __init__(self, client_as_string: str):
664681
interceptors=interceptors,
665682
channel_options=channel_options,
666683
data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER,
684+
payload_store=get_transport_payload_store(),
667685
emit_trace_spans=False,
668686
logger=_LOGGER)
669687

688+
@override
689+
def get_orchestration_history(
690+
self, instance_id: str, *, execution_id: str | None = None,
691+
for_work_item_processing: bool = False) -> list[history.HistoryEvent]:
692+
events = super().get_orchestration_history(
693+
instance_id, execution_id=execution_id,
694+
for_work_item_processing=for_work_item_processing)
695+
hydrate_entity_history(events, self._payload_store, instance_id)
696+
return events
697+
670698
@classmethod
671699
def get_cached(cls, client_as_string: str) -> "SyncDurableFunctionsClient":
672700
"""Get the process-wide client for a durable-client binding configuration.

‎azure-functions-durable/azure/durable_functions/decorators/durable_app.py‎

Lines changed: 42 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
from azure.functions.decorators.function_app import DecoratorApi, FunctionBuilder
1111

1212
from durabletask import task
13+
from durabletask.payload import PayloadStore
1314

1415
from .metadata import OrchestrationTrigger, ActivityTrigger, EntityTrigger, \
1516
DurableClient
@@ -19,9 +20,9 @@
1920
builtin_http_activity,
2021
builtin_http_poll_orchestrator,
2122
)
22-
from ..internal.compat.activity import wrap_activity
23+
from ..internal.compat.activity import wrap_activity, wrap_activity_payloads
24+
from ..internal.invocation import preserve_function_source, wrap_invocation
2325
from ..worker import DurableFunctionsWorker
24-
from ..orchestrator import Orchestrator
2526

2627

2728
class Blueprint(TriggerApi, BindingApi):
@@ -166,6 +167,7 @@ def configure_history_export(self, writer: Any) -> None:
166167
def _configure_orchestrator_callable(
167168
self,
168169
wrap: Callable[[Callable[..., Any]], FunctionBuilder],
170+
context_name: str,
169171
input_type: Optional[type] = None
170172
) -> Callable[[task.Orchestrator[Any, Any]], FunctionBuilder]:
171173
"""Obtain decorator to construct an Orchestrator class from a user-defined Function.
@@ -193,7 +195,13 @@ def decorator(orchestrator_func: task.Orchestrator[Any, Any]) -> FunctionBuilder
193195
# feed it to a v1-style ``context.get_input()``.
194196
orchestrator_func._df_input_type = input_type # type: ignore[attr-defined] # noqa: E501
195197

196-
handle = Orchestrator.create(orchestrator_func)
198+
worker = DurableFunctionsWorker()
199+
200+
async def handle(context: func.OrchestrationContext) -> str:
201+
return await worker.execute_orchestration_request_async(orchestrator_func, context)
202+
203+
handle.orchestrator_function = orchestrator_func # pyright: ignore[reportFunctionMemberAccess]
204+
handle = wrap_invocation(handle, "context", registered_trigger_name=context_name)
197205

198206
# invoke next decorator, with the Orchestrator as input
199207
handle.__name__ = orchestrator_func.__name__
@@ -204,6 +212,7 @@ def decorator(orchestrator_func: task.Orchestrator[Any, Any]) -> FunctionBuilder
204212
def _configure_entity_callable(
205213
self,
206214
wrap: Callable[[Callable[..., Any]], FunctionBuilder],
215+
context_name: str,
207216
entity_name: Optional[str] = None
208217
) -> Callable[[task.Entity[Any, Any]], FunctionBuilder]:
209218
"""Obtain decorator to construct an Entity class from a user-defined Function.
@@ -235,18 +244,16 @@ def decorator(entity_func: task.Entity[Any, Any]) -> FunctionBuilder:
235244
# Construct an orchestrator based on the end-user code
236245
worker = DurableFunctionsWorker()
237246

238-
# TODO: Because this handle method is the one actually exposed to the Functions SDK decorator,
239-
# the parameter name will always be "context" here, even if the user specified a different name.
240-
# We need to find a way to allow custom context names (like "ctx").
241247
# The generated handle is what the Azure Functions host registers,
242248
# so its ``context`` parameter must be annotated with
243249
# ``azure.functions.EntityContext`` for the host's entityTrigger
244250
# binding converter to accept it; at runtime the host passes that
245251
# transport context (exposing ``.body``).
246-
def handle(context: func.EntityContext) -> str:
247-
return worker.execute_entity_batch_request(entity_func, context)
252+
async def handle(context: func.EntityContext) -> str:
253+
return await worker.execute_entity_batch_request_async(entity_func, context)
248254

249255
handle.entity_function = entity_func # pyright: ignore[reportFunctionMemberAccess]
256+
handle = wrap_invocation(handle, "context", registered_trigger_name=context_name)
250257

251258
# invoke next decorator, with the Entity as input
252259
handle.__name__ = entity_func.__name__
@@ -294,14 +301,15 @@ def orchestration_trigger(self, context_name: str,
294301
def wrap(fb: FunctionBuilder) -> FunctionBuilder:
295302

296303
def decorator() -> FunctionBuilder:
304+
registered = fb._function._func # pyright: ignore[reportPrivateUsage]
297305
fb.add_trigger(
298-
trigger=OrchestrationTrigger(name=context_name,
306+
trigger=OrchestrationTrigger(name=getattr(registered, "_df_trigger_name", context_name),
299307
orchestration=orchestration))
300308
return fb
301309

302310
return decorator()
303311

304-
return self._configure_orchestrator_callable(wrap, input_type=input_type)
312+
return self._configure_orchestrator_callable(wrap, context_name, input_type=input_type)
305313

306314
def activity_trigger(self, input_name: str,
307315
activity: Optional[str] = None
@@ -326,7 +334,13 @@ def decorator(user_fn: Callable[..., Any]) -> FunctionBuilder:
326334
# Adapt a durabletask-native two-argument activity ((ctx, input))
327335
# to the host's single-input convention; one-argument activities
328336
# pass through unchanged.
329-
return wrap(wrap_activity(user_fn, input_name))
337+
function = (user_fn._function._func # pyright: ignore[reportPrivateUsage]
338+
if isinstance(user_fn, FunctionBuilder) else user_fn)
339+
registered = wrap_activity_payloads(wrap_activity(function, input_name), input_name)
340+
if isinstance(user_fn, FunctionBuilder):
341+
user_fn._function._func = registered # pyright: ignore[reportPrivateUsage]
342+
return wrap(user_fn)
343+
return wrap(registered)
330344

331345
return decorator
332346

@@ -347,14 +361,15 @@ def entity_trigger(self,
347361
@self._build_function
348362
def wrap(fb: FunctionBuilder) -> FunctionBuilder:
349363
def decorator() -> FunctionBuilder:
364+
registered = fb._function._func # pyright: ignore[reportPrivateUsage]
350365
fb.add_trigger(
351-
trigger=EntityTrigger(name=context_name,
366+
trigger=EntityTrigger(name=getattr(registered, "_df_trigger_name", context_name),
352367
entity_name=entity_name))
353368
return fb
354369

355370
return decorator()
356371

357-
return self._configure_entity_callable(wrap, entity_name)
372+
return self._configure_entity_callable(wrap, context_name, entity_name)
358373

359374
def durable_client_input(self,
360375
client_name: str,
@@ -439,6 +454,7 @@ def set_client_metadata(client_bound: Callable[..., Any]) -> None:
439454
annotations[client_name] = str
440455
client_bound.__annotations__ = annotations
441456
setattr(client_bound, "client_function", function)
457+
preserve_function_source(client_bound, function)
442458

443459
if is_async_function:
444460
@wraps(function)
@@ -477,6 +493,19 @@ class DFApp(Blueprint, FunctionRegister):
477493
Exports the decorators required to declare and index DF Function-types.
478494
"""
479495

496+
def configure_large_payloads(self, *, payload_store: PayloadStore) -> None:
497+
"""Enable payload externalization for this app and its blueprints.
498+
499+
Call once at app startup in every worker process, before invocations.
500+
The store is shared by all durable clients, orchestrations, entities,
501+
and activities in the process. All scaled-out workers must have access
502+
to the same backing storage. Registering a different store in the same
503+
process raises ValueError. Re-registering the same object is allowed.
504+
"""
505+
from ..internal.payloads import configure_payload_store
506+
507+
configure_payload_store(payload_store)
508+
480509
def register_functions(self, function_container: DecoratorApi) -> None:
481510
"""Register the functions of a blueprint into this app.
482511

0 commit comments

Comments
 (0)