fix(eventhubs): safely clean up uAMQP producer sessions after discovery failure

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Fixed uAMQP session destruction after a failed `End`. Native resources and the connection polling reference are released when the last session owner is destroyed; callers must still end sessions before destruction.
+
 - Fixed uAMQP message annotation encoding so messages with annotations, including Event Hubs partition keys, include the required AMQP section descriptor. [[#7426]](https://github.com/Azure/azure-sdk-for-cpp/issues/7426)
 
 ### Other Changes

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/session.hpp

@@ -233,11 +233,17 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
 
     /** @brief Ends operations on the session.
      *
+     * @note With uAMQP, a failed protocol shutdown throws. The native session and its polling
+     * reference are then released when the last session owner is destroyed. Calling End is
+     * still required before destroying a begun session.
      */
     void End(Azure::Core::Context const& context);
 
     /** @brief Ends operations on the session.
      *
+     * @note With uAMQP, a failed protocol shutdown throws. The native session and its polling
+     * reference are then released when the last session owner is destroyed. Calling End is
+     * still required before destroying a begun session.
      */
     void End(
         std::string const& condition_value,

sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/session_impl.hpp

@@ -42,6 +42,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   class SessionImpl final : public std::enable_shared_from_this<SessionImpl> {
   public:
+    // Internal seam for exercising native shutdown failures without a live broker.
+    using EndSession = int (*)(SESSION_HANDLE, char const*, char const*);
+
     SessionImpl(
         std::shared_ptr<_detail::ConnectionImpl> parentConnection,
         _internal::Endpoint& newEndpoint,
@@ -50,7 +53,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     SessionImpl(
         std::shared_ptr<_detail::ConnectionImpl> parentConnection,
         _internal::SessionOptions const& options,
-        _internal::SessionEvents* eventHandler = nullptr);
+        _internal::SessionEvents* eventHandler = nullptr,
+        EndSession endSession = session_end);
     ~SessionImpl() noexcept;
 
     SessionImpl(SessionImpl const&) = delete;
@@ -79,8 +83,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   private:
     SessionImpl();
+    EndSession m_endSession{session_end};
     bool m_connectionAsyncStarted{false};
     bool m_isBegun{false};
+    // A failed protocol shutdown still requires native destruction and a polling release.
+    bool m_endFailed{false};
     std::shared_ptr<_detail::ConnectionImpl> m_connectionToPoll;
     UniqueAmqpSession m_session;
     _internal::SessionOptions m_options;

sdk/core/azure-core-amqp/src/impl/uamqp/amqp/session.cpp

@@ -34,19 +34,23 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   SessionImpl::~SessionImpl() noexcept
   {
-    if (m_isBegun)
+    if (m_isBegun && !m_endFailed)
     {
       AZURE_ASSERT_MSG(false, "Session was not ended before destruction.");
     }
 
-    // If we have a mismatched begin/end pair, we need to stop polling on the connection so it
-    // gets cleaned up properly.
+    // A failed End leaves the native session alive while management clients and links own it.
+    // The last owner reaches here after those objects have gone. session_destroy frees the
+    // endpoint and native resources even if the protocol end frame cannot be sent.
+    {
+      auto lock{m_connectionToPoll->Lock()};
+      m_session.reset();
+    }
+    // Release our polling reference exactly once, outside the connection lock.
     if (m_connectionAsyncStarted)
     {
       m_connectionToPoll->EnableAsyncOperation(false);
     }
-    auto lock{m_connectionToPoll->Lock()};
-    m_session.reset();
   }
 
   SessionImpl::SessionImpl(
@@ -87,8 +91,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
   SessionImpl::SessionImpl(
       std::shared_ptr<_detail::ConnectionImpl> connection,
       _internal::SessionOptions const& options,
-      _internal::SessionEvents* eventHandler)
-      : m_connectionToPoll(connection),
+      _internal::SessionEvents* eventHandler,
+      EndSession endSession)
+      : m_endSession{endSession}, m_connectionToPoll(connection),
         m_session{session_create(*connection, SessionImpl::OnLinkAttachedFn, this)},
         m_options{options}, m_eventHandler{eventHandler}
 
@@ -159,28 +164,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     }
 
     m_isBegun = true;
+    m_endFailed = false;
 
     // Mark the connection as async so that we can use the async APIs.
     GetConnection()->EnableAsyncOperation(true);
     m_connectionAsyncStarted = true;
   }
-  void SessionImpl::End(Azure::Core::Context const&)
-  {
-    if (!m_isBegun)
-    {
-      throw std::runtime_error("Session End without corresponding Begin.");
-    }
-    // When we end the session, it clears all the links, so we need to ensure that the
-    //    m_newLinkAttachedQueue.Clear();
-    if (session_end(m_session.get(), nullptr, nullptr))
-    {
-      throw std::runtime_error("Could not begin session");
-    }
-    // Mark the connection as async so that we can use the async APIs.
-    GetConnection()->EnableAsyncOperation(false);
-    m_connectionAsyncStarted = false;
-    m_isBegun = false;
-  }
+  void SessionImpl::End(Azure::Core::Context const& context) { End({}, {}, context); }
 
   void SessionImpl::End(
       const std::string& condition,
@@ -191,19 +181,20 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     {
       throw std::runtime_error("Session End without corresponding Begin.");
     }
-    // When we end the session, it clears all the links, so we need to ensure that the
-    //    m_newLinkAttachedQueue.Clear();
-    if (session_end(
+    // The native end releases the link endpoints, even if sending the end frame fails.
+    if (m_endSession(
             m_session.get(),
             condition.empty() ? nullptr : condition.c_str(),
             description.empty() ? nullptr : description.c_str()))
     {
-      throw std::runtime_error("Could not begin session");
+      m_endFailed = true;
+      throw std::runtime_error("Could not end session");
     }
-    // Mark the connection as async so that we can use the async APIs.
+    // Release the polling reference acquired by Begin.
     GetConnection()->EnableAsyncOperation(false);
     m_connectionAsyncStarted = false;
     m_isBegun = false;
+    m_endFailed = false;
   }
 
   void SessionImpl::SendDetach(

sdk/core/azure-core-amqp/test/ut/CMakeLists.txt

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+# cspell: words aziotsharedutil
+
 cmake_minimum_required (VERSION 3.13)
 
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
@@ -29,6 +31,7 @@ add_executable(azure-core-amqp-tests
   async_operation_queue_tests.cpp
   azure_core_amqp_tests.cpp
   claim_based_security_tests.cpp
+  cleanup_test_listener.hpp
   connection_string_tests.cpp
   connection_tests.cpp
   global_state_tests.cpp
@@ -37,14 +40,16 @@ add_executable(azure-core-amqp-tests
   message_source_target.cpp
   operation_timeout_tests.cpp
   mock_amqp_server.hpp
+  session_cleanup_tests.cpp
   session_tests.cpp
   amqp_performative_tests.cpp
   ${UAMQP_ONLY_TESTS}
 )
 
 
 if (USE_UAMQP)
-target_include_directories(azure-core-amqp-tests PRIVATE ../../src/amqp/impl/uamqp/amqp/private)
+target_include_directories(azure-core-amqp-tests SYSTEM PRIVATE ${UAMQP_INC_FOLDER})
+target_link_libraries(azure-core-amqp-tests PRIVATE umock_c azure_macro_utils_c aziotsharedutil)
 endif()
 if (USE_RUST_AMQP)
 target_include_directories(azure-core-amqp-tests PRIVATE ../../src/impl/rust_amqp/rust_amqp/rust_wrapper)

sdk/core/azure-core-amqp/test/ut/cleanup_test_listener.hpp

@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#pragma once
+
+#include <azure/core/amqp/internal/common/global_state.hpp>
+#include <azure/core/platform.hpp>
+
+#if defined(AZ_PLATFORM_WINDOWS)
+#include <winsock2.h>
+#else
+#include <unistd.h>
+
+#include <netinet/in.h>
+#include <sys/socket.h>
+#endif
+
+#include <cstdint>
+#include <stdexcept>
+
+namespace Azure { namespace Core { namespace Amqp { namespace Tests {
+  // Allows Begin to connect over TCP without starting a peer AMQP session. Cleanup
+  // fault tests do not need peer callbacks, and must not race a mock server's session teardown.
+  class CleanupTestListener final {
+  public:
+    CleanupTestListener()
+    {
+      Common::_detail::GlobalStateHolder::GlobalStateInstance();
+      m_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+      if (m_socket == InvalidSocket)
+      {
+        throw std::runtime_error("Could not create cleanup test listener");
+      }
+      sockaddr_in address{};
+      address.sin_family = AF_INET;
+      address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+#if defined(AZ_PLATFORM_WINDOWS)
+      int size = sizeof(address);
+#else
+      socklen_t size = sizeof(address);
+#endif
+      if (bind(m_socket, reinterpret_cast<sockaddr*>(&address), sizeof(address)) != 0
+          || listen(m_socket, 1) != 0
+          || getsockname(m_socket, reinterpret_cast<sockaddr*>(&address), &size) != 0)
+      {
+        Close();
+        throw std::runtime_error("Could not listen on cleanup test port");
+      }
+      m_port = ntohs(address.sin_port);
+    }
+
+    ~CleanupTestListener() { Close(); }
+    CleanupTestListener(CleanupTestListener const&) = delete;
+    CleanupTestListener& operator=(CleanupTestListener const&) = delete;
+    std::uint16_t GetPort() const { return m_port; }
+
+  private:
+    using Socket = decltype(socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
+    static constexpr Socket InvalidSocket = static_cast<Socket>(-1);
+    Socket m_socket{InvalidSocket};
+    std::uint16_t m_port{};
+
+    void Close()
+    {
+      if (m_socket != InvalidSocket)
+      {
+#if defined(AZ_PLATFORM_WINDOWS)
+        closesocket(m_socket);
+#else
+        close(m_socket);
+#endif
+        m_socket = InvalidSocket;
+      }
+    }
+  };
+}}}} // namespace Azure::Core::Amqp::Tests