Skip to content

Commit 6c0865f

Browse files
Fix Functions RPC URLs and orchestration query page sizes (#342)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
1 parent 06bb277 commit 6c0865f

4 files changed

Lines changed: 265 additions & 2 deletions

File tree

‎azure-functions-durable/CHANGELOG.md‎

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

88
## Unreleased
99

10+
FIXED
11+
12+
- Functions clients now remove HTTP URL paths from the host RPC address before
13+
they connect through gRPC. Both async and sync clients keep the host and port.
14+
- Functions orchestration queries now request at most 1000 instances per page
15+
to support Azure Table Storage. Smaller explicit page sizes are kept. Queries
16+
still return all matching instances, including results after an empty page
17+
with a continuation token.
18+
1019
## v2.0.0rc5
1120

1221
ADDED

‎azure-functions-durable/README.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,14 @@ calls (`context.call_http(...)`), recurring scheduled tasks, and history export.
3636
`OrchestrationQuery.task_hub_names` for filtering orchestration queries by one
3737
or more task hubs. Filtering support depends on the backend.
3838

39+
For both Functions clients, `get_all_orchestration_states()` requests at most
40+
1000 instances per page to support Azure Table Storage. An explicit
41+
`OrchestrationQuery.max_instance_count` below 1000 keeps that page size.
42+
The method follows continuation tokens and returns all matching instances;
43+
the page size is not a total result limit. The `get_status_all()` compatibility
44+
method uses the same limit. Core SDK and Durable Task Scheduler clients keep
45+
their existing page sizes.
46+
3947
`EntityInstanceId` and the compatibility `EntityId` are immutable and hashable,
4048
so they can be used as dictionary keys or set elements. Construct a new ID
4149
instead of assigning or deleting attributes on an existing ID.

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

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import logging
88
import threading
99

10+
from dataclasses import replace
1011
from datetime import datetime, timedelta
1112
from typing import Any, Mapping, Optional, Union, cast, override
1213
from warnings import deprecated
@@ -17,6 +18,7 @@
1718
from durabletask.client import (
1819
AsyncTaskHubGrpcClient,
1920
OrchestrationQuery,
21+
OrchestrationState,
2022
OrchestrationStatus,
2123
TaskHubGrpcClient,
2224
)
@@ -45,6 +47,24 @@
4547
_LOGGER = logging.getLogger(__name__)
4648

4749

50+
def _get_functions_grpc_address(rpc_base_url: str) -> str:
51+
if not rpc_base_url.lower().startswith(("http://", "https://")):
52+
return rpc_base_url
53+
address = urlparse(rpc_base_url)
54+
return f"{address.scheme}://{address.netloc}"
55+
56+
57+
def _get_functions_orchestration_query(
58+
query: OrchestrationQuery | None) -> OrchestrationQuery:
59+
if query is None:
60+
query = OrchestrationQuery()
61+
# The host maps max_instance_count to a page size, not a total result limit.
62+
# Azure Table Storage accepts at most 1000 instances per page.
63+
if query.max_instance_count is None or query.max_instance_count > 1000:
64+
return replace(query, max_instance_count=1000)
65+
return query
66+
67+
4868
def _first_forwarded_value(value: str) -> str:
4969
return value.split(",", 1)[0].strip().strip('"')
5070

@@ -178,7 +198,7 @@ def __init__(self, client_as_string: str):
178198
# We pass in None for the metadata so we don't construct an additional interceptor in the parent class
179199
# Since the parent class doesn't use anything metadata for anything else, we can set it as None
180200
super().__init__(
181-
host_address=self.rpcBaseUrl,
201+
host_address=_get_functions_grpc_address(self.rpcBaseUrl),
182202
secure_channel=False,
183203
metadata=None,
184204
interceptors=interceptors,
@@ -200,6 +220,18 @@ def __init__(self, client_as_string: str):
200220
self._creation_loop = None
201221
self._close_scheduled = False
202222

223+
@override
224+
async def get_all_orchestration_states(
225+
self, orchestration_query: OrchestrationQuery | None = None
226+
) -> list[OrchestrationState]:
227+
"""Get all matching instances, with at most 1000 instances per page.
228+
229+
Smaller explicit ``max_instance_count`` values are kept. The page size
230+
does not limit the total number of instances returned.
231+
"""
232+
return await super().get_all_orchestration_states(
233+
_get_functions_orchestration_query(orchestration_query))
234+
203235
@override
204236
async def get_orchestration_history(
205237
self, instance_id: str, *, execution_id: str | None = None,
@@ -675,7 +707,7 @@ def __init__(self, client_as_string: str):
675707
max_receive_message_length=self.maxGrpcMessageSizeInBytes,
676708
max_send_message_length=self.maxGrpcMessageSizeInBytes)
677709
super().__init__(
678-
host_address=self.rpcBaseUrl,
710+
host_address=_get_functions_grpc_address(self.rpcBaseUrl),
679711
secure_channel=False,
680712
metadata=None,
681713
interceptors=interceptors,
@@ -685,6 +717,18 @@ def __init__(self, client_as_string: str):
685717
emit_trace_spans=False,
686718
logger=_LOGGER)
687719

720+
@override
721+
def get_all_orchestration_states(
722+
self, orchestration_query: OrchestrationQuery | None = None
723+
) -> list[OrchestrationState]:
724+
"""Get all matching instances, with at most 1000 instances per page.
725+
726+
Smaller explicit ``max_instance_count`` values are kept. The page size
727+
does not limit the total number of instances returned.
728+
"""
729+
return super().get_all_orchestration_states(
730+
_get_functions_orchestration_query(orchestration_query))
731+
688732
@override
689733
def get_orchestration_history(
690734
self, instance_id: str, *, execution_id: str | None = None,
Lines changed: 202 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,202 @@
1+
# Copyright (c) Microsoft Corporation.
2+
# Licensed under the MIT License.
3+
4+
import json
5+
from concurrent.futures import ThreadPoolExecutor
6+
from datetime import datetime, timezone
7+
from unittest.mock import AsyncMock, Mock, patch
8+
9+
import grpc
10+
import pytest
11+
from google.protobuf.wrappers_pb2 import StringValue
12+
13+
from azure.durable_functions import DurableFunctionsClient, SyncDurableFunctionsClient
14+
from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationQuery, OrchestrationStatus, TaskHubGrpcClient
15+
from durabletask.internal import orchestrator_service_pb2 as pb
16+
from durabletask.internal import orchestrator_service_pb2_grpc as rpc
17+
18+
19+
def _config(address: str) -> str:
20+
return json.dumps({
21+
"rpcBaseUrl": address,
22+
"taskHubName": "TestHub",
23+
"connectionName": "TestStorage",
24+
"requiredQueryStringParameters": "code=test-key",
25+
"maxGrpcMessageSizeInBytes": 8388608,
26+
})
27+
28+
29+
@pytest.mark.parametrize("use_async", [False, True])
30+
@pytest.mark.parametrize("address, expected", [
31+
("http://127.0.0.1:8080/durabletask/", "http://127.0.0.1:8080"),
32+
("https://host:443/durabletask/?code=key#fragment", "https://host:443"),
33+
("http://[::1]:8080/durabletask/", "http://[::1]:8080"),
34+
("HTTP://localhost:8080/", "http://localhost:8080"),
35+
("localhost:8080", "localhost:8080"),
36+
("dns:///localhost:8080", "dns:///localhost:8080"),
37+
("unix:///tmp/durabletask.sock", "unix:///tmp/durabletask.sock"),
38+
("grpc://localhost:8080", "grpc://localhost:8080"),
39+
("", ""),
40+
])
41+
def test_functions_client_normalizes_only_http_urls(use_async, address, expected):
42+
base = AsyncTaskHubGrpcClient if use_async else TaskHubGrpcClient
43+
client_type = DurableFunctionsClient if use_async else SyncDurableFunctionsClient
44+
with patch.object(base, "__init__", return_value=None) as init:
45+
client = client_type(_config(address))
46+
47+
assert client.rpcBaseUrl == address
48+
assert client.connectionName == "TestStorage"
49+
kwargs = init.call_args.kwargs
50+
assert kwargs["host_address"] == expected
51+
assert kwargs["metadata"] is None
52+
assert kwargs["channel_options"].max_receive_message_length == 8388608
53+
assert kwargs["channel_options"].max_send_message_length == 8388608
54+
interceptor, = kwargs["interceptors"]
55+
assert ("taskhub", "TestHub") in interceptor._metadata
56+
assert interceptor.required_query_string_parameters == "code=test-key"
57+
58+
59+
@pytest.fixture
60+
def local_grpc_server():
61+
metadata = []
62+
63+
class Service(rpc.TaskHubSidecarServiceServicer):
64+
def GetInstance(self, request, context):
65+
metadata.append(dict(context.invocation_metadata()))
66+
return pb.GetInstanceResponse(exists=False)
67+
68+
with ThreadPoolExecutor(max_workers=1) as executor:
69+
server = grpc.server(executor)
70+
rpc.add_TaskHubSidecarServiceServicer_to_server(Service(), server)
71+
port = server.add_insecure_port("127.0.0.1:0")
72+
server.start()
73+
try:
74+
yield f"http://127.0.0.1:{port}/durabletask/", metadata
75+
finally:
76+
server.stop(0).wait()
77+
78+
79+
@pytest.mark.asyncio
80+
@pytest.mark.parametrize("use_async", [False, True])
81+
async def test_functions_client_reaches_grpc_server_with_url_path(local_grpc_server, use_async):
82+
address, metadata = local_grpc_server
83+
if use_async:
84+
client = DurableFunctionsClient(_config(address))
85+
try:
86+
response = await client._get_stub().GetInstance(
87+
pb.GetInstanceRequest(instanceId="instance"), timeout=2)
88+
finally:
89+
await client.close()
90+
else:
91+
client = SyncDurableFunctionsClient(_config(address))
92+
try:
93+
response = client._stub.GetInstance(
94+
pb.GetInstanceRequest(instanceId="instance"), timeout=2)
95+
finally:
96+
client.close()
97+
assert not response.exists
98+
assert metadata[0]["taskhub"] == "TestHub"
99+
assert metadata[0]["x-user-agent"].startswith("durabletask-python/")
100+
101+
102+
@pytest.mark.asyncio
103+
@pytest.mark.parametrize("use_async", [False, True])
104+
@pytest.mark.parametrize("page_size, expected", [
105+
((1 << 31) - 1, 1000), (None, 1000), (1001, 1000),
106+
(1000, 1000), (25, 25), (0, 0),
107+
])
108+
async def test_functions_query_limits_page_size_without_changing_query(use_async, page_size, expected):
109+
query = OrchestrationQuery(
110+
created_time_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
111+
created_time_to=datetime(2026, 2, 1, tzinfo=timezone.utc),
112+
runtime_status=[OrchestrationStatus.RUNNING],
113+
max_instance_count=page_size,
114+
fetch_inputs_and_outputs=True,
115+
instance_id_prefix="prefix",
116+
task_hub_names=["TestHub"],
117+
)
118+
stub = Mock()
119+
client = (DurableFunctionsClient(_config("localhost:8080")) if use_async
120+
else SyncDurableFunctionsClient(_config("localhost:8080")))
121+
response = pb.QueryInstancesResponse()
122+
try:
123+
if use_async:
124+
stub.QueryInstances = AsyncMock(return_value=response)
125+
with patch.object(client, "_get_stub", return_value=stub):
126+
assert await client.get_all_orchestration_states(query) == []
127+
else:
128+
stub.QueryInstances.return_value = response
129+
with patch.object(client, "_stub", stub):
130+
assert client.get_all_orchestration_states(query) == []
131+
request = stub.QueryInstances.call_args.args[0].query
132+
assert request.maxInstanceCount == expected
133+
assert request.fetchInputsAndOutputs
134+
assert request.instanceIdPrefix.value == "prefix"
135+
assert list(request.runtimeStatus) == [pb.ORCHESTRATION_STATUS_RUNNING]
136+
assert [hub.value for hub in request.taskHubNames] == ["TestHub"]
137+
assert request.createdTimeFrom.ToDatetime(timezone.utc) == query.created_time_from
138+
assert request.createdTimeTo.ToDatetime(timezone.utc) == query.created_time_to
139+
assert query.max_instance_count == page_size
140+
finally:
141+
if use_async:
142+
await client.close()
143+
else:
144+
client.close()
145+
146+
147+
@pytest.mark.asyncio
148+
@pytest.mark.parametrize("method, page_size", [
149+
("async", None), ("sync", None), ("get_status_all", None),
150+
("async", 25), ("sync", 25),
151+
])
152+
async def test_functions_query_reads_all_pages_including_empty_pages(method, page_size):
153+
first_page_count = page_size or 1000
154+
query = OrchestrationQuery(max_instance_count=page_size) if page_size else None
155+
responses = iter([
156+
pb.QueryInstancesResponse(
157+
orchestrationState=[
158+
pb.OrchestrationState(instanceId=str(i), name="test",
159+
orchestrationStatus=pb.ORCHESTRATION_STATUS_RUNNING)
160+
for i in range(first_page_count)
161+
],
162+
continuationToken=StringValue(value="next")),
163+
pb.QueryInstancesResponse(continuationToken=StringValue(value="after-empty")),
164+
pb.QueryInstancesResponse(
165+
orchestrationState=[
166+
pb.OrchestrationState(instanceId=str(i), name="test",
167+
orchestrationStatus=pb.ORCHESTRATION_STATUS_RUNNING)
168+
for i in range(first_page_count, first_page_count + 2)
169+
]),
170+
])
171+
tokens = []
172+
173+
def query_instances(request):
174+
assert request.query.maxInstanceCount == first_page_count
175+
tokens.append(request.query.continuationToken.value)
176+
assert request.query.fetchInputsAndOutputs == (method == "get_status_all")
177+
return next(responses)
178+
179+
stub = Mock()
180+
if method == "sync":
181+
client = SyncDurableFunctionsClient(_config("localhost:8080"))
182+
stub.QueryInstances.side_effect = query_instances
183+
try:
184+
with patch.object(client, "_stub", stub):
185+
states = client.get_all_orchestration_states(query)
186+
finally:
187+
client.close()
188+
else:
189+
client = DurableFunctionsClient(_config("localhost:8080"))
190+
stub.QueryInstances = AsyncMock(side_effect=query_instances)
191+
try:
192+
with patch.object(client, "_get_stub", return_value=stub):
193+
if method == "get_status_all":
194+
with pytest.warns(DeprecationWarning):
195+
states = await client.get_status_all()
196+
else:
197+
states = await client.get_all_orchestration_states(query)
198+
finally:
199+
await client.close()
200+
201+
assert [state.instance_id for state in states] == [str(i) for i in range(first_page_count + 2)]
202+
assert tokens == ["", "next", "after-empty"]

0 commit comments

Comments
 (0)