Do not reuse a connection that fail while uploading (#863)

sdk/core/azure-core/CHANGELOG.md

@@ -20,6 +20,7 @@
 ### Bug Fixes
 
 - Prevent pipeline of length zero to be created.
+- Avoid re-using a connection when an uploading request fail when using CurlTransport.
 
 ### New Features
 

sdk/core/azure-core/inc/azure/core/http/curl/curl.hpp

@@ -202,6 +202,22 @@ namespace Azure { namespace Core { namespace Http {
       EndOfHeaders,
     };
 
+    /**
+     * @brief This is used to set the current state of a session.
+     *
+     * @remark The session needs to know what's the state on it when an exception occurs so the
+     * connection is not moved back to the connection pool. When a new request is going to be sent,
+     * the session will be in `PERFORM` until the request has been uploaded and a response code is
+     * received from the server. At that point the state will change to `STREAMING`.
+     * If there is any error before changing the state, the connection need to be cleaned up.
+     *
+     */
+    enum class SessionState
+    {
+      PERFORM,
+      STREAMING
+    };
+
     /**
      * @brief stateful component used to read and parse a buffer to construct a valid HTTP
      * RawResponse.
@@ -337,6 +353,16 @@ namespace Azure { namespace Core { namespace Http {
      */
     curl_socket_t m_curlSocket;
 
+    /**
+     * @brief The current state of the session.
+     *
+     * @remark The state of the session is used to determine if a connection can be moved back to
+     * the connection pool or not. A connection can be re-used only when the session state is
+     * `STREAMING` and the response has been read completely.
+     *
+     */
+    SessionState m_sessionState;
+
     /**
      * @brief unique ptr for the HTTP RawResponse. The session is responsable for creating the
      * response once that an HTTP status line is received.
@@ -350,13 +376,6 @@ namespace Azure { namespace Core { namespace Http {
      */
     Request& m_request;
 
-    /**
-     * @brief Controls the progress of a body buffer upload when using libcurl callbacks. Woks
-     * like an offset to move the pointer to read the body from the HTTP Request on each callback.
-     *
-     */
-    int64_t m_uploadedBytes;
-
     /**
      * @brief Control field to handle the case when part of HTTP response body was copied to the
      * inner buffer. When a libcurl stream tries to read part of the body, this field will help to
@@ -430,22 +449,6 @@ namespace Azure { namespace Core { namespace Http {
      */
     CURLcode SetHeaders();
 
-    /**
-     * @brief Set up libcurl callback functions for writing and user data. User data ptr for all
-     * callbacks is set to reference the session object.
-     *
-     * @return returns the libcurl result after setting up.
-     */
-    CURLcode SetWriteResponse();
-
-    /**
-     * @brief Set up libcurl callback functions for reading and user data. User data ptr for all
-     * callbacks is set to reference the session object.
-     *
-     * @return returns the libcurl result after setting up.
-     */
-    CURLcode SetReadRequest();
-
     /**
      * @brief Function used when working with Streams to manually write from the HTTP Request to
      * the wire.
@@ -522,8 +525,14 @@ namespace Azure { namespace Core { namespace Http {
      */
     bool IsEOF()
     {
-      return this->m_isChunkedResponseType ? this->m_chunkSize == 0
-                                           : this->m_contentLength == this->m_sessionTotalRead;
+      auto eof = this->m_isChunkedResponseType ? this->m_chunkSize == 0
+                                               : this->m_contentLength == this->m_sessionTotalRead;
+
+      // `IsEOF` is called before trying to move a connection back to the connection pool.
+      // If the session state is `PERFORM` it means the request could not complete an upload
+      // operation (might have throw while uploading).
+      // Connection should not be moved back to the connection pool on this scenario.
+      return eof && m_sessionState != SessionState::PERFORM;
     }
 
   public:
@@ -538,7 +547,6 @@ namespace Azure { namespace Core { namespace Http {
       this->m_bodyStartInBuffer = -1;
       this->m_innerBufferSize = Details::c_DefaultLibcurlReaderSize;
       this->m_isChunkedResponseType = false;
-      this->m_uploadedBytes = 0;
       this->m_sessionTotalRead = 0;
     }
 
@@ -549,6 +557,7 @@ namespace Azure { namespace Core { namespace Http {
       // in the wire.
       // By not moving the connection back to the pool, it gets destroyed calling the connection
       // destructor to clean libcurl handle and close the connection.
+      // IsEOF will also handle a connection that fail to complete an upload request.
       if (this->IsEOF())
       {
         CurlConnectionPool::MoveConnectionBackToPool(

sdk/core/azure-core/src/http/curl/curl.cpp

@@ -191,6 +191,9 @@ std::unique_ptr<RawResponse> CurlTransport::Send(Context const& context, Request
 CURLcode CurlSession::Perform(Context const& context)
 {
 
+  // Set the session state
+  m_sessionState = SessionState::PERFORM;
+
   // Get the socket that libcurl is using from handle. Will use this to wait while reading/writing
   // into wire
   auto result = curl_easy_getinfo(
@@ -238,9 +241,11 @@ CURLcode CurlSession::Perform(Context const& context)
   LogThis("Parse server response");
   ReadStatusLineAndHeadersFromRawResponse(context);
 
-  // Upload body for PUT
+  // non-PUT request are ready to be stream at this point. Only PUT request would start an uploading
+  // transfer where we want to maintain the `PERFORM` state.
   if (this->m_request.GetMethod() != HttpMethod::Put)
   {
+    m_sessionState = SessionState::STREAMING;
     return result;
   }
 
@@ -272,6 +277,9 @@ CURLcode CurlSession::Perform(Context const& context)
 
   LogThis("Upload completed. Parse server response");
   ReadStatusLineAndHeadersFromRawResponse(context);
+  // If no throw at this point, the request is ready to stream.
+  // If any throw happened before this point, the state will remain as PERFORM.
+  m_sessionState = SessionState::STREAMING;
   return result;
 }
 
@@ -348,7 +356,6 @@ CURLcode CurlSession::SendBuffer(Context const& context, uint8_t const* buffer,
         case CURLE_OK:
         {
           sentBytesTotal += sentBytesPerRequest;
-          this->m_uploadedBytes += sentBytesPerRequest;
           break;
         }
         case CURLE_AGAIN:
@@ -388,7 +395,6 @@ CURLcode CurlSession::UploadBody(Context const& context)
   // NOTE: if stream is on top a contiguous memory, we can avoid allocating this copying buffer
   auto streamBody = this->m_request.GetBodyStream();
   CURLcode sendResult = CURLE_OK;
-  this->m_uploadedBytes = 0;
 
   int64_t uploadChunkSize = this->m_request.GetUploadChunkSize();
   if (uploadChunkSize <= 0)