From f834941c72ec3567736ff88ba56e121f3e30659f Mon Sep 17 00:00:00 2001 From: Pranjal Patel Date: Wed, 12 Aug 2026 17:18:53 -0700 Subject: [PATCH 1/4] feat(servicebus): send server-timeout on management operations Management operations (peek, deferred receive, lock renewal, session state, session listing, schedule/cancel scheduled) now send the `com.microsoft:server-timeout` application property, asking the service to bound the operation on its side. The value is the caller's remaining time less a one second buffer, so the service answers before the client gives up, or 60 seconds when the caller supplied no timeout. Previously no bound was sent at all, so a stalled service could hold a management call until the AMQP link itself failed. The property is set in the base handler, after the retry loop computes the remaining time, so it reflects the time actually left on the attempt. Both transports pass application properties through unchanged, so no transport change is needed. REQUEST_RESPONSE_TIMEOUT already held the correct vendor-prefixed key and was previously unused. Matches the .NET, Java and Go SDKs (.NET ServiceBusRetryOptions.TryTimeout, Java MessageUtils.adjustServerTimeout, Go defaultServerTimeout). Refs: AB#38822503 --- sdk/servicebus/azure-servicebus/CHANGELOG.md | 1 + .../azure/servicebus/_base_handler.py | 12 +- .../azure/servicebus/_common/constants.py | 6 + .../azure/servicebus/_common/utils.py | 18 +++ .../servicebus/aio/_base_handler_async.py | 12 +- .../tests/unittests/test_server_timeout.py | 150 ++++++++++++++++++ 6 files changed, 197 insertions(+), 2 deletions(-) create mode 100644 sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index aeca072347ff..eb74911475b0 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -6,6 +6,7 @@ - Added `ServiceBusReceivedMessage.from_bytes()` classmethod to construct a `ServiceBusReceivedMessage` from raw AMQP payload bytes without requiring the deprecated `uamqp` library. ([#43979](https://github.com/Azure/azure-sdk-for-python/issues/43979)) - Added `ServiceBusClient.list_queue_sessions()` and `ServiceBusClient.list_subscription_sessions()` (sync and async) to list session IDs for entities with active messages, with optional filtering by session-state update timestamp. The methods return an `ItemPaged[str]` (`AsyncItemPaged[str]` on the async client) so callers can iterate every session transparently or page with `by_page()`. Implements the `com.microsoft:get-message-sessions` management operation. ([#46575](https://github.com/Azure/azure-sdk-for-python/pull/46575)) +- Management operations (peek, deferred receive, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs. ### Bugs Fixed diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py index 31ad3df26f72..fa3baacfb287 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py @@ -25,13 +25,19 @@ OperationTimeoutError, SessionLockLostError, ) -from ._common.utils import create_properties, strip_protocol_from_uri, parse_sas_credential +from ._common.utils import ( + create_properties, + strip_protocol_from_uri, + parse_sas_credential, + get_server_timeout_ms, +) from ._common.constants import ( CONTAINER_PREFIX, MANAGEMENT_PATH_SUFFIX, TOKEN_TYPE_SASTOKEN, MGMT_REQUEST_OP_TYPE_ENTITY_MGMT, ASSOCIATEDLINKPROPERTYNAME, + REQUEST_RESPONSE_TIMEOUT, ) if TYPE_CHECKING: @@ -499,6 +505,10 @@ def _mgmt_request_response( except AttributeError: pass + application_properties[REQUEST_RESPONSE_TIMEOUT] = self._amqp_transport.AMQP_UINT_VALUE( + get_server_timeout_ms(timeout) + ) + mgmt_msg = self._amqp_transport.create_mgmt_msg( message=message, application_properties=application_properties, diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py index 061b53d8bd6b..57d940c8b621 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py @@ -61,6 +61,12 @@ PYAMQP_LIBRARY = "pyamqp" OPERATION_TIMEOUT = VENDOR + b":timeout" +# Bounds a management operation on the service side when the caller gave no timeout. +DEFAULT_SERVER_TIMEOUT_MS = 60000 +# Subtracted from the caller's remaining time so the service answers before the client +# gives up. +SERVER_TIMEOUT_BUFFER_MS = 1000 + MANAGEMENT_PATH_SUFFIX = "/$management" MGMT_RESPONSE_SESSION_STATE = b"session-state" diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py index 92f6f9e981fc..92c682b07ceb 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py @@ -37,6 +37,8 @@ DEAD_LETTER_QUEUE_SUFFIX, TRANSFER_DEAD_LETTER_QUEUE_SUFFIX, USER_AGENT_PREFIX, + DEFAULT_SERVER_TIMEOUT_MS, + SERVER_TIMEOUT_BUFFER_MS, ) from ..amqp import AmqpAnnotatedMessage @@ -89,6 +91,22 @@ def utc_now(): return datetime.datetime.now(timezone.utc) +def get_server_timeout_ms(timeout: Optional[float]) -> int: + """Return the server-timeout for a management operation, in milliseconds. + + This is a service-side bound, not a client-side one. It clamps at zero, since under + a second of remaining time there is no room for the service to answer first. + + :param float or None timeout: The caller's remaining timeout in seconds, or None. + :rtype: int + :returns: The remaining time less the buffer, or the default if no timeout was given. + """ + if timeout is None: + return DEFAULT_SERVER_TIMEOUT_MS + remaining_ms = int(timeout * 1000) - SERVER_TIMEOUT_BUFFER_MS + return max(remaining_ms, 0) + + def build_uri(address, entity): parsed = urlparse(address) if parsed.path: diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_base_handler_async.py b/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_base_handler_async.py index bde1a82d6771..12addc4d3dbb 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_base_handler_async.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/aio/_base_handler_async.py @@ -13,13 +13,19 @@ from ._transport._pyamqp_transport_async import PyamqpTransportAsync from .._base_handler import _generate_sas_token, BaseHandler as BaseHandlerSync, _get_backoff_time from .._common._configuration import Configuration -from .._common.utils import create_properties, strip_protocol_from_uri, parse_sas_credential +from .._common.utils import ( + create_properties, + strip_protocol_from_uri, + parse_sas_credential, + get_server_timeout_ms, +) from .._common.constants import ( TOKEN_TYPE_SASTOKEN, MGMT_REQUEST_OP_TYPE_ENTITY_MGMT, ASSOCIATEDLINKPROPERTYNAME, CONTAINER_PREFIX, MANAGEMENT_PATH_SUFFIX, + REQUEST_RESPONSE_TIMEOUT, ) from ..exceptions import ( ServiceBusConnectionError, @@ -340,6 +346,10 @@ async def _mgmt_request_response( except AttributeError: pass + application_properties[REQUEST_RESPONSE_TIMEOUT] = self._amqp_transport.AMQP_UINT_VALUE( + get_server_timeout_ms(timeout) + ) + mgmt_msg = self._amqp_transport.create_mgmt_msg( # type: ignore # TODO: fix mypy message=message, application_properties=application_properties, diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py new file mode 100644 index 000000000000..f46c220c6c70 --- /dev/null +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py @@ -0,0 +1,150 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. + +"""Unit tests for the management server-timeout. + +Management operations send the `com.microsoft:server-timeout` property, asking the service to +bound the operation on its side. The value is the caller's remaining time less a one second +buffer, so the service answers before the client gives up, or 60 seconds when the caller +supplied no timeout. + +Previously no bound was sent at all, so a stalled service could hold a management call until +the AMQP link itself failed. Matches the .NET, Java and Go SDKs. +""" + +from unittest.mock import MagicMock + +import pytest + +from azure.servicebus._common.constants import REQUEST_RESPONSE_TIMEOUT +from azure.servicebus._common.utils import get_server_timeout_ms +from azure.servicebus._transport._pyamqp_transport import PyamqpTransport + + +class TestServerTimeoutMillis: + """`get_server_timeout_ms` converts the caller's remaining time into the + value advertised to the service.""" + + @pytest.mark.parametrize( + "remaining_seconds,expected_ms", + [ + (None, 60000), # no caller timeout: the default, not "no bound at all" + (120, 119000), + (60, 59000), # equals the default, but must still take the buffer path + (10, 9000), + (1.5, 500), + (1, 0), # at the buffer, nothing left to give the service + (0.5, 0), + (-5, 0), # deadline already passed: clamped, never a negative uint + ], + ) + def test_remaining_time_less_buffer(self, remaining_seconds, expected_ms): + assert get_server_timeout_ms(remaining_seconds) == expected_ms + + def test_wire_contract(self): + # The key .NET and Java send, encoded as an unsigned int of milliseconds. + assert REQUEST_RESPONSE_TIMEOUT == b"com.microsoft:server-timeout" + encoded = PyamqpTransport.AMQP_UINT_VALUE(get_server_timeout_ms(None)) + assert encoded == {"TYPE": "UINT", "VALUE": 60000} + + +class TestManagementRequestSetsServerTimeout: + """The property must actually reach the outgoing management message, on both + the associated-link and no-associated-link paths.""" + + def _make_handler(self): + from azure.servicebus._base_handler import BaseHandler + + captured = {} + + def fake_create_mgmt_msg(message, application_properties, config, reply_to, **kwargs): + captured.clear() + captured.update(application_properties) + return MagicMock() + + handler = BaseHandler.__new__(BaseHandler) + handler._amqp_transport = MagicMock() + handler._amqp_transport.create_mgmt_msg = fake_create_mgmt_msg + handler._amqp_transport.AMQP_UINT_VALUE = PyamqpTransport.AMQP_UINT_VALUE + handler._amqp_transport.get_handler_link_name = lambda h: "link-1" + handler._amqp_transport.mgmt_client_request = lambda *args, **kwargs: "response" + handler._amqp_transport.TIMEOUT_ERROR = TimeoutError + handler._open = lambda: None + handler._handler = MagicMock() + handler._config = MagicMock(encoding="UTF-8") + handler._mgmt_target = "queue/$management" + return handler, captured + + def test_default_sent_when_caller_gave_no_timeout(self): + # The gap this closes: without a caller timeout the service previously received no bound at + # all. + handler, captured = self._make_handler() + handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None) + assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 60000} + + def test_remaining_time_less_buffer_sent(self): + handler, captured = self._make_handler() + handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=10) + assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 9000} + + def test_clamped_below_buffer(self): + handler, captured = self._make_handler() + handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=0.4) + assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 0} + + def test_associated_link_name_preserved(self): + handler, captured = self._make_handler() + handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None) + assert b"associated-link-name" in captured + assert REQUEST_RESPONSE_TIMEOUT in captured + + def test_sent_on_calls_without_an_associated_link(self): + # Operations such as list_sessions pass `keep_alive_associated_link=False` and start from an + # empty property map; they must still be bounded. + handler, captured = self._make_handler() + handler._mgmt_request_response( + b"op", + {}, + lambda *a: None, + keep_alive_associated_link=False, + timeout=None, + ) + assert captured == {REQUEST_RESPONSE_TIMEOUT: {"TYPE": "UINT", "VALUE": 60000}} + + +class TestAsyncParity: + """The async management path is a separate implementation and can drift from the sync one.""" + + @pytest.mark.asyncio + async def test_async_management_sets_server_timeout(self): + from azure.servicebus.aio._base_handler_async import BaseHandler as AsyncBaseHandler + + captured = {} + + def fake_create_mgmt_msg(message, application_properties, config, reply_to, **kwargs): + captured.clear() + captured.update(application_properties) + return MagicMock() + + async def fake_request(*args, **kwargs): + return "response" + + async def fake_open(): + return None + + handler = AsyncBaseHandler.__new__(AsyncBaseHandler) + handler._amqp_transport = MagicMock() + handler._amqp_transport.create_mgmt_msg = fake_create_mgmt_msg + handler._amqp_transport.AMQP_UINT_VALUE = PyamqpTransport.AMQP_UINT_VALUE + handler._amqp_transport.get_handler_link_name = lambda h: "link-1" + handler._amqp_transport.mgmt_client_request_async = fake_request + handler._amqp_transport.TIMEOUT_ERROR = TimeoutError + handler._open = fake_open + handler._handler = MagicMock() + handler._config = MagicMock(encoding="UTF-8") + handler._mgmt_target = "queue/$management" + + await handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None) + assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 60000} + + await handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=10) + assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 9000} From e2e1d9d573045db000e857bbfa2546328fa50885 Mon Sep 17 00:00:00 2001 From: Pranjal Patel Date: Wed, 12 Aug 2026 17:49:48 -0700 Subject: [PATCH 2/4] fix(servicebus): cap server-timeout at the AMQP uint maximum Address review feedback on the management server-timeout change. The public management APIs bound `timeout` only at zero, so a caller could pass a value whose millisecond form exceeds a uint32 (about 49.7 days). The AMQP encoder would then raise while packing the message rather than performing the operation. Clamp before scaling as well as after, so a very large or infinite float cannot overflow the int conversion either. Also document message settlement over the management link as affected. It reaches the service through the common handler when a message cannot be settled on the receive link, for example after the lock has expired or for a deferred message, and passes no timeout, so it takes the 60 second default. Refs: AB#38822503 --- sdk/servicebus/azure-servicebus/CHANGELOG.md | 2 +- .../azure/servicebus/_common/constants.py | 2 ++ .../azure/servicebus/_common/utils.py | 9 ++++++--- .../tests/unittests/test_server_timeout.py | 14 +++++++++++++- 4 files changed, 22 insertions(+), 5 deletions(-) diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index eb74911475b0..e06d480c172f 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -6,7 +6,7 @@ - Added `ServiceBusReceivedMessage.from_bytes()` classmethod to construct a `ServiceBusReceivedMessage` from raw AMQP payload bytes without requiring the deprecated `uamqp` library. ([#43979](https://github.com/Azure/azure-sdk-for-python/issues/43979)) - Added `ServiceBusClient.list_queue_sessions()` and `ServiceBusClient.list_subscription_sessions()` (sync and async) to list session IDs for entities with active messages, with optional filtering by session-state update timestamp. The methods return an `ItemPaged[str]` (`AsyncItemPaged[str]` on the async client) so callers can iterate every session transparently or page with `by_page()`. Implements the `com.microsoft:get-message-sessions` management operation. ([#46575](https://github.com/Azure/azure-sdk-for-python/pull/46575)) -- Management operations (peek, deferred receive, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs. +- Management operations (peek, deferred receive, message settlement over the management link, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs. ### Bugs Fixed diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py index 57d940c8b621..5f7c67d0e916 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py @@ -66,6 +66,8 @@ # Subtracted from the caller's remaining time so the service answers before the client # gives up. SERVER_TIMEOUT_BUFFER_MS = 1000 +# The property is encoded as an AMQP uint, so cap at its maximum (about 49.7 days). +MAX_SERVER_TIMEOUT_MS = 2**32 - 1 MANAGEMENT_PATH_SUFFIX = "/$management" diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py index 92c682b07ceb..64445f1ccddd 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py @@ -39,6 +39,7 @@ USER_AGENT_PREFIX, DEFAULT_SERVER_TIMEOUT_MS, SERVER_TIMEOUT_BUFFER_MS, + MAX_SERVER_TIMEOUT_MS, ) from ..amqp import AmqpAnnotatedMessage @@ -95,7 +96,8 @@ def get_server_timeout_ms(timeout: Optional[float]) -> int: """Return the server-timeout for a management operation, in milliseconds. This is a service-side bound, not a client-side one. It clamps at zero, since under - a second of remaining time there is no room for the service to answer first. + a second of remaining time there is no room for the service to answer first, and at + the AMQP uint maximum, since the value is encoded as one. :param float or None timeout: The caller's remaining timeout in seconds, or None. :rtype: int @@ -103,8 +105,9 @@ def get_server_timeout_ms(timeout: Optional[float]) -> int: """ if timeout is None: return DEFAULT_SERVER_TIMEOUT_MS - remaining_ms = int(timeout * 1000) - SERVER_TIMEOUT_BUFFER_MS - return max(remaining_ms, 0) + capped = min(timeout, MAX_SERVER_TIMEOUT_MS / 1000) + remaining_ms = int(capped * 1000) - SERVER_TIMEOUT_BUFFER_MS + return min(max(remaining_ms, 0), MAX_SERVER_TIMEOUT_MS) def build_uri(address, entity): diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py index f46c220c6c70..36d5a71278a9 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py @@ -13,9 +13,11 @@ from unittest.mock import MagicMock +import struct + import pytest -from azure.servicebus._common.constants import REQUEST_RESPONSE_TIMEOUT +from azure.servicebus._common.constants import MAX_SERVER_TIMEOUT_MS, REQUEST_RESPONSE_TIMEOUT from azure.servicebus._common.utils import get_server_timeout_ms from azure.servicebus._transport._pyamqp_transport import PyamqpTransport @@ -46,6 +48,16 @@ def test_wire_contract(self): encoded = PyamqpTransport.AMQP_UINT_VALUE(get_server_timeout_ms(None)) assert encoded == {"TYPE": "UINT", "VALUE": 60000} + @pytest.mark.parametrize("remaining_seconds", [4294968, 5_000_000, 1e12, float("inf")]) + def test_capped_at_the_amqp_uint_maximum(self, remaining_seconds): + # `timeout` is only bounded at zero, so a large value would overflow the uint encoder. + result = get_server_timeout_ms(remaining_seconds) + assert result <= MAX_SERVER_TIMEOUT_MS + struct.pack(">I", result) # raises if it does not fit + + def test_just_below_the_cap_is_not_clamped(self): + assert get_server_timeout_ms(4294967) == 4294966000 + class TestManagementRequestSetsServerTimeout: """The property must actually reach the outgoing management message, on both From 12a85bc1b6895410ad02aab45f97aae4a72dda8f Mon Sep 17 00:00:00 2001 From: Pranjal Patel Date: Fri, 14 Aug 2026 09:56:52 -0700 Subject: [PATCH 3/4] test(servicebus): cover the response half and the uamqp value type Address review feedback. Nothing tested the response half: every assertion covered what the client sends, while the changelog promises a retryable OperationTimeoutError that rests on the service returning com.microsoft:timeout in errorCondition. Push a fake non-200 response with that condition through mgmt_handlers.default and assert the type, for both transports, with no live namespace. uamqp is the one place the value type changes, since pyamqp uses a plain dict where uamqp uses an AMQPuInt that has not previously gone into application_properties. One test stubs the transport type to prove the property is built from the transport's own AMQP_UINT_VALUE, and one encodes a real message when uamqp is installed. Also drop the redundant outer min in get_server_timeout_ms, which never binds because the buffer is subtracted after the cap, and move the changelog entry to Other Changes since it changes existing operations rather than adding a feature. api.metadata.yml carries the parserVersion bump only. That field is gated by the consistency check and the package still had 0.3.30 while the check generates 0.3.31. apiMdSha256 and pythonVersion are unchanged. Refs: AB#38822503 --- sdk/servicebus/azure-servicebus/CHANGELOG.md | 2 +- .../azure-servicebus/api.metadata.yml | 2 +- .../azure/servicebus/_common/utils.py | 3 +- .../tests/unittests/test_server_timeout.py | 95 ++++++++++++++++++- 4 files changed, 93 insertions(+), 9 deletions(-) diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index e06d480c172f..40648e0aab3f 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -6,7 +6,6 @@ - Added `ServiceBusReceivedMessage.from_bytes()` classmethod to construct a `ServiceBusReceivedMessage` from raw AMQP payload bytes without requiring the deprecated `uamqp` library. ([#43979](https://github.com/Azure/azure-sdk-for-python/issues/43979)) - Added `ServiceBusClient.list_queue_sessions()` and `ServiceBusClient.list_subscription_sessions()` (sync and async) to list session IDs for entities with active messages, with optional filtering by session-state update timestamp. The methods return an `ItemPaged[str]` (`AsyncItemPaged[str]` on the async client) so callers can iterate every session transparently or page with `by_page()`. Implements the `com.microsoft:get-message-sessions` management operation. ([#46575](https://github.com/Azure/azure-sdk-for-python/pull/46575)) -- Management operations (peek, deferred receive, message settlement over the management link, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs. ### Bugs Fixed @@ -23,6 +22,7 @@ ### Other Changes - When using the async `AmqpOverWebsocket` transport on Python 3.10 or later, `aiohttp>=3.14.0` is now recommended. Earlier `aiohttp` versions have a WebSocket heartbeat bug ([aio-libs/aiohttp#12030](https://github.com/aio-libs/aiohttp/pull/12030)) that can cause the connection to be dropped during long message processing, surfacing as a `SocketError` ("Cannot write to closing transport"). Python 3.9 users must upgrade Python to install an `aiohttp` release containing this fix. ([#44028](https://github.com/Azure/azure-sdk-for-python/issues/44028)) +- Management operations (peek, deferred receive, message settlement over the management link, lock renewal, session state, session listing, schedule/cancel) now send `com.microsoft:server-timeout`: the caller's remaining time less one second, or 60 seconds when none was given. Previously no bound was sent, so a stalled service held the call until the AMQP link failed; it now raises a retryable `OperationTimeoutError`, so a persistently stalled service surfaces after roughly four minutes at default retry settings. Matches the .NET, Java and Go SDKs. ## 7.14.3 (2025-11-11) diff --git a/sdk/servicebus/azure-servicebus/api.metadata.yml b/sdk/servicebus/azure-servicebus/api.metadata.yml index 55a0751fd231..66a9e7205066 100644 --- a/sdk/servicebus/azure-servicebus/api.metadata.yml +++ b/sdk/servicebus/azure-servicebus/api.metadata.yml @@ -1,3 +1,3 @@ apiMdSha256: 68c237729c78165bea132f330a2e89519b2e6ba19373d40c48c55f09e8a994de -parserVersion: 0.3.30 +parserVersion: 0.3.31 pythonVersion: 3.12.10 diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py index 64445f1ccddd..15ec4c26ffd3 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py @@ -106,8 +106,7 @@ def get_server_timeout_ms(timeout: Optional[float]) -> int: if timeout is None: return DEFAULT_SERVER_TIMEOUT_MS capped = min(timeout, MAX_SERVER_TIMEOUT_MS / 1000) - remaining_ms = int(capped * 1000) - SERVER_TIMEOUT_BUFFER_MS - return min(max(remaining_ms, 0), MAX_SERVER_TIMEOUT_MS) + return max(int(capped * 1000) - SERVER_TIMEOUT_BUFFER_MS, 0) def build_uri(address, entity): diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py index 36d5a71278a9..90db62bb6025 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py @@ -17,9 +17,16 @@ import pytest -from azure.servicebus._common.constants import MAX_SERVER_TIMEOUT_MS, REQUEST_RESPONSE_TIMEOUT +from azure.servicebus._common import mgmt_handlers +from azure.servicebus._common.constants import ( + ERROR_CODE_TIMEOUT, + MAX_SERVER_TIMEOUT_MS, + MGMT_RESPONSE_MESSAGE_ERROR_CONDITION, + REQUEST_RESPONSE_TIMEOUT, +) from azure.servicebus._common.utils import get_server_timeout_ms from azure.servicebus._transport._pyamqp_transport import PyamqpTransport +from azure.servicebus.exceptions import OperationTimeoutError class TestServerTimeoutMillis: @@ -87,8 +94,7 @@ def fake_create_mgmt_msg(message, application_properties, config, reply_to, **kw return handler, captured def test_default_sent_when_caller_gave_no_timeout(self): - # The gap this closes: without a caller timeout the service previously received no bound at - # all. + # The gap this closes: previously no bound was sent at all. handler, captured = self._make_handler() handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None) assert captured[REQUEST_RESPONSE_TIMEOUT] == {"TYPE": "UINT", "VALUE": 60000} @@ -110,8 +116,7 @@ def test_associated_link_name_preserved(self): assert REQUEST_RESPONSE_TIMEOUT in captured def test_sent_on_calls_without_an_associated_link(self): - # Operations such as list_sessions pass `keep_alive_associated_link=False` and start from an - # empty property map; they must still be bounded. + # list_sessions passes keep_alive_associated_link=False, starting from an empty map. handler, captured = self._make_handler() handler._mgmt_request_response( b"op", @@ -123,6 +128,86 @@ def test_sent_on_calls_without_an_associated_link(self): assert captured == {REQUEST_RESPONSE_TIMEOUT: {"TYPE": "UINT", "VALUE": 60000}} +class TestServiceTimeoutResponse: + """The response half: the service's answer must surface as a retryable + `OperationTimeoutError`, which rests on `com.microsoft:timeout` in `errorCondition`.""" + + @staticmethod + def _response(condition): + message = MagicMock() + message.application_properties = {MGMT_RESPONSE_MESSAGE_ERROR_CONDITION: condition} + return message + + def test_timeout_condition_raises_retryable_operation_timeout_error(self): + with pytest.raises(OperationTimeoutError) as exc_info: + mgmt_handlers.default(408, self._response(ERROR_CODE_TIMEOUT), "The operation timed out.", PyamqpTransport) + + assert exc_info.value._retryable is True + + def test_success_returns_the_value_untouched(self): + message = self._response(None) + message.value = {"ok": True} + assert mgmt_handlers.default(200, message, None, PyamqpTransport) == {"ok": True} + + def test_uamqp_maps_the_same_condition(self): + uamqp_transport = pytest.importorskip( + "azure.servicebus._transport._uamqp_transport", reason="uamqp not installed" + ) + with pytest.raises(OperationTimeoutError): + mgmt_handlers.default( + 408, + self._response(ERROR_CODE_TIMEOUT), + "The operation timed out.", + uamqp_transport.UamqpTransport, + ) + + +class TestUamqpValueType: + """uamqp is the one place the value type changes: pyamqp uses a plain dict, uamqp an `AMQPuInt`.""" + + def test_transport_supplied_type_is_what_reaches_the_message(self): + + from azure.servicebus._base_handler import BaseHandler + + sentinel = object() + captured = {} + + def fake_create_mgmt_msg(message, application_properties, config, reply_to, **kwargs): + captured.update(application_properties) + return MagicMock() + + handler = BaseHandler.__new__(BaseHandler) + handler._amqp_transport = MagicMock() + handler._amqp_transport.create_mgmt_msg = fake_create_mgmt_msg + handler._amqp_transport.AMQP_UINT_VALUE = lambda ms: sentinel + handler._amqp_transport.get_handler_link_name = lambda h: "link-1" + handler._amqp_transport.mgmt_client_request = lambda *args, **kwargs: "response" + handler._amqp_transport.TIMEOUT_ERROR = TimeoutError + handler._open = lambda: None + handler._handler = MagicMock() + handler._config = MagicMock(encoding="UTF-8") + handler._mgmt_target = "queue/$management" + + handler._mgmt_request_response(b"op", {}, lambda *a: None, timeout=None) + assert captured[REQUEST_RESPONSE_TIMEOUT] is sentinel + + def test_real_uamqp_type_encodes_in_application_properties(self): + uamqp = pytest.importorskip("uamqp", reason="uamqp not installed") + from azure.servicebus._transport._uamqp_transport import UamqpTransport + + value = UamqpTransport.AMQP_UINT_VALUE(get_server_timeout_ms(None)) + assert isinstance(value, uamqp.types.AMQPuInt) + + message = UamqpTransport.create_mgmt_msg( + message={"operation": "peek"}, + application_properties={REQUEST_RESPONSE_TIMEOUT: value}, + config=MagicMock(encoding="UTF-8"), + reply_to="queue/$management", + ) + + assert message.encode_message() + + class TestAsyncParity: """The async management path is a separate implementation and can drift from the sync one.""" From b54c2795945c0fca232ec05fb98b1da0220b8b50 Mon Sep 17 00:00:00 2001 From: Pranjal Patel Date: Fri, 14 Aug 2026 11:05:32 -0700 Subject: [PATCH 4/4] fix(servicebus): skip the uamqp condition test on uamqp, not the transport The test skipped on azure.servicebus._transport._uamqp_transport, but that module imports cleanly without uamqp: UamqpTransport is defined inside the try block that imports it, so the module is present while the class is not. importorskip therefore succeeded and the attribute access raised AttributeError on every CI leg, which do not install uamqp. Skip on uamqp itself and import the class afterwards. Verified both ways: with uamqp installed the test runs, and with uamqp unavailable it skips. Refs: AB#38822503 --- .../tests/unittests/test_server_timeout.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py index 90db62bb6025..7d9c72c1267d 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_server_timeout.py @@ -150,15 +150,16 @@ def test_success_returns_the_value_untouched(self): assert mgmt_handlers.default(200, message, None, PyamqpTransport) == {"ok": True} def test_uamqp_maps_the_same_condition(self): - uamqp_transport = pytest.importorskip( - "azure.servicebus._transport._uamqp_transport", reason="uamqp not installed" - ) + # Skip on uamqp itself: the transport module imports without it, but UamqpTransport is not defined. + pytest.importorskip("uamqp", reason="uamqp not installed") + from azure.servicebus._transport._uamqp_transport import UamqpTransport + with pytest.raises(OperationTimeoutError): mgmt_handlers.default( 408, self._response(ERROR_CODE_TIMEOUT), "The operation timed out.", - uamqp_transport.UamqpTransport, + UamqpTransport, )