LCOV - code coverage report
Current view: top level - libs/pq_proto/src - lib.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 52.9 % 622 329
Test Date: 2025-02-20 13:11:02 Functions: 47.4 % 114 54

            Line data    Source code
       1              : //! Postgres protocol messages serialization-deserialization. See
       2              : //! <https://www.postgresql.org/docs/devel/protocol-message-formats.html>
       3              : //! on message formats.
       4              : #![deny(clippy::undocumented_unsafe_blocks)]
       5              : 
       6              : pub mod framed;
       7              : 
       8              : use byteorder::{BigEndian, ReadBytesExt};
       9              : use bytes::{Buf, BufMut, Bytes, BytesMut};
      10              : use itertools::Itertools;
      11              : use serde::{Deserialize, Serialize};
      12              : use std::{borrow::Cow, fmt, io, str};
      13              : 
      14              : // re-export for use in utils pageserver_feedback.rs
      15              : pub use postgres_protocol::PG_EPOCH;
      16              : 
      17              : pub type Oid = u32;
      18              : pub type SystemId = u64;
      19              : 
      20              : pub const INT8_OID: Oid = 20;
      21              : pub const INT4_OID: Oid = 23;
      22              : pub const TEXT_OID: Oid = 25;
      23              : 
      24              : #[derive(Debug)]
      25              : pub enum FeMessage {
      26              :     // Simple query.
      27              :     Query(Bytes),
      28              :     // Extended query protocol.
      29              :     Parse(FeParseMessage),
      30              :     Describe(FeDescribeMessage),
      31              :     Bind(FeBindMessage),
      32              :     Execute(FeExecuteMessage),
      33              :     Close(FeCloseMessage),
      34              :     Sync,
      35              :     Terminate,
      36              :     CopyData(Bytes),
      37              :     CopyDone,
      38              :     CopyFail,
      39              :     PasswordMessage(Bytes),
      40              : }
      41              : 
      42              : #[derive(Clone, Copy, PartialEq, PartialOrd)]
      43              : pub struct ProtocolVersion(u32);
      44              : 
      45              : impl ProtocolVersion {
      46            0 :     pub const fn new(major: u16, minor: u16) -> Self {
      47            0 :         Self(((major as u32) << 16) | minor as u32)
      48            0 :     }
      49            0 :     pub const fn minor(self) -> u16 {
      50            0 :         self.0 as u16
      51            0 :     }
      52           24 :     pub const fn major(self) -> u16 {
      53           24 :         (self.0 >> 16) as u16
      54           24 :     }
      55              : }
      56              : 
      57              : impl fmt::Debug for ProtocolVersion {
      58            0 :     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
      59            0 :         f.debug_list()
      60            0 :             .entry(&self.major())
      61            0 :             .entry(&self.minor())
      62            0 :             .finish()
      63            0 :     }
      64              : }
      65              : 
      66              : #[derive(Debug)]
      67              : pub enum FeStartupPacket {
      68              :     CancelRequest(CancelKeyData),
      69              :     SslRequest {
      70              :         direct: bool,
      71              :     },
      72              :     GssEncRequest,
      73              :     StartupMessage {
      74              :         version: ProtocolVersion,
      75              :         params: StartupMessageParams,
      76              :     },
      77              : }
      78              : 
      79              : #[derive(Debug, Clone, Default)]
      80              : pub struct StartupMessageParamsBuilder {
      81              :     params: BytesMut,
      82              : }
      83              : 
      84              : impl StartupMessageParamsBuilder {
      85              :     /// Set parameter's value by its name.
      86              :     /// name and value must not contain a \0 byte
      87           25 :     pub fn insert(&mut self, name: &str, value: &str) {
      88           25 :         self.params.put(name.as_bytes());
      89           25 :         self.params.put(&b"\0"[..]);
      90           25 :         self.params.put(value.as_bytes());
      91           25 :         self.params.put(&b"\0"[..]);
      92           25 :     }
      93              : 
      94           17 :     pub fn freeze(self) -> StartupMessageParams {
      95           17 :         StartupMessageParams {
      96           17 :             params: self.params.freeze(),
      97           17 :         }
      98           17 :     }
      99              : }
     100              : 
     101              : #[derive(Debug, Clone, Default)]
     102              : pub struct StartupMessageParams {
     103              :     pub params: Bytes,
     104              : }
     105              : 
     106              : impl StartupMessageParams {
     107              :     /// Get parameter's value by its name.
     108           42 :     pub fn get(&self, name: &str) -> Option<&str> {
     109           58 :         self.iter().find_map(|(k, v)| (k == name).then_some(v))
     110           42 :     }
     111              : 
     112              :     /// Split command-line options according to PostgreSQL's logic,
     113              :     /// taking into account all escape sequences but leaving them as-is.
     114              :     /// [`None`] means that there's no `options` in [`Self`].
     115           24 :     pub fn options_raw(&self) -> Option<impl Iterator<Item = &str>> {
     116           24 :         self.get("options").map(Self::parse_options_raw)
     117           24 :     }
     118              : 
     119              :     /// Split command-line options according to PostgreSQL's logic,
     120              :     /// applying all escape sequences (using owned strings as needed).
     121              :     /// [`None`] means that there's no `options` in [`Self`].
     122            5 :     pub fn options_escaped(&self) -> Option<impl Iterator<Item = Cow<'_, str>>> {
     123            5 :         self.get("options").map(Self::parse_options_escaped)
     124            5 :     }
     125              : 
     126              :     /// Split command-line options according to PostgreSQL's logic,
     127              :     /// taking into account all escape sequences but leaving them as-is.
     128           30 :     pub fn parse_options_raw(input: &str) -> impl Iterator<Item = &str> {
     129           30 :         // See `postgres: pg_split_opts`.
     130           30 :         let mut last_was_escape = false;
     131           30 :         input
     132          594 :             .split(move |c: char| {
     133              :                 // We split by non-escaped whitespace symbols.
     134          594 :                 let should_split = c.is_ascii_whitespace() && !last_was_escape;
     135          594 :                 last_was_escape = c == '\\' && !last_was_escape;
     136          594 :                 should_split
     137          594 :             })
     138           77 :             .filter(|s| !s.is_empty())
     139           30 :     }
     140              : 
     141              :     /// Split command-line options according to PostgreSQL's logic,
     142              :     /// applying all escape sequences (using owned strings as needed).
     143            4 :     pub fn parse_options_escaped(input: &str) -> impl Iterator<Item = Cow<'_, str>> {
     144            4 :         // See `postgres: pg_split_opts`.
     145            7 :         Self::parse_options_raw(input).map(|s| {
     146            7 :             let mut preserve_next_escape = false;
     147           17 :             let escape = |c| {
     148              :                 // We should remove '\\' unless it's preceded by '\\'.
     149           17 :                 let should_remove = c == '\\' && !preserve_next_escape;
     150           17 :                 preserve_next_escape = should_remove;
     151           17 :                 should_remove
     152           17 :             };
     153              : 
     154            7 :             match s.contains('\\') {
     155            3 :                 true => Cow::Owned(s.replace(escape, "")),
     156            4 :                 false => Cow::Borrowed(s),
     157              :             }
     158            7 :         })
     159            4 :     }
     160              : 
     161              :     /// Iterate through key-value pairs in an arbitrary order.
     162           42 :     pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
     163           42 :         let params =
     164           42 :             std::str::from_utf8(&self.params).expect("should be validated as utf8 already");
     165           42 :         params.split_terminator('\0').tuples()
     166           42 :     }
     167              : 
     168              :     // This function is mostly useful in tests.
     169              :     #[doc(hidden)]
     170           17 :     pub fn new<'a, const N: usize>(pairs: [(&'a str, &'a str); N]) -> Self {
     171           17 :         let mut b = StartupMessageParamsBuilder::default();
     172           42 :         for (k, v) in pairs {
     173           25 :             b.insert(k, v)
     174              :         }
     175           17 :         b.freeze()
     176           17 :     }
     177              : }
     178              : 
     179            0 : #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy, Serialize, Deserialize)]
     180              : pub struct CancelKeyData {
     181              :     pub backend_pid: i32,
     182              :     pub cancel_key: i32,
     183              : }
     184              : 
     185            1 : pub fn id_to_cancel_key(id: u64) -> CancelKeyData {
     186            1 :     CancelKeyData {
     187            1 :         backend_pid: (id >> 32) as i32,
     188            1 :         cancel_key: (id & 0xffffffff) as i32,
     189            1 :     }
     190            1 : }
     191              : 
     192              : impl fmt::Display for CancelKeyData {
     193            1 :     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
     194            1 :         let hi = (self.backend_pid as u64) << 32;
     195            1 :         let lo = (self.cancel_key as u64) & 0xffffffff;
     196            1 :         let id = hi | lo;
     197            1 : 
     198            1 :         // This format is more compact and might work better for logs.
     199            1 :         f.debug_tuple("CancelKeyData")
     200            1 :             .field(&format_args!("{:x}", id))
     201            1 :             .finish()
     202            1 :     }
     203              : }
     204              : 
     205              : use rand::distributions::{Distribution, Standard};
     206              : impl Distribution<CancelKeyData> for Standard {
     207            0 :     fn sample<R: rand::Rng + ?Sized>(&self, rng: &mut R) -> CancelKeyData {
     208            0 :         CancelKeyData {
     209            0 :             backend_pid: rng.gen(),
     210            0 :             cancel_key: rng.gen(),
     211            0 :         }
     212            0 :     }
     213              : }
     214              : 
     215              : // We only support the simple case of Parse on unnamed prepared statement and
     216              : // no params
     217              : #[derive(Debug)]
     218              : pub struct FeParseMessage {
     219              :     pub query_string: Bytes,
     220              : }
     221              : 
     222              : #[derive(Debug)]
     223              : pub struct FeDescribeMessage {
     224              :     pub kind: u8, // 'S' to describe a prepared statement; or 'P' to describe a portal.
     225              :                   // we only support unnamed prepared stmt or portal
     226              : }
     227              : 
     228              : // we only support unnamed prepared stmt and portal
     229              : #[derive(Debug)]
     230              : pub struct FeBindMessage;
     231              : 
     232              : // we only support unnamed prepared stmt or portal
     233              : #[derive(Debug)]
     234              : pub struct FeExecuteMessage {
     235              :     /// max # of rows
     236              :     pub maxrows: i32,
     237              : }
     238              : 
     239              : // we only support unnamed prepared stmt and portal
     240              : #[derive(Debug)]
     241              : pub struct FeCloseMessage;
     242              : 
     243              : /// An error occurred while parsing or serializing raw stream into Postgres
     244              : /// messages.
     245              : #[derive(thiserror::Error, Debug)]
     246              : pub enum ProtocolError {
     247              :     /// Invalid packet was received from the client (e.g. unexpected message
     248              :     /// type or broken len).
     249              :     #[error("Protocol error: {0}")]
     250              :     Protocol(String),
     251              :     /// Failed to parse or, (unlikely), serialize a protocol message.
     252              :     #[error("Message parse error: {0}")]
     253              :     BadMessage(String),
     254              : }
     255              : 
     256              : impl ProtocolError {
     257              :     /// Proxy stream.rs uses only io::Error; provide it.
     258            0 :     pub fn into_io_error(self) -> io::Error {
     259            0 :         io::Error::new(io::ErrorKind::Other, self.to_string())
     260            0 :     }
     261              : }
     262              : 
     263              : impl FeMessage {
     264              :     /// Read and parse one message from the `buf` input buffer. If there is at
     265              :     /// least one valid message, returns it, advancing `buf`; redundant copies
     266              :     /// are avoided, as thanks to `bytes` crate ptrs in parsed message point
     267              :     /// directly into the `buf` (processed data is garbage collected after
     268              :     /// parsed message is dropped).
     269              :     ///
     270              :     /// Returns None if `buf` doesn't contain enough data for a single message.
     271              :     /// For efficiency, tries to reserve large enough space in `buf` for the
     272              :     /// next message in this case to save the repeated calls.
     273              :     ///
     274              :     /// Returns Error if message is malformed, the only possible ErrorKind is
     275              :     /// InvalidInput.
     276              :     //
     277              :     // Inspired by rust-postgres Message::parse.
     278           57 :     pub fn parse(buf: &mut BytesMut) -> Result<Option<FeMessage>, ProtocolError> {
     279           57 :         // Every message contains message type byte and 4 bytes len; can't do
     280           57 :         // much without them.
     281           57 :         if buf.len() < 5 {
     282           30 :             let to_read = 5 - buf.len();
     283           30 :             buf.reserve(to_read);
     284           30 :             return Ok(None);
     285           27 :         }
     286           27 : 
     287           27 :         // We shouldn't advance `buf` as probably full message is not there yet,
     288           27 :         // so can't directly use Bytes::get_u32 etc.
     289           27 :         let tag = buf[0];
     290           27 :         let len = (&buf[1..5]).read_u32::<BigEndian>().unwrap();
     291           27 :         if len < 4 {
     292            0 :             return Err(ProtocolError::Protocol(format!(
     293            0 :                 "invalid message length {}",
     294            0 :                 len
     295            0 :             )));
     296           27 :         }
     297           27 : 
     298           27 :         // length field includes itself, but not message type.
     299           27 :         let total_len = len as usize + 1;
     300           27 :         if buf.len() < total_len {
     301              :             // Don't have full message yet.
     302            0 :             let to_read = total_len - buf.len();
     303            0 :             buf.reserve(to_read);
     304            0 :             return Ok(None);
     305           27 :         }
     306           27 : 
     307           27 :         // got the message, advance buffer
     308           27 :         let mut msg = buf.split_to(total_len).freeze();
     309           27 :         msg.advance(5); // consume message type and len
     310           27 : 
     311           27 :         match tag {
     312            2 :             b'Q' => Ok(Some(FeMessage::Query(msg))),
     313            0 :             b'P' => Ok(Some(FeParseMessage::parse(msg)?)),
     314            0 :             b'D' => Ok(Some(FeDescribeMessage::parse(msg)?)),
     315            0 :             b'E' => Ok(Some(FeExecuteMessage::parse(msg)?)),
     316            0 :             b'B' => Ok(Some(FeBindMessage::parse(msg)?)),
     317            0 :             b'C' => Ok(Some(FeCloseMessage::parse(msg)?)),
     318            0 :             b'S' => Ok(Some(FeMessage::Sync)),
     319            0 :             b'X' => Ok(Some(FeMessage::Terminate)),
     320            0 :             b'd' => Ok(Some(FeMessage::CopyData(msg))),
     321            0 :             b'c' => Ok(Some(FeMessage::CopyDone)),
     322            0 :             b'f' => Ok(Some(FeMessage::CopyFail)),
     323           25 :             b'p' => Ok(Some(FeMessage::PasswordMessage(msg))),
     324            0 :             tag => Err(ProtocolError::Protocol(format!(
     325            0 :                 "unknown message tag: {tag},'{msg:?}'"
     326            0 :             ))),
     327              :         }
     328           57 :     }
     329              : }
     330              : 
     331              : impl FeStartupPacket {
     332              :     /// Read and parse startup message from the `buf` input buffer. It is
     333              :     /// different from [`FeMessage::parse`] because startup messages don't have
     334              :     /// message type byte; otherwise, its comments apply.
     335           91 :     pub fn parse(buf: &mut BytesMut) -> Result<Option<FeStartupPacket>, ProtocolError> {
     336              :         /// <https://github.com/postgres/postgres/blob/ca481d3c9ab7bf69ff0c8d71ad3951d407f6a33c/src/include/libpq/pqcomm.h#L118>
     337              :         const MAX_STARTUP_PACKET_LENGTH: usize = 10000;
     338              :         const RESERVED_INVALID_MAJOR_VERSION: u16 = 1234;
     339              :         /// <https://github.com/postgres/postgres/blob/ca481d3c9ab7bf69ff0c8d71ad3951d407f6a33c/src/include/libpq/pqcomm.h#L132>
     340              :         const CANCEL_REQUEST_CODE: ProtocolVersion = ProtocolVersion::new(1234, 5678);
     341              :         /// <https://github.com/postgres/postgres/blob/ca481d3c9ab7bf69ff0c8d71ad3951d407f6a33c/src/include/libpq/pqcomm.h#L166>
     342              :         const NEGOTIATE_SSL_CODE: ProtocolVersion = ProtocolVersion::new(1234, 5679);
     343              :         /// <https://github.com/postgres/postgres/blob/ca481d3c9ab7bf69ff0c8d71ad3951d407f6a33c/src/include/libpq/pqcomm.h#L167>
     344              :         const NEGOTIATE_GSS_CODE: ProtocolVersion = ProtocolVersion::new(1234, 5680);
     345              : 
     346              :         // <https://github.com/postgres/postgres/blob/04bcf9e19a4261fe9c7df37c777592c2e10c32a7/src/backend/tcop/backend_startup.c#L378-L382>
     347              :         // First byte indicates standard SSL handshake message
     348              :         // (It can't be a Postgres startup length because in network byte order
     349              :         // that would be a startup packet hundreds of megabytes long)
     350           91 :         if buf.first() == Some(&0x16) {
     351            0 :             return Ok(Some(FeStartupPacket::SslRequest { direct: true }));
     352           91 :         }
     353           91 : 
     354           91 :         // need at least 4 bytes with packet len
     355           91 :         if buf.len() < 4 {
     356           45 :             let to_read = 4 - buf.len();
     357           45 :             buf.reserve(to_read);
     358           45 :             return Ok(None);
     359           46 :         }
     360           46 : 
     361           46 :         // We shouldn't advance `buf` as probably full message is not there yet,
     362           46 :         // so can't directly use Bytes::get_u32 etc.
     363           46 :         let len = (&buf[0..4]).read_u32::<BigEndian>().unwrap() as usize;
     364           46 :         // The proposed replacement is `!(8..=MAX_STARTUP_PACKET_LENGTH).contains(&len)`
     365           46 :         // which is less readable
     366           46 :         #[allow(clippy::manual_range_contains)]
     367           46 :         if len < 8 || len > MAX_STARTUP_PACKET_LENGTH {
     368            1 :             return Err(ProtocolError::Protocol(format!(
     369            1 :                 "invalid startup packet message length {}",
     370            1 :                 len
     371            1 :             )));
     372           45 :         }
     373           45 : 
     374           45 :         if buf.len() < len {
     375              :             // Don't have full message yet.
     376            0 :             let to_read = len - buf.len();
     377            0 :             buf.reserve(to_read);
     378            0 :             return Ok(None);
     379           45 :         }
     380           45 : 
     381           45 :         // got the message, advance buffer
     382           45 :         let mut msg = buf.split_to(len).freeze();
     383           45 :         msg.advance(4); // consume len
     384           45 : 
     385           45 :         let request_code = ProtocolVersion(msg.get_u32());
     386              :         // StartupMessage, CancelRequest, SSLRequest etc are differentiated by request code.
     387           45 :         let message = match request_code {
     388              :             CANCEL_REQUEST_CODE => {
     389            0 :                 if msg.remaining() != 8 {
     390            0 :                     return Err(ProtocolError::BadMessage(
     391            0 :                         "CancelRequest message is malformed, backend PID / secret key missing"
     392            0 :                             .to_owned(),
     393            0 :                     ));
     394            0 :                 }
     395            0 :                 FeStartupPacket::CancelRequest(CancelKeyData {
     396            0 :                     backend_pid: msg.get_i32(),
     397            0 :                     cancel_key: msg.get_i32(),
     398            0 :                 })
     399              :             }
     400              :             NEGOTIATE_SSL_CODE => {
     401              :                 // Requested upgrade to SSL (aka TLS)
     402           21 :                 FeStartupPacket::SslRequest { direct: false }
     403              :             }
     404              :             NEGOTIATE_GSS_CODE => {
     405              :                 // Requested upgrade to GSSAPI
     406            0 :                 FeStartupPacket::GssEncRequest
     407              :             }
     408           24 :             version if version.major() == RESERVED_INVALID_MAJOR_VERSION => {
     409            0 :                 return Err(ProtocolError::Protocol(format!(
     410            0 :                     "Unrecognized request code {}",
     411            0 :                     version.minor()
     412            0 :                 )));
     413              :             }
     414              :             // TODO bail if protocol major_version is not 3?
     415           24 :             version => {
     416              :                 // StartupMessage
     417              : 
     418           24 :                 let s = str::from_utf8(&msg).map_err(|_e| {
     419            0 :                     ProtocolError::BadMessage("StartupMessage params: invalid utf-8".to_owned())
     420           24 :                 })?;
     421           24 :                 let s = s.strip_suffix('\0').ok_or_else(|| {
     422            0 :                     ProtocolError::Protocol(
     423            0 :                         "StartupMessage params: missing null terminator".to_string(),
     424            0 :                     )
     425           24 :                 })?;
     426              : 
     427           24 :                 FeStartupPacket::StartupMessage {
     428           24 :                     version,
     429           24 :                     params: StartupMessageParams {
     430           24 :                         params: msg.slice_ref(s.as_bytes()),
     431           24 :                     },
     432           24 :                 }
     433              :             }
     434              :         };
     435           45 :         Ok(Some(message))
     436           91 :     }
     437              : }
     438              : 
     439              : impl FeParseMessage {
     440            0 :     fn parse(mut buf: Bytes) -> Result<FeMessage, ProtocolError> {
     441              :         // FIXME: the rust-postgres driver uses a named prepared statement
     442              :         // for copy_out(). We're not prepared to handle that correctly. For
     443              :         // now, just ignore the statement name, assuming that the client never
     444              :         // uses more than one prepared statement at a time.
     445              : 
     446            0 :         let _pstmt_name = read_cstr(&mut buf)?;
     447            0 :         let query_string = read_cstr(&mut buf)?;
     448            0 :         if buf.remaining() < 2 {
     449            0 :             return Err(ProtocolError::BadMessage(
     450            0 :                 "Parse message is malformed, nparams missing".to_string(),
     451            0 :             ));
     452            0 :         }
     453            0 :         let nparams = buf.get_i16();
     454            0 : 
     455            0 :         if nparams != 0 {
     456            0 :             return Err(ProtocolError::BadMessage(
     457            0 :                 "query params not implemented".to_string(),
     458            0 :             ));
     459            0 :         }
     460            0 : 
     461            0 :         Ok(FeMessage::Parse(FeParseMessage { query_string }))
     462            0 :     }
     463              : }
     464              : 
     465              : impl FeDescribeMessage {
     466            0 :     fn parse(mut buf: Bytes) -> Result<FeMessage, ProtocolError> {
     467            0 :         let kind = buf.get_u8();
     468            0 :         let _pstmt_name = read_cstr(&mut buf)?;
     469              : 
     470              :         // FIXME: see FeParseMessage::parse
     471            0 :         if kind != b'S' {
     472            0 :             return Err(ProtocolError::BadMessage(
     473            0 :                 "only prepared statemement Describe is implemented".to_string(),
     474            0 :             ));
     475            0 :         }
     476            0 : 
     477            0 :         Ok(FeMessage::Describe(FeDescribeMessage { kind }))
     478            0 :     }
     479              : }
     480              : 
     481              : impl FeExecuteMessage {
     482            0 :     fn parse(mut buf: Bytes) -> Result<FeMessage, ProtocolError> {
     483            0 :         let portal_name = read_cstr(&mut buf)?;
     484            0 :         if buf.remaining() < 4 {
     485            0 :             return Err(ProtocolError::BadMessage(
     486            0 :                 "FeExecuteMessage message is malformed, maxrows missing".to_string(),
     487            0 :             ));
     488            0 :         }
     489            0 :         let maxrows = buf.get_i32();
     490            0 : 
     491            0 :         if !portal_name.is_empty() {
     492            0 :             return Err(ProtocolError::BadMessage(
     493            0 :                 "named portals not implemented".to_string(),
     494            0 :             ));
     495            0 :         }
     496            0 :         if maxrows != 0 {
     497            0 :             return Err(ProtocolError::BadMessage(
     498            0 :                 "row limit in Execute message not implemented".to_string(),
     499            0 :             ));
     500            0 :         }
     501            0 : 
     502            0 :         Ok(FeMessage::Execute(FeExecuteMessage { maxrows }))
     503            0 :     }
     504              : }
     505              : 
     506              : impl FeBindMessage {
     507            0 :     fn parse(mut buf: Bytes) -> Result<FeMessage, ProtocolError> {
     508            0 :         let portal_name = read_cstr(&mut buf)?;
     509            0 :         let _pstmt_name = read_cstr(&mut buf)?;
     510              : 
     511              :         // FIXME: see FeParseMessage::parse
     512            0 :         if !portal_name.is_empty() {
     513            0 :             return Err(ProtocolError::BadMessage(
     514            0 :                 "named portals not implemented".to_string(),
     515            0 :             ));
     516            0 :         }
     517            0 : 
     518            0 :         Ok(FeMessage::Bind(FeBindMessage))
     519            0 :     }
     520              : }
     521              : 
     522              : impl FeCloseMessage {
     523            0 :     fn parse(mut buf: Bytes) -> Result<FeMessage, ProtocolError> {
     524            0 :         let _kind = buf.get_u8();
     525            0 :         let _pstmt_or_portal_name = read_cstr(&mut buf)?;
     526              : 
     527              :         // FIXME: we do nothing with Close
     528            0 :         Ok(FeMessage::Close(FeCloseMessage))
     529            0 :     }
     530              : }
     531              : 
     532              : // Backend
     533              : 
     534              : #[derive(Debug)]
     535              : pub enum BeMessage<'a> {
     536              :     AuthenticationOk,
     537              :     AuthenticationMD5Password([u8; 4]),
     538              :     AuthenticationSasl(BeAuthenticationSaslMessage<'a>),
     539              :     AuthenticationCleartextPassword,
     540              :     BackendKeyData(CancelKeyData),
     541              :     BindComplete,
     542              :     CommandComplete(&'a [u8]),
     543              :     CopyData(&'a [u8]),
     544              :     CopyDone,
     545              :     CopyFail,
     546              :     CopyInResponse,
     547              :     CopyOutResponse,
     548              :     CopyBothResponse,
     549              :     CloseComplete,
     550              :     // None means column is NULL
     551              :     DataRow(&'a [Option<&'a [u8]>]),
     552              :     // None errcode means internal_error will be sent.
     553              :     ErrorResponse(&'a str, Option<&'a [u8; 5]>),
     554              :     /// Single byte - used in response to SSLRequest/GSSENCRequest.
     555              :     EncryptionResponse(bool),
     556              :     NoData,
     557              :     ParameterDescription,
     558              :     ParameterStatus {
     559              :         name: &'a [u8],
     560              :         value: &'a [u8],
     561              :     },
     562              :     ParseComplete,
     563              :     ReadyForQuery,
     564              :     RowDescription(&'a [RowDescriptor<'a>]),
     565              :     XLogData(XLogDataBody<'a>),
     566              :     NoticeResponse(&'a str),
     567              :     NegotiateProtocolVersion {
     568              :         version: ProtocolVersion,
     569              :         options: &'a [&'a str],
     570              :     },
     571              :     KeepAlive(WalSndKeepAlive),
     572              :     /// Batch of interpreted, shard filtered WAL records,
     573              :     /// ready for the pageserver to ingest
     574              :     InterpretedWalRecords(InterpretedWalRecordsBody<'a>),
     575              : 
     576              :     Raw(u8, &'a [u8]),
     577              : }
     578              : 
     579              : /// Common shorthands.
     580              : impl<'a> BeMessage<'a> {
     581              :     /// A [`BeMessage::ParameterStatus`] holding the client encoding, i.e. UTF-8.
     582              :     /// This is a sensible default, given that:
     583              :     ///  * rust strings only support this encoding out of the box.
     584              :     ///  * tokio-postgres, postgres-jdbc (and probably more) mandate it.
     585              :     ///
     586              :     /// TODO: do we need to report `server_encoding` as well?
     587              :     pub const CLIENT_ENCODING: Self = Self::ParameterStatus {
     588              :         name: b"client_encoding",
     589              :         value: b"UTF8",
     590              :     };
     591              : 
     592              :     pub const INTEGER_DATETIMES: Self = Self::ParameterStatus {
     593              :         name: b"integer_datetimes",
     594              :         value: b"on",
     595              :     };
     596              : 
     597              :     /// Build a [`BeMessage::ParameterStatus`] holding the server version.
     598            2 :     pub fn server_version(version: &'a str) -> Self {
     599            2 :         Self::ParameterStatus {
     600            2 :             name: b"server_version",
     601            2 :             value: version.as_bytes(),
     602            2 :         }
     603            2 :     }
     604              : }
     605              : 
     606              : #[derive(Debug)]
     607              : pub enum BeAuthenticationSaslMessage<'a> {
     608              :     Methods(&'a [&'a str]),
     609              :     Continue(&'a [u8]),
     610              :     Final(&'a [u8]),
     611              : }
     612              : 
     613              : #[derive(Debug)]
     614              : pub enum BeParameterStatusMessage<'a> {
     615              :     Encoding(&'a str),
     616              :     ServerVersion(&'a str),
     617              : }
     618              : 
     619              : // One row description in RowDescription packet.
     620              : #[derive(Debug)]
     621              : pub struct RowDescriptor<'a> {
     622              :     pub name: &'a [u8],
     623              :     pub tableoid: Oid,
     624              :     pub attnum: i16,
     625              :     pub typoid: Oid,
     626              :     pub typlen: i16,
     627              :     pub typmod: i32,
     628              :     pub formatcode: i16,
     629              : }
     630              : 
     631              : impl Default for RowDescriptor<'_> {
     632            0 :     fn default() -> RowDescriptor<'static> {
     633            0 :         RowDescriptor {
     634            0 :             name: b"",
     635            0 :             tableoid: 0,
     636            0 :             attnum: 0,
     637            0 :             typoid: 0,
     638            0 :             typlen: 0,
     639            0 :             typmod: 0,
     640            0 :             formatcode: 0,
     641            0 :         }
     642            0 :     }
     643              : }
     644              : 
     645              : impl RowDescriptor<'_> {
     646              :     /// Convenience function to create a RowDescriptor message for an int8 column
     647            0 :     pub const fn int8_col(name: &[u8]) -> RowDescriptor {
     648            0 :         RowDescriptor {
     649            0 :             name,
     650            0 :             tableoid: 0,
     651            0 :             attnum: 0,
     652            0 :             typoid: INT8_OID,
     653            0 :             typlen: 8,
     654            0 :             typmod: 0,
     655            0 :             formatcode: 0,
     656            0 :         }
     657            0 :     }
     658              : 
     659            2 :     pub const fn text_col(name: &[u8]) -> RowDescriptor {
     660            2 :         RowDescriptor {
     661            2 :             name,
     662            2 :             tableoid: 0,
     663            2 :             attnum: 0,
     664            2 :             typoid: TEXT_OID,
     665            2 :             typlen: -1,
     666            2 :             typmod: 0,
     667            2 :             formatcode: 0,
     668            2 :         }
     669            2 :     }
     670              : }
     671              : 
     672              : #[derive(Debug)]
     673              : pub struct XLogDataBody<'a> {
     674              :     pub wal_start: u64,
     675              :     pub wal_end: u64, // current end of WAL on the server
     676              :     pub timestamp: i64,
     677              :     pub data: &'a [u8],
     678              : }
     679              : 
     680              : #[derive(Debug)]
     681              : pub struct WalSndKeepAlive {
     682              :     pub wal_end: u64, // current end of WAL on the server
     683              :     pub timestamp: i64,
     684              :     pub request_reply: bool,
     685              : }
     686              : 
     687              : /// Batch of interpreted WAL records used in the interpreted
     688              : /// safekeeper to pageserver protocol.
     689              : ///
     690              : /// Note that the pageserver uses the RawInterpretedWalRecordsBody
     691              : /// counterpart of this from the neondatabase/rust-postgres repo.
     692              : /// If you're changing this struct, you likely need to change its
     693              : /// twin as well.
     694              : #[derive(Debug)]
     695              : pub struct InterpretedWalRecordsBody<'a> {
     696              :     /// End of raw WAL in [`Self::data`]
     697              :     pub streaming_lsn: u64,
     698              :     /// Current end of WAL on the server
     699              :     pub commit_lsn: u64,
     700              :     pub data: &'a [u8],
     701              : }
     702              : 
     703              : pub static HELLO_WORLD_ROW: BeMessage = BeMessage::DataRow(&[Some(b"hello world")]);
     704              : 
     705              : // single text column
     706              : pub static SINGLE_COL_ROWDESC: BeMessage = BeMessage::RowDescription(&[RowDescriptor {
     707              :     name: b"data",
     708              :     tableoid: 0,
     709              :     attnum: 0,
     710              :     typoid: TEXT_OID,
     711              :     typlen: -1,
     712              :     typmod: 0,
     713              :     formatcode: 0,
     714              : }]);
     715              : 
     716              : /// Call f() to write body of the message and prepend it with 4-byte len as
     717              : /// prescribed by the protocol.
     718           75 : fn write_body<R>(buf: &mut BytesMut, f: impl FnOnce(&mut BytesMut) -> R) -> R {
     719           75 :     let base = buf.len();
     720           75 :     buf.extend_from_slice(&[0; 4]);
     721           75 : 
     722           75 :     let res = f(buf);
     723           75 : 
     724           75 :     let size = i32::try_from(buf.len() - base).expect("message too big to transmit");
     725           75 :     (&mut buf[base..]).put_slice(&size.to_be_bytes());
     726           75 : 
     727           75 :     res
     728           75 : }
     729              : 
     730              : /// Safe write of s into buf as cstring (String in the protocol).
     731           56 : fn write_cstr(s: impl AsRef<[u8]>, buf: &mut BytesMut) -> Result<(), ProtocolError> {
     732           56 :     let bytes = s.as_ref();
     733           56 :     if bytes.contains(&0) {
     734            0 :         return Err(ProtocolError::BadMessage(
     735            0 :             "string contains embedded null".to_owned(),
     736            0 :         ));
     737           56 :     }
     738           56 :     buf.put_slice(bytes);
     739           56 :     buf.put_u8(0);
     740           56 :     Ok(())
     741           56 : }
     742              : 
     743              : /// Read cstring from buf, advancing it.
     744           11 : pub fn read_cstr(buf: &mut Bytes) -> Result<Bytes, ProtocolError> {
     745           11 :     let pos = buf
     746           11 :         .iter()
     747          156 :         .position(|x| *x == 0)
     748           11 :         .ok_or_else(|| ProtocolError::BadMessage("missing cstring terminator".to_owned()))?;
     749           11 :     let result = buf.split_to(pos);
     750           11 :     buf.advance(1); // drop the null terminator
     751           11 :     Ok(result)
     752           11 : }
     753              : 
     754              : pub const SQLSTATE_INTERNAL_ERROR: &[u8; 5] = b"XX000";
     755              : pub const SQLSTATE_ADMIN_SHUTDOWN: &[u8; 5] = b"57P01";
     756              : pub const SQLSTATE_SUCCESSFUL_COMPLETION: &[u8; 5] = b"00000";
     757              : 
     758              : impl BeMessage<'_> {
     759              :     /// Serialize `message` to the given `buf`.
     760              :     /// Apart from smart memory managemet, BytesMut is good here as msg len
     761              :     /// precedes its body and it is handy to write it down first and then fill
     762              :     /// the length. With Write we would have to either calc it manually or have
     763              :     /// one more buffer.
     764           96 :     pub fn write(buf: &mut BytesMut, message: &BeMessage) -> Result<(), ProtocolError> {
     765           96 :         match message {
     766            0 :             BeMessage::Raw(code, data) => {
     767            0 :                 buf.put_u8(*code);
     768            0 :                 write_body(buf, |b| b.put_slice(data))
     769              :             }
     770           12 :             BeMessage::AuthenticationOk => {
     771           12 :                 buf.put_u8(b'R');
     772           12 :                 write_body(buf, |buf| {
     773           12 :                     buf.put_i32(0); // Specifies that the authentication was successful.
     774           12 :                 });
     775           12 :             }
     776              : 
     777            2 :             BeMessage::AuthenticationCleartextPassword => {
     778            2 :                 buf.put_u8(b'R');
     779            2 :                 write_body(buf, |buf| {
     780            2 :                     buf.put_i32(3); // Specifies that clear text password is required.
     781            2 :                 });
     782            2 :             }
     783              : 
     784            0 :             BeMessage::AuthenticationMD5Password(salt) => {
     785            0 :                 buf.put_u8(b'R');
     786            0 :                 write_body(buf, |buf| {
     787            0 :                     buf.put_i32(5); // Specifies that an MD5-encrypted password is required.
     788            0 :                     buf.put_slice(&salt[..]);
     789            0 :                 });
     790            0 :             }
     791              : 
     792           30 :             BeMessage::AuthenticationSasl(msg) => {
     793           30 :                 buf.put_u8(b'R');
     794           30 :                 write_body(buf, |buf| {
     795              :                     use BeAuthenticationSaslMessage::*;
     796           30 :                     match msg {
     797           13 :                         Methods(methods) => {
     798           13 :                             buf.put_i32(10); // Specifies that SASL auth method is used.
     799           25 :                             for method in methods.iter() {
     800           25 :                                 write_cstr(method, buf)?;
     801              :                             }
     802           13 :                             buf.put_u8(0); // zero terminator for the list
     803              :                         }
     804           11 :                         Continue(extra) => {
     805           11 :                             buf.put_i32(11); // Continue SASL auth.
     806           11 :                             buf.put_slice(extra);
     807           11 :                         }
     808            6 :                         Final(extra) => {
     809            6 :                             buf.put_i32(12); // Send final SASL message.
     810            6 :                             buf.put_slice(extra);
     811            6 :                         }
     812              :                     }
     813           30 :                     Ok(())
     814           30 :                 })?;
     815              :             }
     816              : 
     817            0 :             BeMessage::BackendKeyData(key_data) => {
     818            0 :                 buf.put_u8(b'K');
     819            0 :                 write_body(buf, |buf| {
     820            0 :                     buf.put_i32(key_data.backend_pid);
     821            0 :                     buf.put_i32(key_data.cancel_key);
     822            0 :                 });
     823            0 :             }
     824              : 
     825            0 :             BeMessage::BindComplete => {
     826            0 :                 buf.put_u8(b'2');
     827            0 :                 write_body(buf, |_| {});
     828            0 :             }
     829              : 
     830            0 :             BeMessage::CloseComplete => {
     831            0 :                 buf.put_u8(b'3');
     832            0 :                 write_body(buf, |_| {});
     833            0 :             }
     834              : 
     835            2 :             BeMessage::CommandComplete(cmd) => {
     836            2 :                 buf.put_u8(b'C');
     837            2 :                 write_body(buf, |buf| write_cstr(cmd, buf))?;
     838              :             }
     839              : 
     840            0 :             BeMessage::CopyData(data) => {
     841            0 :                 buf.put_u8(b'd');
     842            0 :                 write_body(buf, |buf| {
     843            0 :                     buf.put_slice(data);
     844            0 :                 });
     845            0 :             }
     846              : 
     847            0 :             BeMessage::CopyDone => {
     848            0 :                 buf.put_u8(b'c');
     849            0 :                 write_body(buf, |_| {});
     850            0 :             }
     851              : 
     852            0 :             BeMessage::CopyFail => {
     853            0 :                 buf.put_u8(b'f');
     854            0 :                 write_body(buf, |_| {});
     855            0 :             }
     856              : 
     857            0 :             BeMessage::CopyInResponse => {
     858            0 :                 buf.put_u8(b'G');
     859            0 :                 write_body(buf, |buf| {
     860            0 :                     buf.put_u8(1); // copy_is_binary
     861            0 :                     buf.put_i16(0); // numAttributes
     862            0 :                 });
     863            0 :             }
     864              : 
     865            0 :             BeMessage::CopyOutResponse => {
     866            0 :                 buf.put_u8(b'H');
     867            0 :                 write_body(buf, |buf| {
     868            0 :                     buf.put_u8(0); // copy_is_binary
     869            0 :                     buf.put_i16(0); // numAttributes
     870            0 :                 });
     871            0 :             }
     872              : 
     873            0 :             BeMessage::CopyBothResponse => {
     874            0 :                 buf.put_u8(b'W');
     875            0 :                 write_body(buf, |buf| {
     876            0 :                     // doesn't matter, used only for replication
     877            0 :                     buf.put_u8(0); // copy_is_binary
     878            0 :                     buf.put_i16(0); // numAttributes
     879            0 :                 });
     880            0 :             }
     881              : 
     882            2 :             BeMessage::DataRow(vals) => {
     883            2 :                 buf.put_u8(b'D');
     884            2 :                 write_body(buf, |buf| {
     885            2 :                     buf.put_u16(vals.len() as u16); // num of cols
     886            2 :                     for val_opt in vals.iter() {
     887            2 :                         if let Some(val) = val_opt {
     888            2 :                             buf.put_u32(val.len() as u32);
     889            2 :                             buf.put_slice(val);
     890            2 :                         } else {
     891            0 :                             buf.put_i32(-1);
     892            0 :                         }
     893              :                     }
     894            2 :                 });
     895            2 :             }
     896              : 
     897              :             // ErrorResponse is a zero-terminated array of zero-terminated fields.
     898              :             // First byte of each field represents type of this field. Set just enough fields
     899              :             // to satisfy rust-postgres client: 'S' -- severity, 'C' -- error, 'M' -- error
     900              :             // message text.
     901            1 :             BeMessage::ErrorResponse(error_msg, pg_error_code) => {
     902            1 :                 // 'E' signalizes ErrorResponse messages
     903            1 :                 buf.put_u8(b'E');
     904            1 :                 write_body(buf, |buf| {
     905            1 :                     buf.put_u8(b'S'); // severity
     906            1 :                     buf.put_slice(b"ERROR\0");
     907            1 : 
     908            1 :                     buf.put_u8(b'C'); // SQLSTATE error code
     909            1 :                     buf.put_slice(&terminate_code(
     910            1 :                         pg_error_code.unwrap_or(SQLSTATE_INTERNAL_ERROR),
     911            1 :                     ));
     912            1 : 
     913            1 :                     buf.put_u8(b'M'); // the message
     914            1 :                     write_cstr(error_msg, buf)?;
     915              : 
     916            1 :                     buf.put_u8(0); // terminator
     917            1 :                     Ok(())
     918            1 :                 })?;
     919              :             }
     920              : 
     921              :             // NoticeResponse has the same format as ErrorResponse. From doc: "The frontend should display the
     922              :             // message but continue listening for ReadyForQuery or ErrorResponse"
     923            0 :             BeMessage::NoticeResponse(error_msg) => {
     924            0 :                 // For all the errors set Severity to Error and error code to
     925            0 :                 // 'internal error'.
     926            0 : 
     927            0 :                 // 'N' signalizes NoticeResponse messages
     928            0 :                 buf.put_u8(b'N');
     929            0 :                 write_body(buf, |buf| {
     930            0 :                     buf.put_u8(b'S'); // severity
     931            0 :                     buf.put_slice(b"NOTICE\0");
     932            0 : 
     933            0 :                     buf.put_u8(b'C'); // SQLSTATE error code
     934            0 :                     buf.put_slice(&terminate_code(SQLSTATE_INTERNAL_ERROR));
     935            0 : 
     936            0 :                     buf.put_u8(b'M'); // the message
     937            0 :                     write_cstr(error_msg.as_bytes(), buf)?;
     938              : 
     939            0 :                     buf.put_u8(0); // terminator
     940            0 :                     Ok(())
     941            0 :                 })?;
     942              :             }
     943              : 
     944            0 :             BeMessage::NoData => {
     945            0 :                 buf.put_u8(b'n');
     946            0 :                 write_body(buf, |_| {});
     947            0 :             }
     948              : 
     949           21 :             BeMessage::EncryptionResponse(should_negotiate) => {
     950           21 :                 let response = if *should_negotiate { b'S' } else { b'N' };
     951           21 :                 buf.put_u8(response);
     952              :             }
     953              : 
     954           13 :             BeMessage::ParameterStatus { name, value } => {
     955           13 :                 buf.put_u8(b'S');
     956           13 :                 write_body(buf, |buf| {
     957           13 :                     write_cstr(name, buf)?;
     958           13 :                     write_cstr(value, buf)
     959           13 :                 })?;
     960              :             }
     961              : 
     962            0 :             BeMessage::ParameterDescription => {
     963            0 :                 buf.put_u8(b't');
     964            0 :                 write_body(buf, |buf| {
     965            0 :                     // we don't support params, so always 0
     966            0 :                     buf.put_i16(0);
     967            0 :                 });
     968            0 :             }
     969              : 
     970            0 :             BeMessage::ParseComplete => {
     971            0 :                 buf.put_u8(b'1');
     972            0 :                 write_body(buf, |_| {});
     973            0 :             }
     974              : 
     975           11 :             BeMessage::ReadyForQuery => {
     976           11 :                 buf.put_u8(b'Z');
     977           11 :                 write_body(buf, |buf| {
     978           11 :                     buf.put_u8(b'I');
     979           11 :                 });
     980           11 :             }
     981              : 
     982            2 :             BeMessage::RowDescription(rows) => {
     983            2 :                 buf.put_u8(b'T');
     984            2 :                 write_body(buf, |buf| {
     985            2 :                     buf.put_i16(rows.len() as i16); // # of fields
     986            2 :                     for row in rows.iter() {
     987            2 :                         write_cstr(row.name, buf)?;
     988            2 :                         buf.put_i32(0); /* table oid */
     989            2 :                         buf.put_i16(0); /* attnum */
     990            2 :                         buf.put_u32(row.typoid);
     991            2 :                         buf.put_i16(row.typlen);
     992            2 :                         buf.put_i32(-1); /* typmod */
     993            2 :                         buf.put_i16(0); /* format code */
     994              :                     }
     995            2 :                     Ok(())
     996            2 :                 })?;
     997              :             }
     998              : 
     999            0 :             BeMessage::XLogData(body) => {
    1000            0 :                 buf.put_u8(b'd');
    1001            0 :                 write_body(buf, |buf| {
    1002            0 :                     buf.put_u8(b'w');
    1003            0 :                     buf.put_u64(body.wal_start);
    1004            0 :                     buf.put_u64(body.wal_end);
    1005            0 :                     buf.put_i64(body.timestamp);
    1006            0 :                     buf.put_slice(body.data);
    1007            0 :                 });
    1008            0 :             }
    1009              : 
    1010            0 :             BeMessage::KeepAlive(req) => {
    1011            0 :                 buf.put_u8(b'd');
    1012            0 :                 write_body(buf, |buf| {
    1013            0 :                     buf.put_u8(b'k');
    1014            0 :                     buf.put_u64(req.wal_end);
    1015            0 :                     buf.put_i64(req.timestamp);
    1016            0 :                     buf.put_u8(u8::from(req.request_reply));
    1017            0 :                 });
    1018            0 :             }
    1019              : 
    1020            0 :             BeMessage::NegotiateProtocolVersion { version, options } => {
    1021            0 :                 buf.put_u8(b'v');
    1022            0 :                 write_body(buf, |buf| {
    1023            0 :                     buf.put_u32(version.0);
    1024            0 :                     buf.put_u32(options.len() as u32);
    1025            0 :                     for option in options.iter() {
    1026            0 :                         write_cstr(option, buf)?;
    1027              :                     }
    1028            0 :                     Ok(())
    1029            0 :                 })?
    1030              :             }
    1031              : 
    1032            0 :             BeMessage::InterpretedWalRecords(rec) => {
    1033            0 :                 // We use the COPY_DATA_TAG for our custom message
    1034            0 :                 // since this tag is interpreted as raw bytes.
    1035            0 :                 buf.put_u8(b'd');
    1036            0 :                 write_body(buf, |buf| {
    1037            0 :                     buf.put_u8(b'0'); // matches INTERPRETED_WAL_RECORD_TAG in postgres-protocol
    1038            0 :                                       // dependency
    1039            0 :                     buf.put_u64(rec.streaming_lsn);
    1040            0 :                     buf.put_u64(rec.commit_lsn);
    1041            0 :                     buf.put_slice(rec.data);
    1042            0 :                 });
    1043            0 :             }
    1044              :         }
    1045           96 :         Ok(())
    1046           96 :     }
    1047              : }
    1048              : 
    1049            1 : fn terminate_code(code: &[u8; 5]) -> [u8; 6] {
    1050            1 :     let mut terminated = [0; 6];
    1051            5 :     for (i, &elem) in code.iter().enumerate() {
    1052            5 :         terminated[i] = elem;
    1053            5 :     }
    1054              : 
    1055            1 :     terminated
    1056            1 : }
    1057              : 
    1058              : #[cfg(test)]
    1059              : mod tests {
    1060              :     use super::*;
    1061              : 
    1062              :     #[test]
    1063            1 :     fn test_startup_message_params_options_escaped() {
    1064            4 :         fn split_options(params: &StartupMessageParams) -> Vec<Cow<'_, str>> {
    1065            4 :             params
    1066            4 :                 .options_escaped()
    1067            4 :                 .expect("options are None")
    1068            4 :                 .collect()
    1069            4 :         }
    1070              : 
    1071            4 :         let make_params = |options| StartupMessageParams::new([("options", options)]);
    1072              : 
    1073            1 :         let params = StartupMessageParams::new([]);
    1074            1 :         assert!(params.options_escaped().is_none());
    1075              : 
    1076            1 :         let params = make_params("");
    1077            1 :         assert!(split_options(&params).is_empty());
    1078              : 
    1079            1 :         let params = make_params("foo");
    1080            1 :         assert_eq!(split_options(&params), ["foo"]);
    1081              : 
    1082            1 :         let params = make_params(" foo  bar ");
    1083            1 :         assert_eq!(split_options(&params), ["foo", "bar"]);
    1084              : 
    1085            1 :         let params = make_params("foo\\ bar \\ \\\\ baz\\  lol");
    1086            1 :         assert_eq!(split_options(&params), ["foo bar", " \\", "baz ", "lol"]);
    1087            1 :     }
    1088              : 
    1089              :     #[test]
    1090            1 :     fn parse_fe_startup_packet_regression() {
    1091            1 :         let data = [0, 0, 0, 7, 0, 0, 0, 0];
    1092            1 :         FeStartupPacket::parse(&mut BytesMut::from_iter(data)).unwrap_err();
    1093            1 :     }
    1094              : 
    1095              :     #[test]
    1096            1 :     fn cancel_key_data() {
    1097            1 :         let key = CancelKeyData {
    1098            1 :             backend_pid: -1817212860,
    1099            1 :             cancel_key: -1183897012,
    1100            1 :         };
    1101            1 :         assert_eq!(format!("{key}"), "CancelKeyData(93af8844b96f2a4c)");
    1102            1 :     }
    1103              : }
        

Generated by: LCOV version 2.1-beta