fix(mysql): retry EBADF without TCP keepalive

crates/dbx-core/src/db/mysql.rs

@@ -781,6 +781,51 @@ async fn connect_pool_attempt(
     extra_setup_queries: &[String],
     setup_mode: MySqlSetupMode,
     eof_mode: MySqlEofMode,
+) -> Result<MySqlPool, String> {
+    let result = connect_pool_attempt_with_keepalive(
+        url,
+        ca_cert_path,
+        timeout,
+        max_connections,
+        idle_timeout_secs,
+        setup_database,
+        extra_setup_queries,
+        setup_mode,
+        eof_mode,
+        MySqlTcpKeepaliveMode::Enabled,
+    )
+    .await;
+    if result.as_ref().err().is_some_and(|error| mysql_error_should_retry_without_tcp_keepalive(error)) {
+        log::info!("MySQL connection returned EBADF; retrying with TCP keepalive disabled");
+        return connect_pool_attempt_with_keepalive(
+            url,
+            ca_cert_path,
+            timeout,
+            max_connections,
+            idle_timeout_secs,
+            setup_database,
+            extra_setup_queries,
+            setup_mode,
+            eof_mode,
+            MySqlTcpKeepaliveMode::Disabled,
+        )
+        .await;
+    }
+    result
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn connect_pool_attempt_with_keepalive(
+    url: &str,
+    ca_cert_path: Option<&str>,
+    timeout: Duration,
+    max_connections: usize,
+    idle_timeout_secs: Option<u64>,
+    setup_database: Option<&str>,
+    extra_setup_queries: &[String],
+    setup_mode: MySqlSetupMode,
+    eof_mode: MySqlEofMode,
+    tcp_keepalive_mode: MySqlTcpKeepaliveMode,
 ) -> Result<MySqlPool, String> {
     let pool = create_pool(
         url,
@@ -791,6 +836,7 @@ async fn connect_pool_attempt(
         extra_setup_queries,
         setup_mode,
         eof_mode,
+        tcp_keepalive_mode,
     )?;
     verify_pool_connection_with_setup_fallback(
         pool,
@@ -803,6 +849,7 @@ async fn connect_pool_attempt(
         extra_setup_queries,
         setup_mode,
         eof_mode,
+        tcp_keepalive_mode,
     )
     .await
 }
@@ -825,12 +872,27 @@ enum MySqlEofMode {
     Legacy,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum MySqlTcpKeepaliveMode {
+    Enabled,
+    Disabled,
+}
+
 impl MySqlEofMode {
     fn deprecate_eof(self) -> bool {
         self == Self::Deprecate
     }
 }
 
+impl MySqlTcpKeepaliveMode {
+    fn duration(self) -> Option<Duration> {
+        match self {
+            Self::Enabled => Some(Duration::from_millis(u64::from(MYSQL_TCP_KEEPALIVE_MS))),
+            Self::Disabled => None,
+        }
+    }
+}
+
 const MYSQL_GROUP_CONCAT_MAX_LEN: u64 = 1_048_576;
 
 impl MySqlSetupMode {
@@ -842,6 +904,7 @@ impl MySqlSetupMode {
     }
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn verify_pool_connection_with_setup_fallback(
     pool: MySqlPool,
     timeout: Duration,
@@ -853,6 +916,7 @@ async fn verify_pool_connection_with_setup_fallback(
     extra_setup_queries: &[String],
     setup_mode: MySqlSetupMode,
     eof_mode: MySqlEofMode,
+    tcp_keepalive_mode: MySqlTcpKeepaliveMode,
 ) -> Result<MySqlPool, String> {
     match verify_pool_connection(&pool, timeout).await {
         Ok(()) => Ok(pool),
@@ -872,6 +936,7 @@ async fn verify_pool_connection_with_setup_fallback(
                 extra_setup_queries,
                 fallback_mode,
                 eof_mode,
+                tcp_keepalive_mode,
             )?;
             verify_pool_connection(&fallback_pool, timeout).await.map(|_| fallback_pool)
         }
@@ -914,6 +979,7 @@ fn create_pool(
     extra_setup_queries: &[String],
     setup_mode: MySqlSetupMode,
     eof_mode: MySqlEofMode,
+    tcp_keepalive_mode: MySqlTcpKeepaliveMode,
 ) -> Result<MySqlPool, String> {
     let tls_url = mysql_tls_url(url)?;
     let local_infile_paths = mysql_local_infile_paths(&tls_url.url);
@@ -946,7 +1012,7 @@ fn create_pool(
         .stmt_cache_size(0)
         .prefer_socket(false)
         .pool_opts(Some(pool_opts))
-        .tcp_keepalive(Some(Duration::from_millis(u64::from(MYSQL_TCP_KEEPALIVE_MS))))
+        .tcp_keepalive(tcp_keepalive_mode.duration())
         .deprecate_eof(eof_mode.deprecate_eof())
         .setup(setup_queries);
     if let Some(ssl_opts) = mysql_ssl_opts(base_ssl_opts, url, ca_cert_path, &tls_url.files)? {
@@ -1487,6 +1553,11 @@ fn mysql_error_should_retry_with_legacy_eof(error: &str) -> bool {
     error.to_ascii_lowercase().contains("packets out of sync")
 }
 
+fn mysql_error_should_retry_without_tcp_keepalive(error: &str) -> bool {
+    let error = error.to_ascii_lowercase();
+    error.contains("bad file descriptor") && error.contains("os error 9")
+}
+
 fn mysql_error_should_retry_with_text_protocol(error: &str) -> bool {
     let lower = error.to_ascii_lowercase();
     (lower.contains("1105") && lower.contains("hy000"))
@@ -5397,9 +5468,24 @@ mod tests {
         assert!(!opts.deprecate_eof());
     }
 
+    #[test]
+    fn mysql_bad_file_descriptor_retries_without_tcp_keepalive() {
+        let error = "MySQL connection failed: Input/output error: Input/output error: Bad file descriptor (os error 9)";
+
+        assert!(mysql_error_should_retry_without_tcp_keepalive(error));
+        assert!(!mysql_error_should_retry_without_tcp_keepalive(
+            "MySQL connection failed: Connection reset by peer (os error 54)"
+        ));
+    }
+
     #[test]
     fn mysql_tcp_keepalive_uses_milliseconds_not_seconds() {
         assert_eq!(MYSQL_TCP_KEEPALIVE_MS, 30_000);
+        assert_eq!(
+            MySqlTcpKeepaliveMode::Enabled.duration(),
+            Some(Duration::from_millis(u64::from(MYSQL_TCP_KEEPALIVE_MS)))
+        );
+        assert_eq!(MySqlTcpKeepaliveMode::Disabled.duration(), None);
     }
 
     #[test]