Skip to content

Commit b4c4ad6

Browse files
Bernd VerstCopilot
andcommitted
Merge latest main and preserve Azure Managed changelog entries
Resolve the Unreleased changelog conflict by retaining resource audience configuration and the incoming asynchronous Azure Blob payload fix. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
2 parents 466c648 + ba15b23 commit b4c4ad6

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)