diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index 7781e49..9317ed1 100644 --- a/.github/workflows/fuzz.yml +++ b/.github/workflows/fuzz.yml @@ -22,7 +22,10 @@ jobs: - uses: actions/checkout@v4 - name: Install nightly toolchain - uses: dtolnay/rust-toolchain@nightly + # Pinned around the 2026-07-24 codegen ICE (rustc da86f4d07) that panics in rustc_codegen_ssa (operand.rs:291 "not immediate") while compiling tokio under -Zsanitizer=address. nightly-2026-07-23 ships the last-good compiler (rustc 6f72b5dd5, 2026-07-22). Un-pin once fixed upstream. + uses: dtolnay/rust-toolchain@master + with: + toolchain: nightly-2026-07-23 - name: Install system dependencies run: | diff --git a/src/connection/libpq.rs b/src/connection/libpq.rs index 0d31518..8283556 100644 --- a/src/connection/libpq.rs +++ b/src/connection/libpq.rs @@ -690,8 +690,8 @@ impl PgReplicationConnection { output_plugin: Option<&str>, options: &ReplicationSlotOptions, ) -> Result { - crate::sql_builder::check_create_slot_version(self.server_version(), slot_type, options)?; - let sql = crate::sql_builder::build_create_slot_sql( + let sql = crate::sql_builder::prepare_create_slot( + self.server_version(), slot_name, slot_type, output_plugin, @@ -708,9 +708,12 @@ impl PgReplicationConnection { two_phase: Option, failover: Option, ) -> Result { - crate::sql_builder::check_alter_slot_version(self.server_version(), two_phase)?; - let alter_slot_sql = - crate::sql_builder::build_alter_slot_sql(slot_name, two_phase, failover)?; + let alter_slot_sql = crate::sql_builder::prepare_alter_slot( + self.server_version(), + slot_name, + two_phase, + failover, + )?; debug!("Altering replication slot: {}", alter_slot_sql); let result = self.exec(&alter_slot_sql)?; @@ -754,8 +757,7 @@ impl PgReplicationConnection { &mut self, slot_name: &str, ) -> Result { - crate::sql_builder::check_read_slot_version(self.server_version())?; - let sql = crate::sql_builder::build_read_slot_sql(slot_name)?; + let sql = crate::sql_builder::prepare_read_slot(self.server_version(), slot_name)?; debug!("Reading replication slot: {}", sql); let result = self.exec(&sql)?; @@ -909,6 +911,21 @@ impl PgReplicationConnection { fn push_pending_message_for_testing(&mut self, msg: Bytes) { self.pending_messages.push_back(msg); } + + /// Test-only: a null connection pre-seeded with COPY-data frames that + /// `get_copy_data_async` serves in order from `pending_messages` before any + /// FFI. `is_replication_conn` is set so the replication-mode gate passes. + /// + /// Mirrors the native backend's `null_for_testing_with_frames`, giving the + /// shared stream pump one frame-injection seam across both adapters. The + /// `conn` pointer is null, so no FFI path (feedback send, CopyDone) is safe + /// to drive here — those stay backend-specific and integration-tested. + pub(crate) fn null_for_testing_with_frames(frames: Vec) -> Self { + let mut conn = Self::null_for_testing(); + conn.is_replication_conn = true; + conn.pending_messages.extend(frames); + conn + } } /// Safe wrapper for a PostgreSQL result. diff --git a/src/connection/native/connection.rs b/src/connection/native/connection.rs index b473672..c567fb9 100644 --- a/src/connection/native/connection.rs +++ b/src/connection/native/connection.rs @@ -815,8 +815,8 @@ impl NativeConnection { output_plugin: Option<&str>, options: &ReplicationSlotOptions, ) -> Result { - crate::sql_builder::check_create_slot_version(self.server_version(), slot_type, options)?; - let sql = crate::sql_builder::build_create_slot_sql( + let sql = crate::sql_builder::prepare_create_slot( + self.server_version(), slot_name, slot_type, output_plugin, @@ -833,8 +833,12 @@ impl NativeConnection { two_phase: Option, failover: Option, ) -> Result { - crate::sql_builder::check_alter_slot_version(self.server_version(), two_phase)?; - let sql = crate::sql_builder::build_alter_slot_sql(slot_name, two_phase, failover)?; + let sql = crate::sql_builder::prepare_alter_slot( + self.server_version(), + slot_name, + two_phase, + failover, + )?; debug!("Altering replication slot: {}", sql); let result = self.exec(&sql)?; @@ -865,8 +869,7 @@ impl NativeConnection { &mut self, slot_name: &str, ) -> Result { - crate::sql_builder::check_read_slot_version(self.server_version())?; - let sql = crate::sql_builder::build_read_slot_sql(slot_name)?; + let sql = crate::sql_builder::prepare_read_slot(self.server_version(), slot_name)?; debug!("Reading replication slot: {}", sql); let result = self.exec(&sql)?; if !result.is_ok() { diff --git a/src/deserializer.rs b/src/deserializer.rs index 71f5a35..efbcf68 100644 --- a/src/deserializer.rs +++ b/src/deserializer.rs @@ -281,7 +281,7 @@ impl<'de> MapAccess<'de> for RowDataMapAccess<'de> { .current_value .take() .expect("next_value_seed called before next_key_seed"); - seed.deserialize(ColumnValueDeserializer { value }) + seed.deserialize(ColumnValueDeserializer::strict(value)) } #[inline] @@ -293,94 +293,219 @@ impl<'de> MapAccess<'de> for RowDataMapAccess<'de> { // --------------------------------------------------------------------------- // ColumnValueDeserializer — deserializes a single ColumnValue +// +// One deserializer, two error policies. `Strict` returns `Err` on any coercion failure; `Lenient` records a `FieldError` and substitutes a default so every field is visited. The policy is a compile-time type parameter, so it monomorphizes away and the strict path keeps its `#[inline]` zero-copy shape. +// +// The one axis the two policies cannot share is the terminal string/bytes visit: strict borrows for `'de` (`visit_borrowed_str`), while the lenient value borrow is decoupled from `'de` (`visit_str`) — so those methods, and only those, live in the two separate `Deserializer` impls below. Every value coercion (parse rule + message) is single-sourced here. // --------------------------------------------------------------------------- -struct ColumnValueDeserializer<'a> { +/// Structural-shape rejections — identical for both policies (both error). +const MSG_SEQ: &str = "sequences are not supported in RowData deserialization"; +const MSG_TUPLE: &str = "tuples are not supported in RowData deserialization"; +const MSG_TUPLE_STRUCT: &str = "tuple structs are not supported in RowData deserialization"; +const MSG_MAP: &str = "nested maps are not supported in RowData deserialization"; +const MSG_STRUCT: &str = "nested structs are not supported in RowData deserialization"; + +#[cold] +#[inline(never)] +fn null_msg(target: &str) -> String { + format!("cannot deserialize NULL as {target} (use Option<{target}>)") +} + +#[cold] +#[inline(never)] +fn binary_msg(target: &str) -> String { + format!("cannot deserialize binary column as {target}") +} + +/// Build a numeric-parse message. Performs UTF-8 validation only on this cold path so that invalid-UTF-8 inputs still surface the expected diagnostic. +#[cold] +#[inline(never)] +fn numeric_parse_msg(b: &[u8], type_name: &str) -> String { + match core::str::from_utf8(b) { + Ok(s) => format!("failed to parse '{s}' as {type_name}"), + Err(e) => format!("invalid UTF-8 for {type_name}: {e}"), + } +} + +/// Compile-time error policy: how a failed column coercion is surfaced. +/// +/// `Strict` returns an `Err`; `Lenient` records a [`FieldError`] against the current column and returns a type-appropriate fallback (so a lenient run visits every field). This is the ONLY axis on which the two column deserializers differ in their value coercions. +trait ErrorPolicy { + /// Surface a coercion failure. `make` builds the message lazily (cold path); + /// `fallback` is the lenient substitute, ignored by `Strict`. + fn fail( + &self, + make: impl FnOnce() -> String, + fallback: impl FnOnce() -> T, + ) -> Result; +} + +/// Strict policy: any coercion failure is a hard error. +struct Strict; + +impl ErrorPolicy for Strict { + #[inline] + fn fail( + &self, + make: impl FnOnce() -> String, + _fallback: impl FnOnce() -> T, + ) -> Result { + Err(ReplicationError::deserialize(make())) + } +} + +/// Lenient policy: record the failure against `field` and substitute a default. +struct Lenient<'a> { + ctx: &'a LenientCtx, + field: &'a str, +} + +impl ErrorPolicy for Lenient<'_> { + #[inline] + fn fail( + &self, + make: impl FnOnce() -> String, + fallback: impl FnOnce() -> T, + ) -> Result { + self.ctx.push(self.field, make()); + Ok(fallback()) + } +} + +struct ColumnValueDeserializer<'a, P> { value: &'a ColumnValue, + policy: P, } -impl<'de> ColumnValueDeserializer<'de> { - /// Get the text content or return an error for the given target type. +impl<'a> ColumnValueDeserializer<'a, Strict> { #[inline] - fn text_or_err(&self, target: &str) -> Result<&'de str, ReplicationError> { + fn strict(value: &'a ColumnValue) -> Self { + Self { + value, + policy: Strict, + } + } + + /// Borrowed UTF-8 text for the strict terminal string paths. Errors on `Null`/`Binary`/invalid-UTF-8 — strict never substitutes. + #[inline] + fn text_or_err(&self, target: &str) -> Result<&'a str, ReplicationError> { match self.value { ColumnValue::Text(b) => core::str::from_utf8(b).map_err(|e| { ReplicationError::deserialize(format!("invalid UTF-8 for {target}: {e}")) }), - ColumnValue::Null => Err(ReplicationError::deserialize(format!( - "cannot deserialize NULL as {target} (use Option<{target}>)" - ))), - ColumnValue::Binary(_) => Err(ReplicationError::deserialize(format!( - "cannot deserialize binary column as {target}" - ))), + ColumnValue::Null => Err(ReplicationError::deserialize(null_msg(target))), + ColumnValue::Binary(_) => Err(ReplicationError::deserialize(binary_msg(target))), } } +} - /// Like [`text_or_err`] but returns raw bytes — skips the UTF-8 pass. - /// Use only for paths where the parser itself rejects non-ASCII bytes - /// (numeric, bool). String-shaped paths must use `text_or_err`. +impl<'a, P: ErrorPolicy> ColumnValueDeserializer<'a, P> { + /// Raw text bytes (no UTF-8 pass) for numeric/bool coercion. `Text` → the + /// bytes; `Null`/`Binary` → policy failure yielding `None`. #[inline] - fn bytes_or_err(&self, target: &str) -> Result<&'de [u8], ReplicationError> { + fn text_bytes(&self, target: &str) -> Result, ReplicationError> { match self.value { - ColumnValue::Text(b) => Ok(b.as_ref()), - ColumnValue::Null => Err(ReplicationError::deserialize(format!( - "cannot deserialize NULL as {target} (use Option<{target}>)" - ))), - ColumnValue::Binary(_) => Err(ReplicationError::deserialize(format!( - "cannot deserialize binary column as {target}" - ))), + ColumnValue::Text(b) => Ok(Some(b.as_ref())), + ColumnValue::Null => self.policy.fail(|| null_msg(target), || None), + ColumnValue::Binary(_) => self.policy.fail(|| binary_msg(target), || None), } } - /// Parse text as a numeric type via the std `FromStr` (used for floats). + /// UTF-8 text. `Text(valid)` → the str; `Null`/`Binary`/invalid → policy failure. #[inline] - fn parse_text(&self, type_name: &str) -> Result - where - T::Err: core::fmt::Display, - { - let s = self.text_or_err(type_name)?; - s.parse::().map_err(|e| { - ReplicationError::deserialize(format!( - "failed to parse '{}' as {}: {}", - s, type_name, e - )) - }) + fn text(&self, target: &str) -> Result, ReplicationError> { + match self.value { + ColumnValue::Text(b) => match core::str::from_utf8(b) { + Ok(s) => Ok(Some(s)), + Err(e) => self + .policy + .fail(|| format!("invalid UTF-8 for {target}: {e}"), || None), + }, + ColumnValue::Null => self.policy.fail(|| null_msg(target), || None), + ColumnValue::Binary(_) => self.policy.fail(|| binary_msg(target), || None), + } } - /// Fast signed-int parse from bytes; emits an error string equivalent - /// to the std parser (token + type name) so existing tests still match. #[inline] - fn parse_signed(&self, type_name: &str) -> Result - where - T: TryFrom, - { - let b = self.bytes_or_err(type_name)?; - parse_int_signed::(b).ok_or_else(|| numeric_parse_error(b, type_name)) + fn signed + Default>(&self, target: &str) -> Result { + let Some(b) = self.text_bytes(target)? else { + return Ok(T::default()); + }; + match parse_int_signed::(b) { + Some(v) => Ok(v), + None => self + .policy + .fail(|| numeric_parse_msg(b, target), T::default), + } } - /// Fast unsigned-int parse from bytes. #[inline] - fn parse_unsigned(&self, type_name: &str) -> Result + fn unsigned + Default>(&self, target: &str) -> Result { + let Some(b) = self.text_bytes(target)? else { + return Ok(T::default()); + }; + match parse_int_unsigned::(b) { + Some(v) => Ok(v), + None => self + .policy + .fail(|| numeric_parse_msg(b, target), T::default), + } + } + + #[inline] + fn float(&self, target: &str) -> Result where - T: TryFrom, + T: core::str::FromStr + Default, + T::Err: core::fmt::Display, { - let b = self.bytes_or_err(type_name)?; - parse_int_unsigned::(b).ok_or_else(|| numeric_parse_error(b, type_name)) + let Some(s) = self.text(target)? else { + return Ok(T::default()); + }; + match s.parse::() { + Ok(v) => Ok(v), + Err(e) => self.policy.fail( + || format!("failed to parse '{s}' as {target}: {e}"), + T::default, + ), + } } -} -/// Build a numeric-parse error. Performs UTF-8 validation only on the cold -/// path so that invalid-UTF-8 inputs still surface the expected diagnostic. -#[cold] -#[inline(never)] -fn numeric_parse_error(b: &[u8], type_name: &str) -> ReplicationError { - match core::str::from_utf8(b) { - Ok(s) => ReplicationError::deserialize(format!("failed to parse '{s}' as {type_name}")), - Err(e) => ReplicationError::deserialize(format!("invalid UTF-8 for {type_name}: {e}")), + #[inline] + fn boolean(&self) -> Result { + let Some(b) = self.text_bytes("bool")? else { + return Ok(false); + }; + match parse_pg_bool(b) { + Some(v) => Ok(v), + None => self.policy.fail( + || { + format!( + "cannot parse '{}' as bool (expected t/f/true/false/1/0/on/off/yes/no)", + lossy_token(b) + ) + }, + || false, + ), + } + } + + #[inline] + fn character(&self) -> Result { + let Some(s) = self.text("char")? else { + return Ok('\0'); + }; + let mut chars = s.chars(); + match (chars.next(), chars.next()) { + (Some(c), None) => Ok(c), + _ => self + .policy + .fail(|| format!("expected single char, got '{s}'"), || '\0'), + } } } -impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de> { +impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de, Strict> { type Error = ReplicationError; #[inline] @@ -397,88 +522,72 @@ impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de> { #[inline] fn deserialize_bool>(self, visitor: V) -> Result { - let b = self.bytes_or_err("bool")?; - let val = parse_pg_bool(b).ok_or_else(|| { - ReplicationError::deserialize(format!( - "cannot parse '{}' as bool (expected t/f/true/false/1/0/on/off/yes/no)", - lossy_token(b) - )) - })?; - visitor.visit_bool(val) + visitor.visit_bool(self.boolean()?) } #[inline] fn deserialize_i8>(self, visitor: V) -> Result { - visitor.visit_i8(self.parse_signed::("i8")?) + visitor.visit_i8(self.signed::("i8")?) } #[inline] fn deserialize_i16>(self, visitor: V) -> Result { - visitor.visit_i16(self.parse_signed::("i16")?) + visitor.visit_i16(self.signed::("i16")?) } #[inline] fn deserialize_i32>(self, visitor: V) -> Result { - visitor.visit_i32(self.parse_signed::("i32")?) + visitor.visit_i32(self.signed::("i32")?) } #[inline] fn deserialize_i64>(self, visitor: V) -> Result { - visitor.visit_i64(self.parse_signed::("i64")?) + visitor.visit_i64(self.signed::("i64")?) } #[inline] fn deserialize_u8>(self, visitor: V) -> Result { - visitor.visit_u8(self.parse_unsigned::("u8")?) + visitor.visit_u8(self.unsigned::("u8")?) } #[inline] fn deserialize_u16>(self, visitor: V) -> Result { - visitor.visit_u16(self.parse_unsigned::("u16")?) + visitor.visit_u16(self.unsigned::("u16")?) } #[inline] fn deserialize_u32>(self, visitor: V) -> Result { - visitor.visit_u32(self.parse_unsigned::("u32")?) + visitor.visit_u32(self.unsigned::("u32")?) } #[inline] fn deserialize_u64>(self, visitor: V) -> Result { - visitor.visit_u64(self.parse_unsigned::("u64")?) + visitor.visit_u64(self.unsigned::("u64")?) } #[inline] fn deserialize_f32>(self, visitor: V) -> Result { - visitor.visit_f32(self.parse_text("f32")?) + visitor.visit_f32(self.float::("f32")?) } #[inline] fn deserialize_f64>(self, visitor: V) -> Result { - visitor.visit_f64(self.parse_text("f64")?) + visitor.visit_f64(self.float::("f64")?) } #[inline] fn deserialize_char>(self, visitor: V) -> Result { - let s = self.text_or_err("char")?; - let mut chars = s.chars(); - match (chars.next(), chars.next()) { - (Some(c), None) => visitor.visit_char(c), - _ => Err(ReplicationError::deserialize(format!( - "expected single char, got '{s}'" - ))), - } + visitor.visit_char(self.character()?) } #[inline] fn deserialize_str>(self, visitor: V) -> Result { - let s = self.text_or_err("str")?; - visitor.visit_borrowed_str(s) + visitor.visit_borrowed_str(self.text_or_err("str")?) } #[inline] fn deserialize_string>(self, visitor: V) -> Result { - let s = self.text_or_err("String")?; - visitor.visit_borrowed_str(s) + visitor.visit_borrowed_str(self.text_or_err("String")?) } #[inline] @@ -537,9 +646,7 @@ impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de> { #[inline] fn deserialize_seq>(self, _visitor: V) -> Result { - Err(ReplicationError::deserialize( - "sequences are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_SEQ)) } #[inline] @@ -548,9 +655,7 @@ impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de> { _len: usize, _visitor: V, ) -> Result { - Err(ReplicationError::deserialize( - "tuples are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_TUPLE)) } #[inline] @@ -560,16 +665,12 @@ impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de> { _len: usize, _visitor: V, ) -> Result { - Err(ReplicationError::deserialize( - "tuple structs are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_TUPLE_STRUCT)) } #[inline] fn deserialize_map>(self, _visitor: V) -> Result { - Err(ReplicationError::deserialize( - "nested maps are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_MAP)) } #[inline] @@ -579,9 +680,7 @@ impl<'de> de::Deserializer<'de> for ColumnValueDeserializer<'de> { _fields: &'static [&'static str], _visitor: V, ) -> Result { - Err(ReplicationError::deserialize( - "nested structs are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_STRUCT)) } #[inline] @@ -759,10 +858,12 @@ impl<'de, 'a> MapAccess<'de> for LenientMapAccess<'a> { .current .take() .expect("next_value_seed called before next_key_seed"); - seed.deserialize(LenientColumnValueDeserializer { + seed.deserialize(ColumnValueDeserializer { value, - field, - ctx: self.ctx, + policy: Lenient { + ctx: self.ctx, + field, + }, }) } @@ -772,65 +873,12 @@ impl<'de, 'a> MapAccess<'de> for LenientMapAccess<'a> { } } -struct LenientColumnValueDeserializer<'a> { - value: &'a ColumnValue, - field: &'a str, - ctx: &'a LenientCtx, -} - -impl<'a> LenientColumnValueDeserializer<'a> { - /// Get the text content, recording a field error and returning `None` on failure. - fn try_text(&self, target: &str) -> Option<&'a str> { - match self.value { - ColumnValue::Text(b) => match core::str::from_utf8(b) { - Ok(s) => Some(s), - Err(e) => { - self.ctx - .push(self.field, format!("invalid UTF-8 for {target}: {e}")); - None - } - }, - ColumnValue::Null => { - self.ctx.push( - self.field, - format!("cannot deserialize NULL as {target} (use Option<{target}>)"), - ); - None - } - ColumnValue::Binary(_) => { - self.ctx.push( - self.field, - format!("cannot deserialize binary column as {target}"), - ); - None - } - } - } - - /// Parse text into `T`; record an error and return the type's default on failure. - fn parse_or_default(&self, target: &str) -> T - where - T: core::str::FromStr + Default, - T::Err: core::fmt::Display, - { - match self.try_text(target) { - Some(s) => s.parse::().unwrap_or_else(|e| { - self.ctx.push( - self.field, - format!("failed to parse '{s}' as {target}: {e}"), - ); - T::default() - }), - None => T::default(), - } - } -} - -impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { +impl<'de, 'a> de::Deserializer<'de> for ColumnValueDeserializer<'a, Lenient<'a>> { type Error = ReplicationError; fn deserialize_any>(self, visitor: V) -> Result { - // Mirror strict deserialize_any; these paths don't error. + // Owned visits: the value borrow is decoupled from `'de`. These paths + // don't error. match self.value { ColumnValue::Null => visitor.visit_none(), ColumnValue::Text(b) => match core::str::from_utf8(b) { @@ -842,82 +890,58 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { } fn deserialize_bool>(self, visitor: V) -> Result { - let v = match self.try_text("bool") { - Some(s) => parse_pg_bool(s.as_bytes()).unwrap_or_else(|| { - self.ctx - .push(self.field, format!("cannot parse '{s}' as bool")); - false - }), - None => false, - }; - visitor.visit_bool(v) + visitor.visit_bool(self.boolean()?) } fn deserialize_i8>(self, visitor: V) -> Result { - visitor.visit_i8(self.parse_or_default::("i8")) + visitor.visit_i8(self.signed::("i8")?) } fn deserialize_i16>(self, visitor: V) -> Result { - visitor.visit_i16(self.parse_or_default::("i16")) + visitor.visit_i16(self.signed::("i16")?) } fn deserialize_i32>(self, visitor: V) -> Result { - visitor.visit_i32(self.parse_or_default::("i32")) + visitor.visit_i32(self.signed::("i32")?) } fn deserialize_i64>(self, visitor: V) -> Result { - visitor.visit_i64(self.parse_or_default::("i64")) + visitor.visit_i64(self.signed::("i64")?) } fn deserialize_u8>(self, visitor: V) -> Result { - visitor.visit_u8(self.parse_or_default::("u8")) + visitor.visit_u8(self.unsigned::("u8")?) } fn deserialize_u16>(self, visitor: V) -> Result { - visitor.visit_u16(self.parse_or_default::("u16")) + visitor.visit_u16(self.unsigned::("u16")?) } fn deserialize_u32>(self, visitor: V) -> Result { - visitor.visit_u32(self.parse_or_default::("u32")) + visitor.visit_u32(self.unsigned::("u32")?) } fn deserialize_u64>(self, visitor: V) -> Result { - visitor.visit_u64(self.parse_or_default::("u64")) + visitor.visit_u64(self.unsigned::("u64")?) } fn deserialize_f32>(self, visitor: V) -> Result { - visitor.visit_f32(self.parse_or_default::("f32")) + visitor.visit_f32(self.float::("f32")?) } fn deserialize_f64>(self, visitor: V) -> Result { - visitor.visit_f64(self.parse_or_default::("f64")) + visitor.visit_f64(self.float::("f64")?) } fn deserialize_char>(self, visitor: V) -> Result { - let c = match self.try_text("char") { - Some(s) => { - let mut chars = s.chars(); - match (chars.next(), chars.next()) { - (Some(c), None) => c, - _ => { - self.ctx - .push(self.field, format!("expected single char, got '{s}'")); - '\0' - } - } - } - None => '\0', - }; - visitor.visit_char(c) + visitor.visit_char(self.character()?) } fn deserialize_str>(self, visitor: V) -> Result { - let s = self.try_text("str").unwrap_or(""); - visitor.visit_str(s) + visitor.visit_str(self.text("str")?.unwrap_or("")) } fn deserialize_string>(self, visitor: V) -> Result { - let s = self.try_text("String").unwrap_or(""); - visitor.visit_str(s) + visitor.visit_str(self.text("String")?.unwrap_or("")) } fn deserialize_bytes>(self, visitor: V) -> Result { match self.value { ColumnValue::Binary(b) | ColumnValue::Text(b) => visitor.visit_bytes(b), ColumnValue::Null => { - self.ctx - .push(self.field, "cannot deserialize NULL as bytes".to_string()); + self.policy + .fail::<()>(|| "cannot deserialize NULL as bytes".to_string(), || ())?; visitor.visit_bytes(&[]) } } @@ -927,10 +951,8 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { match self.value { ColumnValue::Binary(b) | ColumnValue::Text(b) => visitor.visit_byte_buf(b.to_vec()), ColumnValue::Null => { - self.ctx.push( - self.field, - "cannot deserialize NULL as byte_buf".to_string(), - ); + self.policy + .fail::<()>(|| "cannot deserialize NULL as byte_buf".to_string(), || ())?; visitor.visit_byte_buf(Vec::new()) } } @@ -945,8 +967,8 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { fn deserialize_unit>(self, visitor: V) -> Result { if !matches!(self.value, ColumnValue::Null) { - self.ctx - .push(self.field, "expected NULL for unit type".to_string()); + self.policy + .fail::<()>(|| "expected NULL for unit type".to_string(), || ())?; } visitor.visit_unit() } @@ -968,9 +990,7 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { } fn deserialize_seq>(self, _visitor: V) -> Result { - Err(ReplicationError::deserialize( - "sequences are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_SEQ)) } fn deserialize_tuple>( @@ -978,9 +998,7 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { _len: usize, _visitor: V, ) -> Result { - Err(ReplicationError::deserialize( - "tuples are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_TUPLE)) } fn deserialize_tuple_struct>( @@ -989,15 +1007,11 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { _len: usize, _visitor: V, ) -> Result { - Err(ReplicationError::deserialize( - "tuple structs are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_TUPLE_STRUCT)) } fn deserialize_map>(self, _visitor: V) -> Result { - Err(ReplicationError::deserialize( - "nested maps are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_MAP)) } fn deserialize_struct>( @@ -1006,9 +1020,7 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { _fields: &'static [&'static str], _visitor: V, ) -> Result { - Err(ReplicationError::deserialize( - "nested structs are not supported in RowData deserialization", - )) + Err(ReplicationError::deserialize(MSG_STRUCT)) } fn deserialize_enum>( @@ -1017,7 +1029,7 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { _variants: &'static [&'static str], visitor: V, ) -> Result { - // Enum has no sensible per-field default; propagate strict behavior. + // Enum has no sensible per-field default; both policies error. let s = match self.value { ColumnValue::Text(b) => core::str::from_utf8(b).map_err(|e| { ReplicationError::deserialize(format!("invalid UTF-8 for enum: {e}")) @@ -1025,13 +1037,13 @@ impl<'de, 'a> de::Deserializer<'de> for LenientColumnValueDeserializer<'a> { ColumnValue::Null => { return Err(ReplicationError::deserialize(format!( "cannot deserialize NULL as enum (field '{}')", - self.field + self.policy.field ))) } ColumnValue::Binary(_) => { return Err(ReplicationError::deserialize(format!( "cannot deserialize binary as enum (field '{}')", - self.field + self.policy.field ))) } }; @@ -3881,12 +3893,12 @@ mod tests { // Text (valid UTF-8) → visit_borrowed_str let cv = ColumnValue::text("hello"); - let d = super::ColumnValueDeserializer { value: &cv }; + let d = super::ColumnValueDeserializer::strict(&cv); assert_eq!(d.deserialize_any(Rec).unwrap(), Got::Str("hello".into())); // Text (invalid UTF-8) → visit_borrowed_bytes let cv = ColumnValue::text_bytes(Bytes::from_static(&[0xff, 0xfe])); - let d = super::ColumnValueDeserializer { value: &cv }; + let d = super::ColumnValueDeserializer::strict(&cv); assert_eq!( d.deserialize_any(Rec).unwrap(), Got::Bytes(vec![0xff, 0xfe]) @@ -3894,15 +3906,205 @@ mod tests { // Binary → visit_borrowed_bytes let cv = ColumnValue::binary_bytes(Bytes::from_static(&[1, 2, 3])); - let d = super::ColumnValueDeserializer { value: &cv }; + let d = super::ColumnValueDeserializer::strict(&cv); assert_eq!(d.deserialize_any(Rec).unwrap(), Got::Bytes(vec![1, 2, 3])); // Null → visit_none let cv = ColumnValue::Null; - let d = super::ColumnValueDeserializer { value: &cv }; + let d = super::ColumnValueDeserializer::strict(&cv); assert_eq!(d.deserialize_any(Rec).unwrap(), Got::None); } + #[test] + fn test_lenient_column_value_terminal_visits() { + // Directly exercise the lenient ColumnValueDeserializer's owned terminal + // visits: deserialize_any (str/bytes/none), deserialize_bytes (Text + + // NULL), deserialize_str, and deserialize_identifier. + use serde::de::{Deserializer as _, Visitor}; + use std::fmt; + + #[derive(Debug, PartialEq)] + enum Got { + Str(String), + Bytes(Vec), + None, + } + struct Rec; + impl<'de> Visitor<'de> for Rec { + type Value = Got; + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "anything") + } + fn visit_str(self, v: &str) -> Result { + Ok(Got::Str(v.to_string())) + } + fn visit_bytes(self, v: &[u8]) -> Result { + Ok(Got::Bytes(v.to_vec())) + } + fn visit_none(self) -> Result { + Ok(Got::None) + } + } + + let ctx = super::LenientCtx::new(); + + // deserialize_any: Text(valid) → visit_str + let cv = ColumnValue::text("hello"); + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!(d.deserialize_any(Rec).unwrap(), Got::Str("hello".into())); + + // deserialize_any: Text(invalid UTF-8) → visit_bytes + let cv = ColumnValue::text_bytes(Bytes::from_static(&[0xff, 0xfe])); + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!( + d.deserialize_any(Rec).unwrap(), + Got::Bytes(vec![0xff, 0xfe]) + ); + + // deserialize_any: Binary → visit_bytes + let cv = ColumnValue::binary_bytes(Bytes::from_static(&[1, 2, 3])); + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!(d.deserialize_any(Rec).unwrap(), Got::Bytes(vec![1, 2, 3])); + + // deserialize_any: Null → visit_none + let cv = ColumnValue::Null; + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!(d.deserialize_any(Rec).unwrap(), Got::None); + + // deserialize_bytes: Text → visit_bytes + let cv = ColumnValue::text("abc"); + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!( + d.deserialize_bytes(Rec).unwrap(), + Got::Bytes(b"abc".to_vec()) + ); + + // deserialize_str + deserialize_identifier: Text → visit_str + let cv = ColumnValue::text("id"); + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!(d.deserialize_str(Rec).unwrap(), Got::Str("id".into())); + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "f", + }, + }; + assert_eq!( + d.deserialize_identifier(Rec).unwrap(), + Got::Str("id".into()) + ); + + // deserialize_bytes: NULL → records an error and visits empty bytes + let cv = ColumnValue::Null; + let d = super::ColumnValueDeserializer { + value: &cv, + policy: super::Lenient { + ctx: &ctx, + field: "nb", + }, + }; + assert_eq!(d.deserialize_bytes(Rec).unwrap(), Got::Bytes(vec![])); + assert!(ctx.into_errors().iter().any(|e| e.message.contains("NULL"))); + } + + #[test] + fn test_try_lenient_all_integer_widths() { + // Exercises the lenient signed/unsigned/float coercion for every width. + #[derive(Debug, Deserialize)] + struct S { + a: i8, + b: i16, + c: i32, + d: i64, + e: u8, + f: u16, + g: u32, + h: u64, + i: f32, + j: f64, + } + let row = RowData::from_pairs(vec![ + ("a", ColumnValue::text("-1")), + ("b", ColumnValue::text("-2")), + ("c", ColumnValue::text("-3")), + ("d", ColumnValue::text("-4")), + ("e", ColumnValue::text("5")), + ("f", ColumnValue::text("6")), + ("g", ColumnValue::text("7")), + ("h", ColumnValue::text("8")), + ("i", ColumnValue::text("1.5")), + ("j", ColumnValue::text("2.5")), + ]); + let r = row.try_deserialize_into::().unwrap(); + assert!(r.is_clean()); + assert_eq!((r.value.b, r.value.f, r.value.h), (-2, 6, 8)); + assert_eq!((r.value.i, r.value.j), (1.5, 2.5)); + } + + #[test] + fn test_try_lenient_null_integer_defaults_to_zero() { + // NULL through the lenient integer path records an error and substitutes 0. + #[derive(Debug, Deserialize)] + struct S { + n: u64, + m: i16, + } + let row = RowData::from_pairs(vec![("n", ColumnValue::Null), ("m", ColumnValue::Null)]); + let r = row.try_deserialize_into::().unwrap(); + assert_eq!(r.value.n, 0); + assert_eq!(r.value.m, 0); + assert_eq!(r.errors.len(), 2); + assert!(r.errors.iter().all(|e| e.message.contains("NULL"))); + } + + #[test] + fn test_try_lenient_tuple_errors() { + // A tuple-typed field hits the lenient deserialize_tuple structural reject. + #[derive(Debug, Deserialize)] + struct S { + t: (u32, u32), + } + let row = RowData::from_pairs(vec![("t", ColumnValue::text("1"))]); + assert!(row.try_deserialize_into::().is_err()); + } + #[test] fn test_deserialize_borrowed_null_as_bytes_errors() { // NULL through the borrowed deserialize_bytes path must error, not panic. diff --git a/src/sql_builder.rs b/src/sql_builder.rs index 42a56c5..3c01907 100644 --- a/src/sql_builder.rs +++ b/src/sql_builder.rs @@ -627,6 +627,37 @@ mod version_preflight { Ok(()) } + // The `check_*` preflight and the pure `build_*` grammar are deliberately separate (the parser-only `no_std` build compiles the builders without the feature-gated preflight). These `prepare_*` helpers pair them so the "emit-this-SQL-implies-this-version" rule lives in one place instead of being re-paired by each connection backend. Backend-independent glue is shared, not duplicated across the libpq/rustls adapters. + + /// Preflight + build `CREATE_REPLICATION_SLOT`. + pub(crate) fn prepare_create_slot( + server_version: i32, + slot_name: &str, + slot_type: SlotType, + output_plugin: Option<&str>, + options: &ReplicationSlotOptions, + ) -> Result { + check_create_slot_version(server_version, slot_type, options)?; + build_create_slot_sql(slot_name, slot_type, output_plugin, options) + } + + /// Preflight + build `ALTER_REPLICATION_SLOT`. + pub(crate) fn prepare_alter_slot( + server_version: i32, + slot_name: &str, + two_phase: Option, + failover: Option, + ) -> Result { + check_alter_slot_version(server_version, two_phase)?; + build_alter_slot_sql(slot_name, two_phase, failover) + } + + /// Preflight + build `READ_REPLICATION_SLOT`. + pub(crate) fn prepare_read_slot(server_version: i32, slot_name: &str) -> Result { + check_read_slot_version(server_version)?; + build_read_slot_sql(slot_name) + } + /// Preflight `BASE_BACKUP` options against the server version. /// /// Gates: `INCREMENTAL` requires PostgreSQL 17+; any option at all switches the @@ -791,8 +822,7 @@ mod version_preflight { #[cfg(any(feature = "libpq", feature = "rustls-tls"))] pub(crate) use version_preflight::{ - check_alter_slot_version, check_base_backup_version, check_create_slot_version, - check_read_slot_version, + check_base_backup_version, prepare_alter_slot, prepare_create_slot, prepare_read_slot, }; /// Options for building a `CREATE SUBSCRIPTION` SQL statement. diff --git a/src/stream.rs b/src/stream.rs index a2adcc7..b60ed55 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -7035,7 +7035,6 @@ mod tests { assert_eq!(stream.state.last_received_lsn, 0xABCD); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_arms_transport_stop_at_or_past_lsn() { // Arming now lives in next_raw_event (decode_raw_frame is policy-free), so @@ -7057,16 +7056,11 @@ mod tests { assert_eq!(stream2.stop_at_reached, Some(Lsn::new(0x2000))); } - // ---- next_raw_event / next_wal_frame (connection-driven) ---- - // - // These drive the raw pump through a null connection pre-seeded with - // CopyData frames. Native-backend only: the frame-seeding seam reaches into - // the native driver's `pending` queue. The coverage job runs default - // features (rustls-tls), so these count toward the coverage gate. + // These drive the shared raw pump through a null connection pre-seeded with CopyData frames. Both backends now expose the frame-seeding seam (`null_for_testing_with_frames`), so these run on libpq and rustls-tls alike — the pump serves each frame from the pending queue before any FFI. + // The one exception below (`..._terminates_after_transport_stop`) drives CopyDone, a backend-specific FFI send that is unsafe on libpq's null-pointer conn, so it stays rustls-tls-only (integration-tested for libpq). /// Build a test stream whose connection serves `frames` in order from its /// `pending` queue, so the connection-driven raw path is unit-testable. - #[cfg(feature = "rustls-tls")] fn create_test_stream_with_frames( config: ReplicationStreamConfig, frames: Vec, @@ -7076,7 +7070,6 @@ mod tests { stream } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_returns_seeded_wal_payload() { let payload = build_begin_payload(0x2000, 7); @@ -7092,7 +7085,6 @@ mod tests { assert_eq!(stream.state.last_received_lsn, 0x2000); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_returns_header_only_frame() { // A header-only 'w' frame (len == 25) is returned directly with an empty @@ -7107,7 +7099,6 @@ mod tests { assert!(raw.data.is_empty()); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_handles_keepalive_then_wal() { // A keepalive ('k', no reply requested) is consumed internally and @@ -7127,7 +7118,6 @@ mod tests { assert_eq!(stream.state.last_received_lsn, 0x2000); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_skips_empty_and_unknown_frames() { // An empty CopyData payload is skipped; an unknown message type is logged @@ -7148,7 +7138,6 @@ mod tests { assert_eq!(raw.wal_end, Lsn::new(0x2000)); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_cancelled_returns_error() { // A pre-cancelled token short-circuits the pump before any read. @@ -7160,7 +7149,6 @@ mod tests { assert!(matches!(err, ReplicationError::Cancelled(_))); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_raw_event_runs_periodic_feedback_check() { // Pre-arm the throttle counter so the next pumped frame lands on the @@ -7199,7 +7187,6 @@ mod tests { assert!(stream.copy_done_sent); } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_event_returns_decoded_event() { // Drive the TYPED path end-to-end through the seeded pump: a Begin frame @@ -7216,7 +7203,6 @@ mod tests { } } - #[cfg(feature = "rustls-tls")] #[tokio::test] async fn next_event_arms_commit_boundary_stop() { // A Commit whose end_lsn reaches stop_at_lsn is delivered AND arms the