diff --git a/crates/targets/src/target/mysql.rs b/crates/targets/src/target/mysql.rs index efa984611..3a1508088 100644 --- a/crates/targets/src/target/mysql.rs +++ b/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, + client_cert: Option, + client_key: Option, +} + +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 { + 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 { + 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 { 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 { 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>>, /// TLS fingerprint tracking for hot reload (inline fallback path) tls_state: Arc>, + /// 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>>, + /// 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, /// When present, the adapter provides coordinator-managed TLS material; /// otherwise the inline fingerprint path is used as a fallback. tls_adapter: Option>, @@ -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(¤t_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("ed_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("ed_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 { +/// +/// 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 Result { - 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 { diff --git a/crates/targets/src/target/postgres.rs b/crates/targets/src/target/postgres.rs index b9c7543e1..fd4dcf6f0 100644 --- a/crates/targets/src/target/postgres.rs +++ b/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 { + 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 { .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 { 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>, namespace_sql: String, + namespace_delete_sql: String, access_sql: String, store: Option + Send + Sync>>, delivery_counters: Arc, @@ -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");