From 648a298c285e445fc6b440e586a0ac928e03c39f Mon Sep 17 00:00:00 2001 From: Tsuyoshi Ushio Date: Wed, 4 Nov 2026 14:56:56 -0800 Subject: [PATCH] Fix Functions RPC URLs and orchestration query page sizes Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- azure-functions-durable/CHANGELOG.md | 9 + azure-functions-durable/README.md | 8 + .../azure/durable_functions/client.py | 48 ++++- .../test_client_transport.py | 202 ++++++++++++++++++ 4 files changed, 265 insertions(+), 2 deletions(-) create mode 100644 tests/azure-functions-durable/test_client_transport.py diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 4b808272..bfa2c085 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +FIXED + +- Functions clients now remove HTTP URL paths from the host RPC address before +they connect through gRPC. Both async and sync clients keep the host and port. +- Functions orchestration queries now request at most 1000 instances per page +to support Azure Table Storage. Smaller explicit page sizes are kept. Queries +still return all matching instances, including results after an empty page +with a continuation token. + ## v2.0.0rc4 ADDED diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 9bf0d77b..5351e0ac 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -36,6 +36,14 @@ calls (`context.call_http(...)`), recurring scheduled tasks, and history export. `OrchestrationQuery.task_hub_names` for filtering orchestration queries by one or more task hubs. Filtering support depends on the backend. +For both Functions clients, `get_all_orchestration_states()` requests at most +1000 instances per page to support Azure Table Storage. An explicit +`OrchestrationQuery.max_instance_count` below 1000 keeps that page size. +The method follows continuation tokens and returns all matching instances; +the page size is not a total result limit. The `get_status_all()` compatibility +method uses the same limit. Core SDK and Durable Task Scheduler clients keep +their existing page sizes. + `EntityInstanceId` and the compatibility `EntityId` are immutable and hashable, so they can be used as dictionary keys or set elements. Construct a new ID instead of assigning or deleting attributes on an existing ID. diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index 5b637b55..1e4809b6 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/azure-functions-durable/azure/durable_functions/client.py @@ -7,6 +7,7 @@ import logging import threading +from dataclasses import replace from datetime import datetime, timedelta from typing import Any, Mapping, Optional, Union, cast, override from warnings import deprecated @@ -17,6 +18,7 @@ from durabletask.client import ( AsyncTaskHubGrpcClient, OrchestrationQuery, + OrchestrationState, OrchestrationStatus, TaskHubGrpcClient, ) @@ -45,6 +47,24 @@ _LOGGER = logging.getLogger(__name__) +def _get_functions_grpc_address(rpc_base_url: str) -> str: + if not rpc_base_url.lower().startswith(("http://", "https://")): + return rpc_base_url + address = urlparse(rpc_base_url) + return f"{address.scheme}://{address.netloc}" + + +def _get_functions_orchestration_query( + query: OrchestrationQuery | None) -> OrchestrationQuery: + if query is None: + query = OrchestrationQuery() + # The host maps max_instance_count to a page size, not a total result limit. + # Azure Table Storage accepts at most 1000 instances per page. + if query.max_instance_count is None or query.max_instance_count > 1000: + return replace(query, max_instance_count=1000) + return query + + def _first_forwarded_value(value: str) -> str: return value.split(",", 1)[0].strip().strip('"') @@ -178,7 +198,7 @@ def __init__(self, client_as_string: str): # We pass in None for the metadata so we don't construct an additional interceptor in the parent class # Since the parent class doesn't use anything metadata for anything else, we can set it as None super().__init__( - host_address=self.rpcBaseUrl, + host_address=_get_functions_grpc_address(self.rpcBaseUrl), secure_channel=False, metadata=None, interceptors=interceptors, @@ -200,6 +220,18 @@ def __init__(self, client_as_string: str): self._creation_loop = None self._close_scheduled = False + @override + async def get_all_orchestration_states( + self, orchestration_query: OrchestrationQuery | None = None + ) -> list[OrchestrationState]: + """Get all matching instances, with at most 1000 instances per page. + + Smaller explicit ``max_instance_count`` values are kept. The page size + does not limit the total number of instances returned. + """ + return await super().get_all_orchestration_states( + _get_functions_orchestration_query(orchestration_query)) + @override async def get_orchestration_history( self, instance_id: str, *, execution_id: str | None = None, @@ -675,7 +707,7 @@ def __init__(self, client_as_string: str): max_receive_message_length=self.maxGrpcMessageSizeInBytes, max_send_message_length=self.maxGrpcMessageSizeInBytes) super().__init__( - host_address=self.rpcBaseUrl, + host_address=_get_functions_grpc_address(self.rpcBaseUrl), secure_channel=False, metadata=None, interceptors=interceptors, @@ -685,6 +717,18 @@ def __init__(self, client_as_string: str): emit_trace_spans=False, logger=_LOGGER) + @override + def get_all_orchestration_states( + self, orchestration_query: OrchestrationQuery | None = None + ) -> list[OrchestrationState]: + """Get all matching instances, with at most 1000 instances per page. + + Smaller explicit ``max_instance_count`` values are kept. The page size + does not limit the total number of instances returned. + """ + return super().get_all_orchestration_states( + _get_functions_orchestration_query(orchestration_query)) + @override def get_orchestration_history( self, instance_id: str, *, execution_id: str | None = None, diff --git a/tests/azure-functions-durable/test_client_transport.py b/tests/azure-functions-durable/test_client_transport.py new file mode 100644 index 00000000..33578cdf --- /dev/null +++ b/tests/azure-functions-durable/test_client_transport.py @@ -0,0 +1,202 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import json +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from unittest.mock import AsyncMock, Mock, patch + +import grpc +import pytest +from google.protobuf.wrappers_pb2 import StringValue + +from azure.durable_functions import DurableFunctionsClient, SyncDurableFunctionsClient +from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationQuery, OrchestrationStatus, TaskHubGrpcClient +from durabletask.internal import orchestrator_service_pb2 as pb +from durabletask.internal import orchestrator_service_pb2_grpc as rpc + + +def _config(address: str) -> str: + return json.dumps({ + "rpcBaseUrl": address, + "taskHubName": "TestHub", + "connectionName": "TestStorage", + "requiredQueryStringParameters": "code=test-key", + "maxGrpcMessageSizeInBytes": 8388608, + }) + + +@pytest.mark.parametrize("use_async", [False, True]) +@pytest.mark.parametrize("address, expected", [ + ("http://127.0.0.1:8080/durabletask/", "http://127.0.0.1:8080"), + ("https://host:443/durabletask/?code=key#fragment", "https://host:443"), + ("http://[::1]:8080/durabletask/", "http://[::1]:8080"), + ("HTTP://localhost:8080/", "http://localhost:8080"), + ("localhost:8080", "localhost:8080"), + ("dns:///localhost:8080", "dns:///localhost:8080"), + ("unix:///tmp/durabletask.sock", "unix:///tmp/durabletask.sock"), + ("grpc://localhost:8080", "grpc://localhost:8080"), + ("", ""), +]) +def test_functions_client_normalizes_only_http_urls(use_async, address, expected): + base = AsyncTaskHubGrpcClient if use_async else TaskHubGrpcClient + client_type = DurableFunctionsClient if use_async else SyncDurableFunctionsClient + with patch.object(base, "__init__", return_value=None) as init: + client = client_type(_config(address)) + + assert client.rpcBaseUrl == address + assert client.connectionName == "TestStorage" + kwargs = init.call_args.kwargs + assert kwargs["host_address"] == expected + assert kwargs["metadata"] is None + assert kwargs["channel_options"].max_receive_message_length == 8388608 + assert kwargs["channel_options"].max_send_message_length == 8388608 + interceptor, = kwargs["interceptors"] + assert ("taskhub", "TestHub") in interceptor._metadata + assert interceptor.required_query_string_parameters == "code=test-key" + + +@pytest.fixture +def local_grpc_server(): + metadata = [] + + class Service(rpc.TaskHubSidecarServiceServicer): + def GetInstance(self, request, context): + metadata.append(dict(context.invocation_metadata())) + return pb.GetInstanceResponse(exists=False) + + with ThreadPoolExecutor(max_workers=1) as executor: + server = grpc.server(executor) + rpc.add_TaskHubSidecarServiceServicer_to_server(Service(), server) + port = server.add_insecure_port("127.0.0.1:0") + server.start() + try: + yield f"http://127.0.0.1:{port}/durabletask/", metadata + finally: + server.stop(0).wait() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_async", [False, True]) +async def test_functions_client_reaches_grpc_server_with_url_path(local_grpc_server, use_async): + address, metadata = local_grpc_server + if use_async: + client = DurableFunctionsClient(_config(address)) + try: + response = await client._get_stub().GetInstance( + pb.GetInstanceRequest(instanceId="instance"), timeout=2) + finally: + await client.close() + else: + client = SyncDurableFunctionsClient(_config(address)) + try: + response = client._stub.GetInstance( + pb.GetInstanceRequest(instanceId="instance"), timeout=2) + finally: + client.close() + assert not response.exists + assert metadata[0]["taskhub"] == "TestHub" + assert metadata[0]["x-user-agent"].startswith("durabletask-python/") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_async", [False, True]) +@pytest.mark.parametrize("page_size, expected", [ + ((1 << 31) - 1, 1000), (None, 1000), (1001, 1000), + (1000, 1000), (25, 25), (0, 0), +]) +async def test_functions_query_limits_page_size_without_changing_query(use_async, page_size, expected): + query = OrchestrationQuery( + created_time_from=datetime(2026, 1, 1, tzinfo=timezone.utc), + created_time_to=datetime(2026, 2, 1, tzinfo=timezone.utc), + runtime_status=[OrchestrationStatus.RUNNING], + max_instance_count=page_size, + fetch_inputs_and_outputs=True, + instance_id_prefix="prefix", + task_hub_names=["TestHub"], + ) + stub = Mock() + client = (DurableFunctionsClient(_config("localhost:8080")) if use_async + else SyncDurableFunctionsClient(_config("localhost:8080"))) + response = pb.QueryInstancesResponse() + try: + if use_async: + stub.QueryInstances = AsyncMock(return_value=response) + with patch.object(client, "_get_stub", return_value=stub): + assert await client.get_all_orchestration_states(query) == [] + else: + stub.QueryInstances.return_value = response + with patch.object(client, "_stub", stub): + assert client.get_all_orchestration_states(query) == [] + request = stub.QueryInstances.call_args.args[0].query + assert request.maxInstanceCount == expected + assert request.fetchInputsAndOutputs + assert request.instanceIdPrefix.value == "prefix" + assert list(request.runtimeStatus) == [pb.ORCHESTRATION_STATUS_RUNNING] + assert [hub.value for hub in request.taskHubNames] == ["TestHub"] + assert request.createdTimeFrom.ToDatetime(timezone.utc) == query.created_time_from + assert request.createdTimeTo.ToDatetime(timezone.utc) == query.created_time_to + assert query.max_instance_count == page_size + finally: + if use_async: + await client.close() + else: + client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method, page_size", [ + ("async", None), ("sync", None), ("get_status_all", None), + ("async", 25), ("sync", 25), +]) +async def test_functions_query_reads_all_pages_including_empty_pages(method, page_size): + first_page_count = page_size or 1000 + query = OrchestrationQuery(max_instance_count=page_size) if page_size else None + responses = iter([ + pb.QueryInstancesResponse( + orchestrationState=[ + pb.OrchestrationState(instanceId=str(i), name="test", + orchestrationStatus=pb.ORCHESTRATION_STATUS_RUNNING) + for i in range(first_page_count) + ], + continuationToken=StringValue(value="next")), + pb.QueryInstancesResponse(continuationToken=StringValue(value="after-empty")), + pb.QueryInstancesResponse( + orchestrationState=[ + pb.OrchestrationState(instanceId=str(i), name="test", + orchestrationStatus=pb.ORCHESTRATION_STATUS_RUNNING) + for i in range(first_page_count, first_page_count + 2) + ]), + ]) + tokens = [] + + def query_instances(request): + assert request.query.maxInstanceCount == first_page_count + tokens.append(request.query.continuationToken.value) + assert request.query.fetchInputsAndOutputs == (method == "get_status_all") + return next(responses) + + stub = Mock() + if method == "sync": + client = SyncDurableFunctionsClient(_config("localhost:8080")) + stub.QueryInstances.side_effect = query_instances + try: + with patch.object(client, "_stub", stub): + states = client.get_all_orchestration_states(query) + finally: + client.close() + else: + client = DurableFunctionsClient(_config("localhost:8080")) + stub.QueryInstances = AsyncMock(side_effect=query_instances) + try: + with patch.object(client, "_get_stub", return_value=stub): + if method == "get_status_all": + with pytest.warns(DeprecationWarning): + states = await client.get_status_all() + else: + states = await client.get_all_orchestration_states(query) + finally: + await client.close() + + assert [state.instance_id for state in states] == [str(i) for i in range(first_page_count + 2)] + assert tokens == ["", "next", "after-empty"]