Fix Functions RPC URLs and orchestration query page sizes

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.0rc5
 
 ADDED

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.

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,

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"]