fix(targets): harden Postgres/MySQL SQL backends (pool timeouts, error classification, idempotency) (#4500)

fix(targets): harden Postgres/MySQL SQL backends (pools, error class, idempotency)

Postgres (postgres.rs):
- Add deadpool wait/create/recycle timeouts + tokio-postgres connect_timeout,
  and wrap every pool.get() in a Tokio hard-limit timeout so an unreachable
  broker/DB can no longer block send_body/probe_table/is_active forever;
  checkout timeouts map to TargetError::Timeout to trigger store replay.
- namespace format now DELETEs the row on s3:ObjectRemoved:* events instead of
  UPSERTing, keeping the table consistent with the object lifecycle.
- map_pg_error: SQLSTATE class 40 (serialization_failure 40001,
  deadlock_detected 40P01, transaction rollback) is now transient (Timeout,
  retryable) instead of a permanent Request. Extracted map_pg_sqlstate for
  unit-testable classification.

MySQL (mysql.rs):
- Wrap pool.get_conn() in a Tokio checkout timeout across insert/init/liveness
  paths so an unreachable server cannot block the delivery thread.
- Add an event_id idempotency key: tables created by the target now carry an
  event_id VARCHAR(255) PRIMARY KEY, inserts use ON DUPLICATE KEY UPDATE, and
  replays reuse the stable store key so a lost-ack replay no longer appends a
  duplicate audit row. Legacy two-column tables are detected and fall back to
  the non-idempotent insert with a warning (backward compatible).
- redact_mysql_dsn splits on the last '@' so a password containing '@' no
  longer leaks its tail into Debug output.
- Disconnect the stale pool before dropping it on inline TLS reload instead of
  leaking its connections.
- Cache TLS file mtimes and only recompute the inline fingerprint when a cert
  file changes, avoiding a 3-file read+hash on every checkout.

Adds unit tests for SQLSTATE classification, namespace delete SQL, removal-event
detection, DSN redaction with '@' in the password, and the MySQL insert/DDL
builders.

Relates to rustfs/backlog#976
Relates to rustfs/backlog#973
Relates to rustfs/backlog#983

Co-authored-by: heihutu <heihutu@gmail.com>

crates/targets/src/target/mysql.rs

@@ -36,8 +36,98 @@ use std::fmt;
 use std::marker::PhantomData;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, SystemTime};
 use tokio::sync::Mutex;
 use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+/// Bounds `pool.get_conn()` so an unreachable MySQL server (or an exhausted
+/// pool) cannot block the delivery thread indefinitely. A timeout maps to
+/// `TargetError::Timeout`, a connectivity error, so the payload stays queued
+/// for replay.
+const MYSQL_CONN_CHECKOUT_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// Name of the optional idempotency-key column / primary key. Present on tables
+/// created by this target; absent on legacy two-column tables.
+const MYSQL_EVENT_ID_COLUMN: &str = "event_id";
+
+/// Modification timestamps of the three TLS material files, used to avoid
+/// re-reading and re-hashing certificate files on every pool checkout.
+///
+/// The inline TLS fingerprint is only recomputed when one of these mtimes
+/// changes, which still catches on-disk rotation while eliminating the
+/// per-send file reads.
+#[derive(Clone, PartialEq, Eq)]
+struct TlsFileMtimes {
+    ca: Option<SystemTime>,
+    client_cert: Option<SystemTime>,
+    client_key: Option<SystemTime>,
+}
+
+impl TlsFileMtimes {
+    /// Reads the current mtimes of the configured TLS files. A missing/empty
+    /// path yields `None`; an unreadable path also yields `None`, which is
+    /// treated conservatively as "changed" so the fingerprint is recomputed.
+    fn read(args: &MySqlArgs) -> Self {
+        fn mtime(path: &str) -> Option<SystemTime> {
+            if path.is_empty() {
+                return None;
+            }
+            std::fs::metadata(path).and_then(|m| m.modified()).ok()
+        }
+        TlsFileMtimes {
+            ca: mtime(&args.tls_ca),
+            client_cert: mtime(&args.tls_client_cert),
+            client_key: mtime(&args.tls_client_key),
+        }
+    }
+}
+
+/// Checks out a connection from the pool under a Tokio timeout.
+///
+/// `get_conn()` failures are always transient here (connection lost or pool
+/// temporarily exhausted), so both an error and a timeout map to a
+/// connectivity error that keeps the payload queued for replay.
+async fn checkout_conn(pool: &Pool) -> Result<Conn, TargetError> {
+    match tokio::time::timeout(MYSQL_CONN_CHECKOUT_TIMEOUT, pool.get_conn()).await {
+        Ok(Ok(conn)) => Ok(conn),
+        Ok(Err(_)) => Err(TargetError::NotConnected),
+        Err(_) => Err(TargetError::Timeout(format!(
+            "MySQL connection checkout timed out after {}s",
+            MYSQL_CONN_CHECKOUT_TIMEOUT.as_secs()
+        ))),
+    }
+}
+
+/// INSERT for tables that carry the `event_id` idempotency key. Replays of the
+/// same physical event share the same key, so `ON DUPLICATE KEY UPDATE` makes
+/// the write a no-op instead of appending a duplicate audit row.
+pub(crate) fn mysql_insert_sql_with_event_id(quoted_table: &str) -> String {
+    format!(
+        "INSERT INTO {quoted_table} ({MYSQL_EVENT_ID_COLUMN}, event_time, event_data) \
+         VALUES (?, ?, CAST(? AS JSON)) \
+         ON DUPLICATE KEY UPDATE {MYSQL_EVENT_ID_COLUMN} = {MYSQL_EVENT_ID_COLUMN}"
+    )
+}
+
+/// Legacy INSERT for pre-existing two-column tables that lack the `event_id`
+/// key. Idempotency is not available in this mode (replays may duplicate).
+pub(crate) fn mysql_insert_sql_legacy(quoted_table: &str) -> String {
+    format!("INSERT INTO {quoted_table} (event_time, event_data) VALUES (?, CAST(? AS JSON))")
+}
+
+/// DDL used to create the target table. Tables created here carry the
+/// `event_id` primary key so that store replays are idempotent.
+pub(crate) fn mysql_create_table_sql(quoted_table: &str) -> String {
+    format!(
+        "CREATE TABLE IF NOT EXISTS {quoted_table} (\
+         {MYSQL_EVENT_ID_COLUMN} VARCHAR(255) NOT NULL, \
+         event_time DATETIME(6) NOT NULL, \
+         event_data JSON NOT NULL, \
+         PRIMARY KEY ({MYSQL_EVENT_ID_COLUMN}))"
+    )
+}
 
 /// Arguments for configuring a MySQL notification target.
 ///
@@ -296,6 +386,11 @@ fn split_mysql_scheme(input: &str) -> (&str, &str) {
 }
 
 /// Returns a redacted version of the DSN string with the password replaced by `***`.
+///
+/// The credentials/host boundary is the *last* `@` before the `tcp(...)` host
+/// component. Splitting on the first `@` would leak the tail of a password that
+/// itself contains `@` (e.g. `user:p@ss@tcp(host:3306)/db`). We therefore split
+/// on the last `@` and replace the entire password segment.
 pub(crate) fn redact_mysql_dsn(dsn_string: &str) -> String {
     let input = dsn_string.trim();
     if input.is_empty() {
@@ -304,10 +399,12 @@ pub(crate) fn redact_mysql_dsn(dsn_string: &str) -> String {
 
     let (prefix, remainder) = split_mysql_scheme(input);
 
-    match remainder.split_once('@') {
+    match remainder.rsplit_once('@') {
         Some((credentials, host_part)) => match credentials.split_once(':') {
+            // `user` is everything before the first `:`; the password (which may
+            // contain `@` or `:`) is fully replaced, so nothing after it leaks.
             Some((user, _)) => format!("{}{}:***@{}", prefix, user.trim(), host_part.trim()),
-            None => format!("{prefix}***@{host_part}"),
+            None => format!("{prefix}***@{}", host_part.trim()),
         },
         None => format!("{prefix}***"),
     }
@@ -410,7 +507,12 @@ pub(crate) fn extract_event_time(body: &[u8]) -> Result<String, TargetError> {
     Ok(dt.format("%Y-%m-%d %H:%M:%S%.6f").to_string())
 }
 
-async fn validate_existing_schema(conn: &mut Conn, table: &str) -> Result<(), TargetError> {
+/// Validates the required `event_time`/`event_data` columns and reports whether
+/// the optional `event_id` idempotency key column is present.
+///
+/// Returns `Ok(true)` when the `event_id` column exists (idempotent inserts
+/// available), `Ok(false)` for a valid legacy two-column table.
+async fn validate_existing_schema(conn: &mut Conn, table: &str) -> Result<bool, TargetError> {
     let quoted = quote_table_name(table)?;
     let sql = format!("SHOW COLUMNS FROM {quoted}");
 
@@ -421,13 +523,16 @@ async fn validate_existing_schema(conn: &mut Conn, table: &str) -> Result<(), Ta
 
     let mut has_event_time = false;
     let mut has_event_data = false;
+    let mut has_event_id = false;
 
     for row in &columns {
         let field: String = row.get(0).unwrap_or_default();
         let col_type: String = row.get(1).unwrap_or_default();
         let nullable: String = row.get(2).unwrap_or_default();
 
-        if field == "event_time" {
+        if field == MYSQL_EVENT_ID_COLUMN {
+            has_event_id = true;
+        } else if field == "event_time" {
             has_event_time = true;
             if col_type.to_lowercase() != "datetime(6)" {
                 return Err(TargetError::Initialization(
@@ -465,7 +570,7 @@ async fn validate_existing_schema(conn: &mut Conn, table: &str) -> Result<(), Ta
         ));
     }
 
-    Ok(())
+    Ok(has_event_id)
 }
 
 /// A notification target that writes events to a MySQL/TiDB table.
@@ -516,6 +621,14 @@ where
     pool: Arc<Mutex<Option<Pool>>>,
     /// TLS fingerprint tracking for hot reload (inline fallback path)
     tls_state: Arc<parking_lot::Mutex<super::TargetTlsState>>,
+    /// Cached mtimes of the TLS material files. The inline fingerprint is only
+    /// recomputed when these change, avoiding a per-send read of all three cert
+    /// files.
+    tls_mtime_cache: Arc<parking_lot::Mutex<Option<TlsFileMtimes>>>,
+    /// Whether the target table carries the `event_id` idempotency key. Set when
+    /// the pool is built; when `false` the legacy (non-idempotent) insert is used
+    /// for backward compatibility with pre-existing two-column tables.
+    idempotency_supported: Arc<AtomicBool>,
     /// When present, the adapter provides coordinator-managed TLS material;
     /// otherwise the inline fingerprint path is used as a fallback.
     tls_adapter: Option<TlsReloadAdapter<Pool>>,
@@ -556,6 +669,8 @@ where
             // Pool is lazily initialized on first use to avoid unnecessary connections at startup and allow for better error handling
             pool: Arc::new(Mutex::new(None)),
             tls_state: Arc::new(parking_lot::Mutex::new(super::TargetTlsState::default())),
+            tls_mtime_cache: Arc::new(parking_lot::Mutex::new(None)),
+            idempotency_supported: Arc::new(AtomicBool::new(false)),
             tls_adapter: None,
             delivery_counters: Arc::new(TargetDeliveryCounters::default()),
             _phantom: PhantomData,
@@ -591,16 +706,40 @@ where
         }
 
         // Inline fingerprint fallback path (no coordinator).
-        let next_fingerprint =
-            super::build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
-        let tls_changed = {
-            let tls_state_guard = self.tls_state.lock();
-            tls_state_guard.needs_update(&next_fingerprint)
+        //
+        // Recomputing the TLS content fingerprint reads and hashes up to three
+        // certificate files. To avoid doing that on every checkout, we first
+        // compare the cheap file mtimes and only recompute the fingerprint when
+        // a file's mtime changed (or on the first call).
+        let current_mtimes = TlsFileMtimes::read(&self.args);
+        let mtimes_unchanged = {
+            let cache = self.tls_mtime_cache.lock();
+            cache.as_ref() == Some(&current_mtimes)
         };
-        if tls_changed {
-            let mut guard = self.pool.lock().await;
-            *guard = None;
-            self.tls_state.lock().refresh(next_fingerprint);
+
+        if !mtimes_unchanged {
+            let next_fingerprint =
+                super::build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
+                    .await?;
+            let tls_changed = {
+                let tls_state_guard = self.tls_state.lock();
+                tls_state_guard.needs_update(&next_fingerprint)
+            };
+            if tls_changed {
+                // Disconnect the old pool before dropping it so its connections
+                // are closed gracefully instead of leaked on TLS rotation.
+                let old_pool = {
+                    let mut guard = self.pool.lock().await;
+                    guard.take()
+                };
+                if let Some(old_pool) = old_pool
+                    && let Err(err) = old_pool.disconnect().await
+                {
+                    warn!(target_id = %self.id, error = %err, "Failed to disconnect stale MySQL pool during TLS reload");
+                }
+                self.tls_state.lock().refresh(next_fingerprint);
+            }
+            *self.tls_mtime_cache.lock() = Some(current_mtimes);
         }
 
         {
@@ -610,7 +749,7 @@ where
             }
         }
 
-        let pool = build_mysql_pool_from_args(&self.args).await?;
+        let (pool, idempotency) = build_mysql_pool_from_args(&self.args).await?;
 
         // Double-check: another caller may have initialized the pool
         // while we were doing I/O.
@@ -622,12 +761,20 @@ where
             );
             return Ok(existing.clone());
         }
+        self.idempotency_supported.store(idempotency, Ordering::Relaxed);
         *guard = Some(pool.clone());
         Ok(pool)
     }
 
-    /// Inserts an event directly into the MySQL table.
-    async fn insert_event(&self, body: &[u8], meta: &QueuedPayloadMeta) -> Result<(), TargetError> {
+    /// Inserts an event into the MySQL table.
+    ///
+    /// `event_id` is a stable per-event identifier used as the idempotency key:
+    /// the store key for replays, or a fresh UUID for immediate delivery. When
+    /// the table carries the `event_id` primary key, the insert is idempotent
+    /// (`ON DUPLICATE KEY UPDATE` no-op) so a replay after a lost ack does not
+    /// append a duplicate audit row. On legacy two-column tables the key is
+    /// ignored and the legacy insert is used.
+    async fn insert_event(&self, body: &[u8], meta: &QueuedPayloadMeta, event_id: &str) -> Result<(), TargetError> {
         debug!(
             target_id = %self.id,
             bucket = %meta.bucket_name,
@@ -641,20 +788,25 @@ where
         // At this point the pool has already been initialized (get_or_init_pool
         // succeeded above), so get_conn() failures are always transient: the
         // connection was lost or the pool is temporarily exhausted.
-        let mut conn = pool.get_conn().await.map_err(|_| TargetError::NotConnected)?;
+        let mut conn = checkout_conn(&pool).await?;
 
         let event_time = extract_event_time(body)?;
         let event_data =
             std::str::from_utf8(body).map_err(|e| TargetError::Serialization(format!("Event body is not valid UTF-8: {e}")))?;
 
-        let sql = format!(
-            "INSERT INTO {} (event_time, event_data) VALUES (?, CAST(? AS JSON))",
-            quote_table_name(&self.args.table)?
-        );
+        let quoted_table = quote_table_name(&self.args.table)?;
 
-        conn.exec_drop(sql, (event_time.as_str(), event_data))
-            .await
-            .map_err(|err| map_mysql_error(err, "Failed to insert event"))?;
+        if self.idempotency_supported.load(Ordering::Relaxed) {
+            let sql = mysql_insert_sql_with_event_id(&quoted_table);
+            conn.exec_drop(sql, (event_id, event_time.as_str(), event_data))
+                .await
+                .map_err(|err| map_mysql_error(err, "Failed to insert event"))?;
+        } else {
+            let sql = mysql_insert_sql_legacy(&quoted_table);
+            conn.exec_drop(sql, (event_time.as_str(), event_data))
+                .await
+                .map_err(|err| map_mysql_error(err, "Failed to insert event"))?;
+        }
 
         self.delivery_counters.record_success();
         debug!(target_id = %self.id, "MySQL event inserted");
@@ -668,6 +820,8 @@ where
             store: self.store.as_ref().map(|s| s.boxed_clone()),
             pool: Arc::clone(&self.pool),
             tls_state: Arc::clone(&self.tls_state),
+            tls_mtime_cache: Arc::clone(&self.tls_mtime_cache),
+            idempotency_supported: Arc::clone(&self.idempotency_supported),
             tls_adapter: self.tls_adapter.clone(),
             delivery_counters: Arc::clone(&self.delivery_counters),
             _phantom: PhantomData,
@@ -681,7 +835,11 @@ where
 /// This is a standalone function so it can be called both from
 /// `get_or_init_pool` (inline fallback) and from `build_tls_material`
 /// (coordinator path).
-async fn build_mysql_pool_from_args(args: &MySqlArgs) -> Result<Pool, TargetError> {
+///
+/// Returns the pool together with a boolean indicating whether the target table
+/// carries the `event_id` idempotency key (`true`) or is a legacy two-column
+/// table (`false`).
+async fn build_mysql_pool_from_args(args: &MySqlArgs) -> Result<(Pool, bool), TargetError> {
     let dsn = MySqlDsn::parse(&args.dsn_string)?;
 
     let mut builder = OptsBuilder::default()
@@ -731,21 +889,29 @@ async fn build_mysql_pool_from_args(args: &MySqlArgs) -> Result<Pool, TargetErro
     // short reads/writes to the pool cache. All I/O (connecting,
     // DDL, schema validation) happens outside the lock so that
     // concurrent callers are not blocked by a slow MySQL server.
-    let mut conn = pool.get_conn().await.map_err(|_| TargetError::NotConnected)?;
+    let mut conn = checkout_conn(&pool).await?;
 
     conn.query_drop("SELECT 1").await.map_err(|_| TargetError::NotConnected)?;
 
-    let ddl = format!(
-        "CREATE TABLE IF NOT EXISTS {} (event_time DATETIME(6) NOT NULL, event_data JSON NOT NULL)",
-        quote_table_name(&args.table)?
-    );
-    conn.query_drop(ddl)
+    let quoted_table = quote_table_name(&args.table)?;
+    // Tables created here carry the `event_id` primary key so that store
+    // replays are idempotent. Pre-existing legacy tables are left untouched by
+    // `CREATE TABLE IF NOT EXISTS`.
+    conn.query_drop(mysql_create_table_sql(&quoted_table))
         .await
         .map_err(|e| TargetError::Initialization(format!("Failed to create MySQL table: {e}")))?;
 
-    validate_existing_schema(&mut conn, &args.table).await?;
+    let idempotency_supported = validate_existing_schema(&mut conn, &args.table).await?;
+    if !idempotency_supported {
+        warn!(
+            table = %args.table,
+            "MySQL table lacks the '{}' idempotency key column; store replays may create duplicate rows. \
+             Add an '{}' VARCHAR(255) PRIMARY KEY column to enable exactly-once inserts.",
+            MYSQL_EVENT_ID_COLUMN, MYSQL_EVENT_ID_COLUMN
+        );
+    }
 
-    Ok(pool)
+    Ok((pool, idempotency_supported))
 }
 
 /// Maps a mysql_async error to `TargetError`:
@@ -819,7 +985,10 @@ where
             debug!("Event saved to queue store for MySQL target: {}", self.id);
             Ok(())
         } else {
-            if let Err(err) = self.insert_event(&queued.body, &queued.meta).await {
+            // No queue: deliver immediately. A fresh UUID is the idempotency
+            // key so caller-side retries produce distinct rows.
+            let event_id = Uuid::new_v4().to_string();
+            if let Err(err) = self.insert_event(&queued.body, &queued.meta, &event_id).await {
                 self.delivery_counters.record_final_failure();
                 return Err(err);
             }
@@ -857,7 +1026,10 @@ where
             }
         }
 
-        if let Err(e) = self.insert_event(&body, &meta).await {
+        // Use the stable store key as the idempotency key so replays of the
+        // same physical event are deduplicated by the `event_id` primary key.
+        let event_id = key.to_string();
+        if let Err(e) = self.insert_event(&body, &meta, &event_id).await {
             if is_connectivity_error(&e) {
                 warn!(target_id = %self.id, "MySQL not reachable, event remains in queue store");
                 return Err(e);
@@ -940,7 +1112,9 @@ where
     }
 
     async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
-        build_mysql_pool_from_args(&self.args).await
+        let (pool, idempotency) = build_mysql_pool_from_args(&self.args).await?;
+        self.idempotency_supported.store(idempotency, Ordering::Relaxed);
+        Ok(pool)
     }
 
     async fn apply_tls_material(
@@ -1068,6 +1242,51 @@ mod tests {
         assert_eq!(redacted, "root:***@tcp(127.0.0.1:4000)/testdb");
     }
 
+    #[test]
+    fn redact_dsn_password_containing_at_sign_does_not_leak() {
+        // A password containing '@' must not leak its tail into the redacted
+        // output; the credentials/host boundary is the last '@'.
+        let redacted = redact_mysql_dsn("rustfs:p@ss@w0rd@tcp(mysql.example.com:3306)/rustfs_events");
+        assert_eq!(redacted, "rustfs:***@tcp(mysql.example.com:3306)/rustfs_events");
+        assert!(!redacted.contains("ss@w0rd"));
+        assert!(!redacted.contains("w0rd"));
+    }
+
+    #[test]
+    fn redact_dsn_password_with_at_and_prefix() {
+        let redacted = redact_mysql_dsn("mysql://rustfs:a@b:c@tcp(127.0.0.1:3306)/mydb");
+        assert_eq!(redacted, "mysql://rustfs:***@tcp(127.0.0.1:3306)/mydb");
+        assert!(!redacted.contains("a@b"));
+    }
+
+    #[test]
+    fn insert_sql_with_event_id_is_idempotent() {
+        let sql = mysql_insert_sql_with_event_id("`rustfs_events`");
+        assert!(sql.contains("event_id, event_time, event_data"));
+        assert!(sql.contains("CAST(? AS JSON)"));
+        assert!(sql.contains("ON DUPLICATE KEY UPDATE event_id = event_id"));
+        assert!(sql.contains("`rustfs_events`"));
+    }
+
+    #[test]
+    fn insert_sql_legacy_has_no_idempotency_clause() {
+        let sql = mysql_insert_sql_legacy("`rustfs_events`");
+        assert!(sql.contains("(event_time, event_data)"));
+        assert!(sql.contains("CAST(? AS JSON)"));
+        assert!(!sql.contains("event_id"));
+        assert!(!sql.contains("ON DUPLICATE KEY"));
+    }
+
+    #[test]
+    fn create_table_sql_defines_event_id_primary_key() {
+        let sql = mysql_create_table_sql("`my_db`.`events`");
+        assert!(sql.contains("CREATE TABLE IF NOT EXISTS `my_db`.`events`"));
+        assert!(sql.contains("event_id VARCHAR(255) NOT NULL"));
+        assert!(sql.contains("event_time DATETIME(6) NOT NULL"));
+        assert!(sql.contains("event_data JSON NOT NULL"));
+        assert!(sql.contains("PRIMARY KEY (event_id)"));
+    }
+
     #[test]
     fn debug_redacts_mysql_secret_fields() {
         let args = MySqlArgs {

crates/targets/src/target/postgres.rs

@@ -42,12 +42,14 @@ use crate::{
     },
 };
 use async_trait::async_trait;
-use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
+use deadpool_postgres::{Client as PooledClient, Manager, ManagerConfig, Pool, RecyclingMethod, Runtime, Timeouts};
 use rustfs_config::{POSTGRES_DSN_STRING, POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY};
+use rustfs_s3_types::EventName;
 use rustfs_tls_runtime::{load_certs, load_private_key};
 use std::fmt;
 use std::path::Path;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio_postgres::Config;
 use tokio_postgres_rustls::MakeRustlsConnect;
 use tracing::{info, instrument, warn};
@@ -56,6 +58,44 @@ use uuid::Uuid;
 
 const TARGET_LOG_KEY_FIELD: &str = "Key";
 
+/// Bounds the underlying TCP connect + startup handshake for a new backend
+/// connection so an unreachable server cannot block a pool slot forever.
+const POSTGRES_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+/// Maximum time `pool.get()` waits for a free slot before returning a timeout.
+const POSTGRES_POOL_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
+/// Maximum time to create a brand-new pooled connection.
+const POSTGRES_POOL_CREATE_TIMEOUT: Duration = Duration::from_secs(15);
+/// Maximum time to recycle (health-check) an idle pooled connection.
+const POSTGRES_POOL_RECYCLE_TIMEOUT: Duration = Duration::from_secs(10);
+/// Absolute ceiling on a single checkout, wrapping `pool.get()` in a Tokio
+/// timeout as a belt-and-suspenders guard on top of the deadpool timeouts.
+const POSTGRES_POOL_CHECKOUT_HARD_LIMIT: Duration = Duration::from_secs(20);
+
+/// Returns `true` for any `s3:ObjectRemoved:*` event.
+///
+/// Used by the `namespace` format so that object deletions remove the row
+/// instead of leaving stale state behind via an UPSERT.
+fn is_object_removed_event(event: &EventName) -> bool {
+    event.as_str().starts_with("s3:ObjectRemoved")
+}
+
+/// Checks out a client from the pool, wrapping `pool.get()` in a Tokio timeout.
+///
+/// The deadpool wait/create timeouts already bound the checkout, but the outer
+/// timeout guarantees a hard ceiling even if a lower layer misbehaves. Any
+/// timeout maps to `TargetError::Timeout`, which is a connectivity error so the
+/// queue store retains the payload for replay.
+async fn checkout_client(pool: &Pool, context: &str) -> Result<PooledClient, TargetError> {
+    match tokio::time::timeout(POSTGRES_POOL_CHECKOUT_HARD_LIMIT, pool.get()).await {
+        Ok(Ok(client)) => Ok(client),
+        Ok(Err(e)) => Err(map_pool_error(e, context)),
+        Err(_) => Err(TargetError::Timeout(format!(
+            "{context}: pool checkout exceeded {}s hard limit",
+            POSTGRES_POOL_CHECKOUT_HARD_LIMIT.as_secs()
+        ))),
+    }
+}
+
 /// Output format selection for the PostgreSQL target.
 ///
 /// - `Namespace`: single-row UPSERT per object key (MinIO `namespace` style).
@@ -388,6 +428,13 @@ pub fn namespace_upsert_sql(schema: &str, table: &str) -> String {
     )
 }
 
+/// SQL for the `namespace` format on object removal. Deletes the row keyed on
+/// the object key so the `namespace` table stays consistent with the object
+/// lifecycle instead of retaining stale state after a delete.
+pub fn namespace_delete_sql(schema: &str, table: &str) -> String {
+    format!("DELETE FROM {} WHERE key = $1", qualified_table(schema, table))
+}
+
 /// SQL for the `access` format. Append-only with `event_id` as PK so that
 /// store-replay scenarios silently skip duplicates while distinct events still
 /// land as separate rows.
@@ -472,6 +519,9 @@ pub fn build_pool(args: &PostgresArgs) -> Result<Pool, TargetError> {
         .port(parsed.port)
         .user(&parsed.user)
         .dbname(&parsed.database)
+        // Bound the TCP connect + startup handshake so an unreachable backend
+        // cannot block a pool slot indefinitely.
+        .connect_timeout(POSTGRES_CONNECT_TIMEOUT)
         .options(format!("-c search_path={}", parsed.schema));
     if let Some(password) = parsed.password.as_deref()
         && !password.is_empty()
@@ -491,30 +541,59 @@ pub fn build_pool(args: &PostgresArgs) -> Result<Pool, TargetError> {
         Manager::from_config(pg_config, tokio_postgres::NoTls, manager_config)
     };
 
+    // Explicit wait/create/recycle timeouts guarantee that `pool.get()` always
+    // returns within a bounded time when the broker/DB is unreachable, instead
+    // of blocking the delivery thread forever. A Tokio runtime is required for
+    // deadpool to honor these timeouts.
     Pool::builder(manager)
+        .runtime(Runtime::Tokio1)
+        .timeouts(Timeouts {
+            wait: Some(POSTGRES_POOL_WAIT_TIMEOUT),
+            create: Some(POSTGRES_POOL_CREATE_TIMEOUT),
+            recycle: Some(POSTGRES_POOL_RECYCLE_TIMEOUT),
+        })
         .build()
         .map_err(|e| TargetError::Configuration(format!("failed to build PostgreSQL pool: {e}")))
 }
 
+/// Classifies a PostgreSQL SQLSTATE code into the proper `TargetError` variant.
+///
+/// Split out from [`map_pg_error`] so the SQLSTATE-to-variant mapping can be
+/// unit-tested without constructing an opaque `tokio_postgres::Error`.
+///
+/// Classification is by SQLSTATE class (first two characters):
+/// - `08` connection exception → `NotConnected` (retry, keep in store).
+/// - `28` invalid authorization → `Authentication` (permanent, surfaced).
+/// - `23`/`42` integrity/syntax → `Configuration` (permanent, surfaced).
+/// - `40` transaction rollback (`40001` serialization_failure,
+///   `40P01` deadlock_detected, …) → `Timeout`, a transient/retryable error:
+///   the transaction should be retried rather than dropped.
+/// - anything else → `Request` (treated as permanent/ambiguous).
+fn map_pg_sqlstate(code: &str, detail: &str) -> TargetError {
+    match code.get(..2).unwrap_or("") {
+        "08" => TargetError::NotConnected,
+        "28" => TargetError::Authentication(detail.to_string()),
+        "23" | "42" => TargetError::Configuration(detail.to_string()),
+        "40" => TargetError::Timeout(detail.to_string()),
+        _ => TargetError::Request(detail.to_string()),
+    }
+}
+
 /// Maps a `tokio_postgres::Error` to the proper `TargetError` variant.
 ///
 /// Connection-class errors (SQLSTATE 08, closed connection, IO) become
 /// `NotConnected` so the queue store retains the payload for replay.
 /// Schema and constraint problems (SQLSTATE 23, 42) become `Configuration`
 /// so they are surfaced to the operator without endless retry.
+/// Transaction-rollback errors (SQLSTATE class 40, e.g. serialization failure
+/// or deadlock) become `Timeout` so they are retried transiently.
 pub fn map_pg_error(err: &tokio_postgres::Error, context: &str) -> TargetError {
     if err.is_closed() {
         return TargetError::NotConnected;
     }
     if let Some(db_err) = err.as_db_error() {
-        let class = db_err.code().code().get(..2).unwrap_or("");
-        return match class {
-            "08" => TargetError::NotConnected,
-            "28" => TargetError::Authentication(format!("{context}: {db_err}")),
-            "23" | "42" => TargetError::Configuration(format!("{context}: {db_err}")),
-            "40" => TargetError::Request(format!("{context}: {db_err}")),
-            _ => TargetError::Request(format!("{context}: {db_err}")),
-        };
+        let detail = format!("{context}: {db_err}");
+        return map_pg_sqlstate(db_err.code().code(), &detail);
     }
     TargetError::NotConnected
 }
@@ -564,6 +643,7 @@ where
     /// otherwise the inline fingerprint path is used as a fallback.
     tls_adapter: Option<TlsReloadAdapter<Pool>>,
     namespace_sql: String,
+    namespace_delete_sql: String,
     access_sql: String,
     store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
     delivery_counters: Arc<TargetDeliveryCounters>,
@@ -582,6 +662,7 @@ where
             tls_state: Arc::clone(&self.tls_state),
             tls_adapter: self.tls_adapter.clone(),
             namespace_sql: self.namespace_sql.clone(),
+            namespace_delete_sql: self.namespace_delete_sql.clone(),
             access_sql: self.access_sql.clone(),
             store: self.store.as_ref().map(|s| s.boxed_clone()),
             delivery_counters: Arc::clone(&self.delivery_counters),
@@ -607,6 +688,7 @@ where
         Ok(Self {
             id: target_id,
             namespace_sql: namespace_upsert_sql(&args.schema, &args.table),
+            namespace_delete_sql: namespace_delete_sql(&args.schema, &args.table),
             access_sql: access_insert_sql(&args.schema, &args.table),
             args,
             pool: Arc::new(parking_lot::Mutex::new(pool)),
@@ -641,10 +723,7 @@ where
         }
 
         let pool = self.pool.lock().clone();
-        let client = pool
-            .get()
-            .await
-            .map_err(|e| map_pool_error(e, "PostgreSQL pool checkout failed"))?;
+        let client = checkout_client(&pool, "PostgreSQL pool checkout failed").await?;
 
         let payload: serde_json::Value =
             serde_json::from_slice(body).map_err(|e| TargetError::Serialization(format!("Failed to parse JSON payload: {e}")))?;
@@ -652,6 +731,12 @@ where
         let key = resolve_payload_key(&payload, meta);
 
         let result = match self.args.format {
+            // For the single-row `namespace` format, an object removal must
+            // delete the row rather than UPSERT it, otherwise stale state
+            // lingers in the table after the object is gone.
+            PostgresFormat::Namespace if is_object_removed_event(&meta.event_name) => {
+                client.execute(&self.namespace_delete_sql, &[&key]).await
+            }
             PostgresFormat::Namespace => client.execute(&self.namespace_sql, &[&key, &payload]).await,
             PostgresFormat::Access => {
                 let event_name_str = meta.event_name.to_string();
@@ -675,10 +760,7 @@ where
     /// configured: events buffer in the store until the schema is fixed.
     async fn probe_table(&self) -> Result<(), TargetError> {
         let pool = self.pool.lock().clone();
-        let client = pool
-            .get()
-            .await
-            .map_err(|e| map_pool_error(e, "PostgreSQL pool checkout failed during init probe"))?;
+        let client = checkout_client(&pool, "PostgreSQL pool checkout failed during init probe").await?;
         let sql = table_probe_sql(&self.args.schema, &self.args.table);
         client
             .execute(sql.as_str(), &[])
@@ -737,12 +819,9 @@ where
             return Ok(false);
         }
 
-        match tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        match tokio::time::timeout(Duration::from_secs(10), async {
             let pool = self.pool.lock().clone();
-            let client = pool
-                .get()
-                .await
-                .map_err(|e| map_pool_error(e, "PostgreSQL pool checkout failed"))?;
+            let client = checkout_client(&pool, "PostgreSQL pool checkout failed").await?;
             client
                 .execute("SELECT 1", &[])
                 .await
@@ -1091,6 +1170,53 @@ mod tests {
         assert!(sql.contains("$4::jsonb"));
     }
 
+    #[test]
+    fn namespace_delete_targets_row_by_key() {
+        let sql = namespace_delete_sql("public", "events");
+        assert!(sql.starts_with("DELETE FROM"));
+        assert!(sql.contains(r#""public"."events""#));
+        assert!(sql.contains("WHERE key = $1"));
+    }
+
+    #[test]
+    fn is_object_removed_event_matches_all_removed_variants() {
+        assert!(is_object_removed_event(&EventName::ObjectRemovedDelete));
+        assert!(is_object_removed_event(&EventName::ObjectRemovedDeleteMarkerCreated));
+        assert!(is_object_removed_event(&EventName::ObjectRemovedDeleteAllVersions));
+        assert!(is_object_removed_event(&EventName::ObjectRemovedAll));
+        assert!(!is_object_removed_event(&EventName::ObjectCreatedPut));
+        assert!(!is_object_removed_event(&EventName::ObjectAccessedGet));
+    }
+
+    #[test]
+    fn map_pg_sqlstate_classifies_transaction_rollback_as_transient() {
+        // 40001 serialization_failure and 40P01 deadlock_detected are transient
+        // and must be retried, not dropped as permanent failures.
+        assert!(matches!(map_pg_sqlstate("40001", "ctx: serialization"), TargetError::Timeout(_)));
+        assert!(matches!(map_pg_sqlstate("40P01", "ctx: deadlock"), TargetError::Timeout(_)));
+        assert!(matches!(map_pg_sqlstate("40000", "ctx: rollback"), TargetError::Timeout(_)));
+    }
+
+    #[test]
+    fn map_pg_sqlstate_classifies_connection_and_permanent_errors() {
+        assert!(matches!(map_pg_sqlstate("08006", "ctx"), TargetError::NotConnected));
+        assert!(matches!(map_pg_sqlstate("08001", "ctx"), TargetError::NotConnected));
+        assert!(matches!(map_pg_sqlstate("28P01", "ctx: auth"), TargetError::Authentication(_)));
+        assert!(matches!(map_pg_sqlstate("23505", "ctx: unique"), TargetError::Configuration(_)));
+        assert!(matches!(map_pg_sqlstate("42P01", "ctx: undefined_table"), TargetError::Configuration(_)));
+        // Unknown class stays permanent (ambiguous → surfaced as Request).
+        assert!(matches!(map_pg_sqlstate("22001", "ctx: data"), TargetError::Request(_)));
+        assert!(matches!(map_pg_sqlstate("", "ctx: empty"), TargetError::Request(_)));
+    }
+
+    #[test]
+    fn transient_pg_errors_are_connectivity_errors() {
+        // Transaction-rollback errors must be treated as connectivity errors so
+        // the queue store retains the payload for replay instead of dropping it.
+        assert!(crate::target::is_connectivity_error(&map_pg_sqlstate("40001", "ctx")));
+        assert!(crate::target::is_connectivity_error(&map_pg_sqlstate("40P01", "ctx")));
+    }
+
     #[test]
     fn table_probe_does_not_select_rows() {
         let sql = table_probe_sql("public", "events");