Bug: Error: IncompleteMessage: connection closed before message completed



### Bug Description

Intermittently few api calls gets closed with the following error `Error: IncompleteMessage: connection closed before message completed`. 

### Expected Behavior

The application should either kept the connection alive or retry connection. 

### Actual Behavior

The application is closing the connection and failing the API call
Detailed hyper create issue: https://github.com/hyperium/hyper/issues/2136

### Steps To Reproduce

Let the application receive more load. For me, for every 1000 transactions, received one failure with this error. 

### Context For The Bug

Not able to talk to connector[basically it can be any outgoing call]. 

### Environment

Are you using hyperswitch hosted version? Yes/No
If yes, please provide the value of the `x-request-id` response header to help us debug your issue.

If not (or if building/running locally), please provide the following details:
1. Operating System : Linux ubuntu
2. Rust version (output of `rustc --version`): `rustc 1.74.0`
3. App version (output of `cargo r --features vergen -- --version`): `router 2024.01.08.0-75-gae83f28-ae83f28-2024-01-23T09:47:32.000000000Z`


### Have you spent some time checking if this bug has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/crates/router/src/routes/metrics.rs b/crates/router/src/routes/metrics.rs
index 6c3293dba9d..a6d588bd43a 100644
--- a/crates/router/src/routes/metrics.rs
+++ b/crates/router/src/routes/metrics.rs
@@ -106,6 +106,7 @@ counter_metric!(APPLE_PAY_MANUAL_FLOW_FAILED_PAYMENT, GLOBAL_METER);
 counter_metric!(APPLE_PAY_SIMPLIFIED_FLOW_FAILED_PAYMENT, GLOBAL_METER);
 
 // Metrics for Auto Retries
+counter_metric!(AUTO_RETRY_CONNECTION_CLOSED, GLOBAL_METER);
 counter_metric!(AUTO_RETRY_ELIGIBLE_REQUEST_COUNT, GLOBAL_METER);
 counter_metric!(AUTO_RETRY_GSM_MISS_COUNT, GLOBAL_METER);
 counter_metric!(AUTO_RETRY_GSM_FETCH_FAILURE_COUNT, GLOBAL_METER);
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 307dff55071..cf231895892 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -552,8 +552,7 @@ pub async fn send_request(
         key: consts::METRICS_HOST_TAG_NAME.into(),
         value: url.host_str().unwrap_or_default().to_string().into(),
     };
-
-    let send_request = async {
+    let request = {
         match request.method {
             Method::Get => client.get(url),
             Method::Post => {
@@ -607,32 +606,92 @@ pub async fn send_request(
         .timeout(Duration::from_secs(
             option_timeout_secs.unwrap_or(crate::consts::REQUEST_TIME_OUT),
         ))
-        .send()
-        .await
-        .map_err(|error| match error {
-            error if error.is_timeout() => {
-                metrics::REQUEST_BUILD_FAILURE.add(&metrics::CONTEXT, 1, &[]);
-                errors::ApiClientError::RequestTimeoutReceived
-            }
-            error if is_connection_closed(&error) => {
-                metrics::REQUEST_BUILD_FAILURE.add(&metrics::CONTEXT, 1, &[]);
-                errors::ApiClientError::ConnectionClosed
-            }
-            _ => errors::ApiClientError::RequestNotSent(error.to_string()),
-        })
-        .into_report()
-        .attach_printable("Unable to send request to connector")
     };
 
-    metrics_request::record_operation_time(
+    // We cannot clone the request type, because it has Form trait which is not clonable. So we are cloning the request builder here.
+    let cloned_send_request = request.try_clone().map(|cloned_request| async {
+        cloned_request
+            .send()
+            .await
+            .map_err(|error| match error {
+                error if error.is_timeout() => {
+                    metrics::REQUEST_BUILD_FAILURE.add(&metrics::CONTEXT, 1, &[]);
+                    errors::ApiClientError::RequestTimeoutReceived
+                }
+                error if is_connection_closed_before_message_could_complete(&error) => {
+                    metrics::REQUEST_BUILD_FAILURE.add(&metrics::CONTEXT, 1, &[]);
+                    errors::ApiClientError::ConnectionClosedIncompleteMessage
+                }
+                _ => errors::ApiClientError::RequestNotSent(error.to_string()),
+            })
+            .into_report()
+            .attach_printable("Unable to send request to connector")
+    });
+
+    let send_request = async {
+        request
+            .send()
+            .await
+            .map_err(|error| match error {
+                error if error.is_timeout() => {
+                    metrics::REQUEST_BUILD_FAILURE.add(&metrics::CONTEXT, 1, &[]);
+                    errors::ApiClientError::RequestTimeoutReceived
+                }
+                error if is_connection_closed_before_message_could_complete(&error) => {
+                    metrics::REQUEST_BUILD_FAILURE.add(&metrics::CONTEXT, 1, &[]);
+                    errors::ApiClientError::ConnectionClosedIncompleteMessage
+                }
+                _ => errors::ApiClientError::RequestNotSent(error.to_string()),
+            })
+            .into_report()
+            .attach_printable("Unable to send request to connector")
+    };
+
+    let response = metrics_request::record_operation_time(
         send_request,
         &metrics::EXTERNAL_REQUEST_TIME,
-        &[metrics_tag],
+        &[metrics_tag.clone()],
     )
-    .await
+    .await;
+    // Retry once if the response is connection closed.
+    //
+    // This is just due to the racy nature of networking.
+    // hyper has a connection pool of idle connections, and it selected one to send your request.
+    // Most of the time, hyper will receive the server’s FIN and drop the dead connection from its pool.
+    // But occasionally, a connection will be selected from the pool
+    // and written to at the same time the server is deciding to close the connection.
+    // Since hyper already wrote some of the request,
+    // it can’t really retry it automatically on a new connection, since the server may have acted already
+    match response {
+        Ok(response) => Ok(response),
+        Err(error)
+            if error.current_context()
+                == &errors::ApiClientError::ConnectionClosedIncompleteMessage =>
+        {
+            metrics::AUTO_RETRY_CONNECTION_CLOSED.add(&metrics::CONTEXT, 1, &[]);
+            match cloned_send_request {
+                Some(cloned_request) => {
+                    logger::info!(
+                        "Retrying request due to connection closed before message could complete"
+                    );
+                    metrics_request::record_operation_time(
+                        cloned_request,
+                        &metrics::EXTERNAL_REQUEST_TIME,
+                        &[metrics_tag],
+                    )
+                    .await
+                }
+                None => {
+                    logger::info!("Retrying request due to connection closed before message could complete failed as request is not clonable");
+                    Err(error)
+                }
+            }
+        }
+        err @ Err(_) => err,
+    }
 }
 
-fn is_connection_closed(error: &reqwest::Error) -> bool {
+fn is_connection_closed_before_message_could_complete(error: &reqwest::Error) -> bool {
     let mut source = error.source();
     while let Some(err) = source {
         if let Some(hyper_err) = err.downcast_ref::<hyper::Error>() {
@@ -1674,7 +1733,7 @@ pub fn build_redirection_form(
 
                     // Initialize the ThreeDSService
                     const threeDS = gateway.get3DSecure();
-            
+
                     const options = {{
                         customerVaultId: '{customer_vault_id}',
                         currency: '{currency}',
diff --git a/crates/storage_impl/src/errors.rs b/crates/storage_impl/src/errors.rs
index ac3a04e85b2..46808a175c3 100644
--- a/crates/storage_impl/src/errors.rs
+++ b/crates/storage_impl/src/errors.rs
@@ -267,7 +267,7 @@ pub enum ApiClientError {
     RequestTimeoutReceived,
 
     #[error("connection closed before a message could complete")]
-    ConnectionClosed,
+    ConnectionClosedIncompleteMessage,
 
     #[error("Server responded with Internal Server Error")]
     InternalServerErrorReceived,
@@ -285,8 +285,8 @@ impl ApiClientError {
     pub fn is_upstream_timeout(&self) -> bool {
         self == &Self::RequestTimeoutReceived
     }
-    pub fn is_connection_closed(&self) -> bool {
-        self == &Self::ConnectionClosed
+    pub fn is_connection_closed_before_message_could_complete(&self) -> bool {
+        self == &Self::ConnectionClosedIncompleteMessage
     }
 }
 

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [x] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
Hyper crate has a bug which allows few calls to connect with a dead connections and fails to send request to the server.  This bug of hyper can be tracked with below link. 
`Error: IncompleteMessage: connection closed before message completed` - [Link](https://github.com/hyperium/hyper/issues/2136)


Temporary fix: Retry connection with the server by sending the request again when call fails with this issue. 

Permanent fix would be, whenever reqwest crate release a stable version with hyper's 1.0 version

[Another Link](https://epi052.github.io/feroxbuster-docs/docs/faq/connection-closed/)
   This is just due to the racy nature of networking. hyper has a connection pool of idle connections, and it selected one to send your request. Most of the time, hyper will receive the server’s FIN and drop the dead connection from its pool. But occasionally, a connection will be selected from the pool and written to at the same time the server is deciding to close the connection. Since hyper already wrote some of the request, it can’t really retry it automatically on a new connection, since the server may have acted already

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `crates/router/src/configs/settings.rs`
-->


## Motivation and Context
Few external api calls fails due to this connection close

## How did you test it?
Running the test case collections

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
