feat(db): classify errors and recycle poisoned pools like Prisma

db::Error now reports a kind that matches PrismaDBExceptionHandler's buckets:
unreachable, capacity, deadlock, serialization failure, lock timeout, read-only
after failover, stale cached plan and rejected data. is_unavailable gates
fail-open and is_requeue_safe says whether a failed write provably never
applied

Pools::recycle_after is the native counterpart of recreating the Prisma
client: after a 25006 (the writer endpoint now points at a read-only node) or
"cached plan must not change result type" (a migration changed a table under
a prepared statement), every pooled connection opened before the error is
dropped on its next acquire, so the pool reconnects and prepares fresh

Native failures reached Python as a ValueError, so allow_requests_on_db_unavailable
treated a native outage as a permanent fault and a write on a dropped connection
was requeued. The bridge now raises NativeDatabaseFailure with the kind and the
Rust answers for is_unavailable and is_requeue_safe, and the two Python gates
return them

litellm/proxy/db/exception_handler.py

@@ -14,6 +14,7 @@ from litellm.proxy._types import (
     ProxyException,
 )
 from litellm.proxy.db.db_lookup_gate import DBLookupDeadlineExceeded
+from litellm.rust_bridge._native import NativeDatabaseFailure
 from litellm.secret_managers.main import str_to_bool
 
 # Bounds the __cause__/__context__ walk in find_database_service_unavailable_error_in_chain.
@@ -175,6 +176,8 @@ class PrismaDBExceptionHandler:
         """
         import prisma.engine.errors
 
+        if isinstance(e, NativeDatabaseFailure):
+            return e.unavailable
         if isinstance(e, (*DB_CONNECTION_ERROR_TYPES, DBLookupDeadlineExceeded)):
             return True
         if isinstance(e, _exception_types(prisma.engine.errors.EngineConnectionError)):
@@ -769,6 +772,8 @@ _DATA_REJECTED_SQLSTATE_CLASSES: Final = frozenset({"22", "23"})
 
 def db_write_failure_is_requeue_safe(e: Exception) -> bool:
     """Whether a failed write provably never applied, so its rows can be sent again without double counting."""
+    if isinstance(e, NativeDatabaseFailure):
+        return e.requeue_safe
     if isinstance(e, DB_CONNECTION_ERROR_TYPES):
         return isinstance(e, DB_RETRY_SAFE_ERROR_TYPES)
     sqlstate: Final = PrismaDBExceptionHandler.postgres_sqlstate(e)

litellm/repositories/database.py

@@ -16,7 +16,7 @@ from litellm.proxy.db.native_database import (
     resolve_native_endpoint,
     writer_token_rotation,
 )
-from litellm.rust_bridge._native import NativeDatabase
+from litellm.rust_bridge._native import NativeDatabase, NativeDatabaseFailure
 
 if TYPE_CHECKING:
     from litellm.proxy.utils import PrismaClient
@@ -43,7 +43,7 @@ def _log_ignored_options(env_var: str, settings: NativeEndpointSettings) -> None
 async def _warn_if_unreachable(native: NativeDatabase) -> None:
     try:
         await native.health_check()
-    except ValueError as error:
+    except NativeDatabaseFailure as error:
         verbose_proxy_logger.warning(
             "Native database pool cannot reach the database, so operations ported to it will fail until it"
             " connects. Prisma is checked separately. Cause: %s",

litellm/rust/crates/db/AGENTS.md

@@ -53,7 +53,9 @@ Python and SQLx run side by side until every repository operation is native, so
 | In-container PgBouncer | `pgbouncer.py` | Inherited: the pooled URL carries PgBouncer's credential |
 | Startup connection budget | `postgres_connection_budget_message` | Counts the Rust writer and reader pools' shares |
 | Read replica routing | `routing_prisma_wrapper.py`, `DATABASE_URL_READ_REPLICA` | `Pools`: callers pick `writer()`, `read()` or `read_after_write()`. `read()` prefers the reader and `read_after_write()` the writer (`WriterPinnedClient`); either falls back to the other side when its pool cannot connect and skips that side for 30s, so reads keep working on the reader while the writer is down. Writes have no fallback |
-| Error classification, `lock_timeout` on rollups | `exception_handler.py`, `rollup_lock_timeout.py` | Planned |
+| `lock_timeout` on rollups | `rollup_lock_timeout.py` | Planned |
+| Error classification | `exception_handler.py` | `Error::kind`, `is_unavailable`, `is_requeue_safe`. The bridge raises `NativeDatabaseFailure` with the kind and both answers, and `is_database_connection_error` and `db_write_failure_is_requeue_safe` return them instead of classifying the error again |
+| Recreate the client after 25006 (writer became read-only) or a stale cached plan | `exception_handler.py`, `_query_first_with_cached_plan_fallback` | `Pools::recycle_after` drops every pooled connection opened before the error, so the next acquire reconnects and prepares fresh. Callers retry a `StalePlan` once |
 | `ServiceTypes.DB` events per round trip | `db_span.py`, `log_db_metrics.py`, `prometheus_services.py`, `otel/model/spans.py` | Python host only. Wrap each native operation in `observe` with the call type and table its Prisma version uses, and the bridge forwards events to `ServiceLoggingEvents`. Nesting under an outer Python `db_span` is not detected. A Rust host has no sink yet and must emit the names below |
 | Auth lookup gate, fail open on DB outage, reader choice per query | `db_lookup_gate.py`, `allow_requests_on_db_unavailable` | With each ported auth read |
 | Connect at boot, so a database that cannot be reached is reported before traffic | `PrismaClient.connect` | The pool stays lazy and never fails boot. `Database.open` pings it once in the background and logs a warning when it cannot connect, so a mistranslated endpoint is not first found by a ported query |

litellm/rust/crates/db/src/endpoint.rs

@@ -105,13 +105,18 @@ impl Endpoint {
         }
     }
 
-    pub fn pool(&self) -> Result<PgPool, Error> {
+    pub fn pool_options(&self) -> Result<PgPoolOptions, Error> {
         let limits = self.pool;
         Ok(PgPoolOptions::new()
             .min_connections(0)
             .max_connections(limits.max_connections)
             .acquire_timeout(seconds(limits.acquire_timeout_seconds)?.unwrap_or(WAIT_FOREVER))
-            .idle_timeout(seconds(limits.idle_timeout_seconds)?)
+            .idle_timeout(seconds(limits.idle_timeout_seconds)?))
+    }
+
+    pub fn pool(&self) -> Result<PgPool, Error> {
+        Ok(self
+            .pool_options()?
             .connect_lazy_with(self.connect_options()))
     }
 

litellm/rust/crates/db/src/error.rs

@@ -10,21 +10,76 @@ pub enum Error {
     InvalidDuration(#[from] std::time::TryFromFloatSecsError),
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+pub enum ErrorKind {
+    Unreachable,
+    Capacity,
+    Deadlock,
+    SerializationFailure,
+    LockTimeout,
+    ReadOnly,
+    StalePlan,
+    DataRejected,
+    Other,
+}
+
 impl Error {
-    pub fn is_retryable(&self) -> bool {
+    pub fn kind(&self) -> ErrorKind {
         match self {
+            Self::Query(sqlx::Error::Io(_)) => ErrorKind::Unreachable,
+            Self::Query(sqlx::Error::PoolTimedOut) => ErrorKind::Capacity,
             Self::Query(sqlx::Error::Database(error)) => {
-                matches!(
-                    error.code().as_deref(),
-                    Some("40P01" | "55P03" | "40001" | "53300")
-                )
+                sqlstate_kind(error.code().as_deref(), error.message())
             }
-            Self::Query(sqlx::Error::Io(_) | sqlx::Error::PoolTimedOut) => true,
+            _ => ErrorKind::Other,
+        }
+    }
+
+    pub fn is_retryable(&self) -> bool {
+        matches!(
+            self.kind(),
+            ErrorKind::Unreachable
+                | ErrorKind::Capacity
+                | ErrorKind::Deadlock
+                | ErrorKind::SerializationFailure
+                | ErrorKind::LockTimeout
+                | ErrorKind::StalePlan
+        )
+    }
+
+    pub fn needs_fresh_connections(&self) -> bool {
+        matches!(self.kind(), ErrorKind::ReadOnly | ErrorKind::StalePlan)
+    }
+
+    pub fn is_unavailable(&self) -> bool {
+        matches!(self.kind(), ErrorKind::Unreachable | ErrorKind::Capacity)
+    }
+
+    pub fn is_requeue_safe(&self) -> bool {
+        match self {
+            Self::Query(sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed) => true,
+            Self::Query(sqlx::Error::Database(_)) => self.kind() != ErrorKind::DataRejected,
             _ => false,
         }
     }
 }
 
+fn sqlstate_kind(code: Option<&str>, message: &str) -> ErrorKind {
+    match code {
+        Some("0A000") if message.contains("cached plan must not change result type") => {
+            ErrorKind::StalePlan
+        }
+        Some("53300") => ErrorKind::Capacity,
+        Some("40P01") => ErrorKind::Deadlock,
+        Some("40001") => ErrorKind::SerializationFailure,
+        Some("55P03") => ErrorKind::LockTimeout,
+        Some("25006") => ErrorKind::ReadOnly,
+        Some(code) if code.starts_with("22") || code.starts_with("23") => ErrorKind::DataRejected,
+        _ => ErrorKind::Other,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::error::Error as StdError;
@@ -32,29 +87,59 @@ mod tests {
     use rstest::rstest;
     use sqlx::PgPool;
 
-    use super::Error;
+    use super::{Error, ErrorKind};
     use crate::test_support::database_pool;
 
     #[rstest]
-    #[case::closed_pool(sqlx::Error::PoolClosed, false)]
-    #[case::pool_timeout(sqlx::Error::PoolTimedOut, true)]
-    #[case::io(sqlx::Error::Io(std::io::ErrorKind::ConnectionReset.into()), true)]
-    #[case::missing_row(sqlx::Error::RowNotFound, false)]
-    fn classifies_client_errors(#[case] error: sqlx::Error, #[case] retryable: bool) {
-        assert_eq!(Error::from(error).is_retryable(), retryable);
+    #[case::closed_pool(sqlx::Error::PoolClosed, ErrorKind::Other, false, false, true)]
+    #[case::pool_timeout(sqlx::Error::PoolTimedOut, ErrorKind::Capacity, true, true, true)]
+    #[case::io(
+        sqlx::Error::Io(std::io::ErrorKind::ConnectionReset.into()),
+        ErrorKind::Unreachable,
+        true,
+        true,
+        false
+    )]
+    #[case::missing_row(sqlx::Error::RowNotFound, ErrorKind::Other, false, false, false)]
+    fn classifies_client_errors(
+        #[case] error: sqlx::Error,
+        #[case] kind: ErrorKind,
+        #[case] retryable: bool,
+        #[case] unavailable: bool,
+        #[case] requeue_safe: bool,
+    ) {
+        let error = Error::from(error);
+
+        assert_eq!(error.kind(), kind);
+        assert_eq!(error.is_retryable(), retryable);
+        assert_eq!(error.is_unavailable(), unavailable);
+        assert_eq!(error.is_requeue_safe(), requeue_safe);
     }
 
     #[rstest]
-    #[case::connection_capacity("53300", true)]
-    #[case::deadlock("40P01", true)]
-    #[case::lock_unavailable("55P03", true)]
-    #[case::serialization_failure("40001", true)]
-    #[case::unique_violation("23505", false)]
+    #[case::connection_capacity("53300", ErrorKind::Capacity, true, true, true)]
+    #[case::deadlock("40P01", ErrorKind::Deadlock, true, false, true)]
+    #[case::lock_unavailable("55P03", ErrorKind::LockTimeout, true, false, true)]
+    #[case::serialization_failure("40001", ErrorKind::SerializationFailure, true, false, true)]
+    #[case::read_only_after_failover("25006", ErrorKind::ReadOnly, false, false, true)]
+    #[case::unique_violation("23505", ErrorKind::DataRejected, false, false, false)]
+    #[case::numeric_overflow("22003", ErrorKind::DataRejected, false, false, false)]
+    #[case::undefined_table("42P01", ErrorKind::Other, false, false, true)]
+    #[case::feature_not_supported_without_a_stale_plan(
+        "0A000",
+        ErrorKind::Other,
+        false,
+        false,
+        true
+    )]
     #[tokio::test]
     #[ignore = "requires DATABASE_URL pointing to a Prisma-migrated PostgreSQL database"]
     async fn classifies_server_errors(
         #[case] code: &str,
+        #[case] kind: ErrorKind,
         #[case] retryable: bool,
+        #[case] unavailable: bool,
+        #[case] requeue_safe: bool,
         #[future(awt)] database_pool: Result<PgPool, Box<dyn StdError>>,
     ) -> Result<(), Box<dyn StdError>> {
         let pool = database_pool?;
@@ -74,7 +159,11 @@ mod tests {
             error.as_database_error().and_then(|error| error.code()),
             Some(code.into())
         );
-        assert_eq!(Error::from(error).is_retryable(), retryable);
+        let error = Error::from(error);
+        assert_eq!(error.kind(), kind);
+        assert_eq!(error.is_retryable(), retryable);
+        assert_eq!(error.is_unavailable(), unavailable);
+        assert_eq!(error.is_requeue_safe(), requeue_safe);
         tx.rollback().await?;
         Ok(())
     }

litellm/rust/crates/db/src/lib.rs

@@ -24,7 +24,7 @@ pub use daily_spend::{DailyEndUserRow, DailyTeamRow, DailyUserRow};
 pub use deadline::within;
 pub use end_user::{EndUser, RateLimits, find_end_user};
 pub use endpoint::{ClientIdentity, Endpoint, PoolLimits, SslMode};
-pub use error::Error;
+pub use error::{Error, ErrorKind};
 pub use failure::DbFailure;
 pub use health::health_check;
 pub use observe::{Call, Observer, Outcome, QueryEvent, observe};

litellm/rust/crates/db/src/pools.rs

@@ -1,4 +1,7 @@
-use std::time::{Duration, Instant};
+use std::{
+    sync::{Arc, Mutex, PoisonError},
+    time::{Duration, Instant},
+};
 
 use litellm_circuit_breaker::{Breaker, Config, Failure};
 use sqlx::{PgPool, Postgres, pool::PoolConnection};
@@ -9,16 +12,46 @@ const RECOVERY: Duration = Duration::from_secs(30);
 
 type Connection = PoolConnection<Postgres>;
 
+#[derive(Default)]
+struct Recycled(Mutex<Option<Instant>>);
+
+impl Recycled {
+    fn mark(&self, now: Instant) {
+        *self.0.lock().unwrap_or_else(PoisonError::into_inner) = Some(now);
+    }
+
+    fn predates(&self, now: Instant, age: Duration) -> bool {
+        let Some(recycled_at) = *self.0.lock().unwrap_or_else(PoisonError::into_inner) else {
+            return false;
+        };
+        now.checked_sub(age)
+            .is_none_or(|opened_at| opened_at <= recycled_at)
+    }
+}
+
 #[derive(Clone)]
 struct Side {
     pool: PgPool,
     breaker: Breaker,
+    recycled: Arc<Recycled>,
 }
 
 impl Side {
     fn new(endpoint: &Endpoint) -> Result<Self, Error> {
+        let recycled = Arc::new(Recycled::default());
+        let pool = endpoint
+            .pool_options()?
+            .before_acquire({
+                let recycled = Arc::clone(&recycled);
+                move |_, metadata| {
+                    let fresh = !recycled.predates(Instant::now(), metadata.age);
+                    Box::pin(async move { Ok(fresh) })
+                }
+            })
+            .connect_lazy_with(endpoint.connect_options());
         Ok(Self {
-            pool: endpoint.pool()?,
+            pool,
+            recycled,
             breaker: Breaker::new(
                 Config {
                     failure_threshold: 1,
@@ -99,6 +132,18 @@ impl Pools {
         acquire_preferring(&self.writer, self.reader.as_ref()).await
     }
 
+    pub fn recycle_after(&self, error: &Error) -> bool {
+        if !error.needs_fresh_connections() {
+            return false;
+        }
+        let now = Instant::now();
+        self.writer.recycled.mark(now);
+        if let Some(reader) = &self.reader {
+            reader.recycled.mark(now);
+        }
+        true
+    }
+
     pub async fn close(&self) {
         self.writer.pool.close().await;
         if let Some(reader) = &self.reader {
@@ -109,12 +154,124 @@ impl Pools {
 
 #[cfg(test)]
 mod tests {
-    use std::{error::Error as StdError, time::Instant};
+    use std::{
+        error::Error as StdError,
+        time::{Duration, Instant},
+    };
 
-    use sqlx::PgConnection;
+    use rstest::rstest;
+    use sqlx::{Connection as _, PgConnection};
 
-    use super::Pools;
-    use crate::{Endpoint, PoolLimits, test_support::database_endpoint};
+    use super::{Pools, Recycled};
+    use crate::{Endpoint, Error, ErrorKind, PoolLimits, test_support::database_endpoint};
+
+    #[rstest]
+    #[case::never_recycled(None, 5, false)]
+    #[case::opened_before_the_recycle(Some(10), 30, true)]
+    #[case::opened_after_the_recycle(Some(10), 5, false)]
+    fn discards_connections_opened_before_a_recycle(
+        #[case] recycled_seconds_ago: Option<u64>,
+        #[case] age_seconds: u64,
+        #[case] stale: bool,
+    ) {
+        let now = Instant::now() + Duration::from_secs(60);
+        let recycled = Recycled::default();
+        if let Some(ago) = recycled_seconds_ago {
+            recycled.mark(now - Duration::from_secs(ago));
+        }
+
+        assert_eq!(
+            recycled.predates(now, Duration::from_secs(age_seconds)),
+            stale
+        );
+    }
+
+    async fn backend_pid(pools: &Pools) -> Result<i32, sqlx::Error> {
+        #[expect(clippy::disallowed_methods, reason = "A probe query needs no schema")]
+        sqlx::query_scalar("SELECT pg_backend_pid()")
+            .fetch_one(&mut *pools.writer().acquire().await?)
+            .await
+    }
+
+    async fn raised(pools: &Pools, code: &str) -> Result<Error, sqlx::Error> {
+        #[expect(
+            clippy::disallowed_methods,
+            reason = "Test query must raise the requested SQLSTATE"
+        )]
+        let error = sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DO $$ BEGIN RAISE EXCEPTION USING ERRCODE = '{code}'; END $$"
+        )))
+        .execute(pools.writer())
+        .await
+        .expect_err("PostgreSQL must raise the requested error");
+        Ok(Error::from(error))
+    }
+
+    #[rstest]
+    #[case::read_only_writer("25006", true)]
+    #[case::lock_timeout("55P03", false)]
+    #[tokio::test]
+    #[ignore = "requires DATABASE_URL pointing to a Prisma-migrated PostgreSQL database"]
+    async fn recycles_pooled_connections_only_after_errors_that_poison_them(
+        #[case] code: &str,
+        #[case] recycled: bool,
+    ) -> Result<(), Box<dyn StdError>> {
+        let pools = Pools::new(&database_endpoint()?, None)?;
+        let before = backend_pid(&pools).await?;
+
+        assert_eq!(pools.recycle_after(&raised(&pools, code).await?), recycled);
+
+        assert_eq!(backend_pid(&pools).await? != before, recycled);
+        Ok(())
+    }
+
+    #[tokio::test]
+    #[ignore = "requires DATABASE_URL pointing to a Prisma-migrated PostgreSQL database"]
+    async fn a_schema_change_under_a_cached_plan_is_recovered_by_a_recycle()
+    -> Result<(), Box<dyn StdError>> {
+        let endpoint = database_endpoint()?;
+        let pools = Pools::new(
+            &Endpoint {
+                pool: PoolLimits {
+                    max_connections: 1,
+                    ..endpoint.pool
+                },
+                ..endpoint.clone()
+            },
+            None,
+        )?;
+        let table = format!("native_stale_plan_probe_{}", std::process::id());
+        let select = format!("SELECT * FROM {table}");
+        #[expect(clippy::disallowed_methods, reason = "A scratch table needs no schema")]
+        let run = |sql: &str| sqlx::query(sqlx::AssertSqlSafe(sql.to_owned()));
+        let mut admin = PgConnection::connect_with(&endpoint.connect_options()).await?;
+        run(&format!("CREATE TABLE {table} (id int)"))
+            .execute(&mut admin)
+            .await?;
+
+        let outcome = async {
+            run(&select).fetch_all(pools.writer()).await?;
+            run(&format!("ALTER TABLE {table} ADD COLUMN extra int"))
+                .execute(&mut admin)
+                .await?;
+            let stale = Error::from(
+                run(&select)
+                    .fetch_all(pools.writer())
+                    .await
+                    .expect_err("the cached plan must be stale"),
+            );
+            let recycled = pools.recycle_after(&stale);
+            run(&select).fetch_all(pools.writer()).await?;
+            Ok::<_, Box<dyn StdError>>((stale.kind(), recycled))
+        }
+        .await;
+        run(&format!("DROP TABLE {table}"))
+            .execute(&mut admin)
+            .await?;
+
+        assert_eq!(outcome?, (ErrorKind::StalePlan, true));
+        Ok(())
+    }
 
     fn read_only(endpoint: &Endpoint) -> Endpoint {
         Endpoint {

litellm/rust/crates/python-bridge/src/db.rs

@@ -50,8 +50,26 @@ pub(crate) struct NativeDatabase {
     events: Arc<dyn Observer>,
 }
 
+pyo3::create_exception!(
+    _native,
+    NativeDatabaseFailure,
+    pyo3::exceptions::PyException,
+    "A native database operation failed. The kind attribute is litellm-db's ErrorKind, and unavailable and requeue_safe are the error's is_unavailable and is_requeue_safe."
+);
+
 fn map_error(error: litellm_db::Error) -> PyErr {
-    PyValueError::new_err(litellm_error::Report::new(&error).to_string())
+    let kind: &'static str = error.kind().into();
+    let failure = NativeDatabaseFailure::new_err(litellm_error::Report::new(&error).to_string());
+    let decided = Python::attach(|py| {
+        let value = failure.value(py);
+        value.setattr("kind", kind)?;
+        value.setattr("unavailable", error.is_unavailable())?;
+        value.setattr("requeue_safe", error.is_requeue_safe())
+    });
+    match decided {
+        Ok(()) => failure,
+        Err(error) => error,
+    }
 }
 
 #[pymethods]

litellm/rust/crates/python-bridge/src/lib.rs

@@ -26,7 +26,7 @@ mod _native {
     use pyo3::{prelude::*, types::PyModule};
 
     #[pymodule_export]
-    use crate::db::NativeDatabase;
+    use crate::db::{NativeDatabase, NativeDatabaseFailure};
     #[cfg(feature = "panic-test")]
     #[pymodule_export]
     use crate::diagnostics::_panic_for_test;
@@ -180,6 +180,7 @@ mod tests {
                 "NativeClickHouseSpendConfig",
                 "NativeClickHouseSpendStorage",
                 "NativeDatabase",
+                "NativeDatabaseFailure",
                 "trace_encode_error",
                 "bundled_model_cost_map",
                 "NativeModelCostMap",

litellm/rust_bridge/_native.pyi

@@ -20,6 +20,11 @@ class ForkedAfterNativeRuntimeStarted(RuntimeError): ...
 class ProcessReservedForForking(RuntimeError): ...
 class NativeRedisStoreError(Exception): ...
 
+class NativeDatabaseFailure(Exception):
+    kind: str
+    unavailable: bool
+    requeue_safe: bool
+
 def trace_encode_error(message: str) -> bytes: ...
 def bundled_model_cost_map() -> bytes: ...
 def native_redis_breaker_states() -> list[str]: ...
@@ -763,6 +768,7 @@ __all__ = [
     "NativeClickHouseSpendStorage",
     "NativeCostMapSource",
     "NativeDatabase",
+    "NativeDatabaseFailure",
     "NativeDeploymentHealth",
     "NativeDiagnosticProcessor",
     "NativeFetchFailed",

tests/unit/proxy/db/test_native_database.py

@@ -14,6 +14,7 @@ from cryptography.x509.oid import NameOID
 from pydantic import TypeAdapter
 
 from litellm._service_logger import ServiceLogging, ServiceTypes
+from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler, db_write_failure_is_requeue_safe
 from litellm.proxy.db.native_database import (
     ServiceLoggingEvents,
     TokenRotation,
@@ -26,7 +27,7 @@ from litellm.proxy.db.native_database import (
 )
 from litellm.proxy.db.pgbouncer import PGBOUNCER_POOLED_ENV_VAR
 from litellm.proxy.db.token_auth import IAMEndpoint, RdsIamTokenAuth
-from litellm.rust_bridge._native import NativeDatabase
+from litellm.rust_bridge._native import NativeDatabase, NativeDatabaseFailure
 from litellm.rust_bridge.database import NativeDatabaseEndpoint, NativePoolLimits
 
 NOW: Final = datetime(2026, 11, 3, 12, 0, 0)
@@ -467,7 +468,7 @@ async def test_native_health_check_failure_reports_a_db_event() -> None:
         ).endpoint,
     )
 
-    with pytest.raises(ValueError, match="database query failed"):
+    with pytest.raises(NativeDatabaseFailure, match="database query failed"):
         await native.health_check()
     await asyncio.wait_for(logging.arrived.wait(), timeout=5)
     await native.close()
@@ -477,3 +478,49 @@ async def test_native_health_check_failure_reports_a_db_event() -> None:
     assert (service, call_type, metadata) == ("postgres", "health_check", {})
     assert duration >= 0.9
     assert failure is not None and "database query failed" in failure
+
+
+async def _native_failure(url: str) -> NativeDatabaseFailure:
+    native: Final = await NativeDatabase.connect(
+        ServiceLoggingEvents(loop=asyncio.get_running_loop(), service_logging=lambda: None),
+        native_endpoint(url, tls_offered=True, max_connections=2).endpoint,
+    )
+    try:
+        with pytest.raises(NativeDatabaseFailure) as raised:
+            await native.health_check()
+    finally:
+        await native.close()
+    return raised.value
+
+
+async def _drop_connection(_: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    writer.close()
+
+
+@pytest.mark.asyncio
+async def test_native_outage_passes_the_db_unavailable_gate_and_is_requeue_safe() -> None:
+    error: Final = await _native_failure("postgresql://litellm:secret@127.0.0.1:1/proxy?pool_timeout=1")
+
+    assert (error.kind, str(error)) == (
+        "capacity",
+        "database query failed: pool timed out while waiting for an open connection",
+    )
+    assert PrismaDBExceptionHandler.is_database_connection_error(error)
+    assert db_write_failure_is_requeue_safe(error)
+
+
+@pytest.mark.asyncio
+async def test_native_dropped_connection_is_unavailable_but_not_requeue_safe() -> None:
+    server: Final = await asyncio.start_server(_drop_connection, "127.0.0.1", 0)
+    _, port = _ADDRESS.validate_python(server.sockets[0].getsockname())
+    try:
+        error: Final = await _native_failure(
+            f"postgresql://litellm:secret@127.0.0.1:{port}/proxy?pool_timeout=5&sslmode=disable"
+        )
+    finally:
+        server.close()
+        await server.wait_closed()
+
+    assert error.kind == "unreachable"
+    assert PrismaDBExceptionHandler.is_database_connection_error(error)
+    assert not db_write_failure_is_requeue_safe(error)