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

Author: Pranjal Patel

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
 

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"
 

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,16 +96,18 @@ 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
     :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)
+    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):

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