[SB] Link-Detach (#34820)

* if it isnt a user started detach

* set error instead of raise + comments

* fix mypy issue for self._error

* fix pylint

* change error condition

* fix comment

---------

Co-authored-by: Kashif Khan <[email protected]>

Author: l0lawrence

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_link_async.py

@@ -100,7 +100,7 @@ def __init__(
         self._is_closed = False
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
-        self._error = None
+        self._error: Optional[AMQPLinkError] = None
 
     async def __aenter__(self) -> "Link":
         await self.attach()
@@ -244,6 +244,11 @@ async def _incoming_detach(self, frame) -> None:
             self._error = error_cls(condition=frame[2][0], description=frame[2][1], info=frame[2][2])
             await self._set_state(LinkState.ERROR)
         else:
+            if self.state != LinkState.DETACH_SENT:
+                # Handle the case of when the remote side detaches without sending an error.
+                # We should detach as per the spec but then retry connecting
+                self._error = AMQPLinkError(condition=ErrorCondition.UnknownError,
+                                          description="Link detached unexpectedly.", retryable=True)
             await self._set_state(LinkState.DETACHED)
 
     async def attach(self) -> None:

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/link.py

@@ -96,7 +96,7 @@ def __init__(
         self._is_closed = False
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
-        self._error = None
+        self._error: Optional[AMQPLinkError] = None
 
     def __enter__(self) -> "Link":
         self.attach()
@@ -239,6 +239,11 @@ def _incoming_detach(self, frame) -> None:
             self._error = error_cls(condition=frame[2][0], description=frame[2][1], info=frame[2][2])
             self._set_state(LinkState.ERROR)
         else:
+            if self.state != LinkState.DETACH_SENT:
+                # Handle the case of when the remote side detaches without sending an error.
+                # We should detach as per the spec but then retry connecting
+                self._error = AMQPLinkError(condition=ErrorCondition.UnknownError,
+                                          description="Link detached unexpectedly.", retryable=True)
             self._set_state(LinkState.DETACHED)
 
     def attach(self) -> None: